From e1c34e912458abaae2cde27c85e0e2bb00b785a6 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Wed, 1 Nov 2023 22:46:02 +0800 Subject: [PATCH] mq：优化 redis stream 的命名 --- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQAutoConfiguration.java | 16 ++++++++-------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/RedisMQTemplate.java | 8 ++++---- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java | 4 ++-- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessage.java | 23 ----------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessageListener.java | 103 ------------------------------------------------------------------------------------------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractRedisChannelMessage.java | 23 +++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractRedisChannelMessageListener.java | 103 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractRedisStreamMessage.java | 23 +++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractRedisStreamMessageListener.java | 113 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessage.java | 23 ----------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessageListener.java | 113 ----------------------------------------------------------------------------------------------------------------- yudao-framework/yudao-spring-boot-starter-mq/《芋道 Spring Boot 事件机制 Event 入门》.md | 1 + 12 files changed, 277 insertions(+), 276 deletions(-) delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessage.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessageListener.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractRedisChannelMessage.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractRedisChannelMessageListener.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractRedisStreamMessage.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractRedisStreamMessageListener.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessage.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessageListener.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/《芋道 Spring Boot 事件机制 Event 入门》.md diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQAutoConfiguration.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQAutoConfiguration.java index 7ecd879..bbc63b7 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQAutoConfiguration.java +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQAutoConfiguration.java @@ -7,8 +7,8 @@ import cn.iocoder.yudao.framework.common.enums.DocumentEnum; import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; import cn.iocoder.yudao.framework.mq.redis.core.interceptor.RedisMessageInterceptor; import cn.iocoder.yudao.framework.mq.redis.core.job.RedisPendingMessageResendJob; -import cn.iocoder.yudao.framework.mq.redis.core.pubsub.AbstractChannelMessageListener; -import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractStreamMessageListener; +import cn.iocoder.yudao.framework.mq.redis.core.pubsub.AbstractRedisChannelMessageListener; +import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessageListener; import cn.iocoder.yudao.framework.redis.config.YudaoRedisAutoConfiguration; import lombok.extern.slf4j.Slf4j; import org.redisson.api.RedissonClient; @@ -57,9 +57,9 @@ public class YudaoRedisMQAutoConfiguration { * 创建 Redis Pub/Sub 广播消费的容器 */ @Bean(initMethod = "start", destroyMethod = "stop") - @ConditionalOnBean(AbstractChannelMessageListener.class) // 只有 AbstractChannelMessageListener 存在的时候，才需要注册 Redis pubsub 监听 + @ConditionalOnBean(AbstractRedisChannelMessageListener.class) // 只有 AbstractChannelMessageListener 存在的时候，才需要注册 Redis pubsub 监听 public RedisMessageListenerContainer redisMessageListenerContainer( - RedisMQTemplate redisMQTemplate, List> listeners) { + RedisMQTemplate redisMQTemplate, List> listeners) { // 创建 RedisMessageListenerContainer 对象 RedisMessageListenerContainer container = new RedisMessageListenerContainer(); // 设置 RedisConnection 工厂。 @@ -78,8 +78,8 @@ public class YudaoRedisMQAutoConfiguration { * 创建 Redis Stream 重新消费的任务 */ @Bean - @ConditionalOnBean(AbstractStreamMessageListener.class) // 只有 AbstractStreamMessageListener 存在的时候，才需要注册 Redis pubsub 监听 - public RedisPendingMessageResendJob redisPendingMessageResendJob(List> listeners, + @ConditionalOnBean(AbstractRedisStreamMessageListener.class) // 只有 AbstractStreamMessageListener 存在的时候，才需要注册 Redis pubsub 监听 + public RedisPendingMessageResendJob redisPendingMessageResendJob(List> listeners, RedisMQTemplate redisTemplate, @Value("${spring.application.name}") String groupName, RedissonClient redissonClient) { @@ -92,9 +92,9 @@ public class YudaoRedisMQAutoConfiguration { * 基础知识：Redis Stream 的 xreadgroup 命令 */ @Bean(initMethod = "start", destroyMethod = "stop") - @ConditionalOnBean(AbstractStreamMessageListener.class) // 只有 AbstractStreamMessageListener 存在的时候，才需要注册 Redis pubsub 监听 + @ConditionalOnBean(AbstractRedisStreamMessageListener.class) // 只有 AbstractStreamMessageListener 存在的时候，才需要注册 Redis pubsub 监听 public StreamMessageListenerContainer> redisStreamMessageListenerContainer( - RedisMQTemplate redisMQTemplate, List> listeners) { + RedisMQTemplate redisMQTemplate, List> listeners) { RedisTemplate redisTemplate = redisMQTemplate.getRedisTemplate(); checkRedisVersion(redisTemplate); // 第一步，创建 StreamMessageListenerContainer 容器 diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/RedisMQTemplate.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/RedisMQTemplate.java index 92a0f77..5755ffa 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/RedisMQTemplate.java +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/RedisMQTemplate.java @@ -3,8 +3,8 @@ package cn.iocoder.yudao.framework.mq.redis.core; import cn.iocoder.yudao.framework.common.util.json.JsonUtils; import cn.iocoder.yudao.framework.mq.redis.core.interceptor.RedisMessageInterceptor; import cn.iocoder.yudao.framework.mq.redis.core.message.AbstractRedisMessage; -import cn.iocoder.yudao.framework.mq.redis.core.pubsub.AbstractChannelMessage; -import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractStreamMessage; +import cn.iocoder.yudao.framework.mq.redis.core.pubsub.AbstractRedisChannelMessage; +import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessage; import lombok.AllArgsConstructor; import lombok.Getter; import org.springframework.data.redis.connection.stream.RecordId; @@ -35,7 +35,7 @@ public class RedisMQTemplate { * * @param message 消息 */ - public void send(T message) { + public void send(T message) { try { sendMessageBefore(message); // 发送消息 @@ -51,7 +51,7 @@ public class RedisMQTemplate { * @param message 消息 * @return 消息记录的编号对象 */ - public RecordId send(T message) { + public RecordId send(T message) { try { sendMessageBefore(message); // 发送消息 diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java index 02ede12..b84f17c 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java @@ -2,7 +2,7 @@ package cn.iocoder.yudao.framework.mq.redis.core.job; import cn.hutool.core.collection.CollUtil; import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; -import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractStreamMessageListener; +import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessageListener; import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.redisson.api.RLock; @@ -33,7 +33,7 @@ public class RedisPendingMessageResendJob { */ private static final int EXPIRE_TIME = 5 * 60; - private final List> listeners; + private final List> listeners; private final RedisMQTemplate redisTemplate; private final String groupName; private final RedissonClient redissonClient; diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessage.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessage.java deleted file mode 100644 index ce03efc..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessage.java +++ /dev/null @@ -1,23 +0,0 @@ -package cn.iocoder.yudao.framework.mq.redis.core.pubsub; - -import cn.iocoder.yudao.framework.mq.redis.core.message.AbstractRedisMessage; -import com.fasterxml.jackson.annotation.JsonIgnore; - -/** - * Redis Channel Message 抽象类 - * - * @author 芋道源码 - */ -public abstract class AbstractChannelMessage extends AbstractRedisMessage { - - /** - * 获得 Redis Channel，默认使用类名 - * - * @return Channel - */ - @JsonIgnore // 避免序列化。原因是，Redis 发布 Channel 消息的时候，已经会指定。 - public String getChannel() { - return getClass().getSimpleName(); - } - -} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessageListener.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessageListener.java deleted file mode 100644 index bae90a9..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessageListener.java +++ /dev/null @@ -1,103 +0,0 @@ -package cn.iocoder.yudao.framework.mq.redis.core.pubsub; - -import cn.hutool.core.util.TypeUtil; -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; -import cn.iocoder.yudao.framework.mq.redis.core.interceptor.RedisMessageInterceptor; -import cn.iocoder.yudao.framework.mq.redis.core.message.AbstractRedisMessage; -import lombok.Setter; -import lombok.SneakyThrows; -import org.springframework.data.redis.connection.Message; -import org.springframework.data.redis.connection.MessageListener; - -import java.lang.reflect.Type; -import java.util.List; - -/** - * Redis Pub/Sub 监听器抽象类，用于实现广播消费 - * - * @param 消息类型。一定要填写噢，不然会报错 - * - * @author 芋道源码 - */ -public abstract class AbstractChannelMessageListener implements MessageListener { - - /** - * 消息类型 - */ - private final Class messageType; - /** - * Redis Channel - */ - private final String channel; - /** - * RedisMQTemplate - */ - @Setter - private RedisMQTemplate redisMQTemplate; - - @SneakyThrows - protected AbstractChannelMessageListener() { - this.messageType = getMessageClass(); - this.channel = messageType.getDeclaredConstructor().newInstance().getChannel(); - } - - /** - * 获得 Sub 订阅的 Redis Channel 通道 - * - * @return channel - */ - public final String getChannel() { - return channel; - } - - @Override - public final void onMessage(Message message, byte[] bytes) { - T messageObj = JsonUtils.parseObject(message.getBody(), messageType); - try { - consumeMessageBefore(messageObj); - // 消费消息 - this.onMessage(messageObj); - } finally { - consumeMessageAfter(messageObj); - } - } - - /** - * 处理消息 - * - * @param message 消息 - */ - public abstract void onMessage(T message); - - /** - * 通过解析类上的泛型，获得消息类型 - * - * @return 消息类型 - */ - @SuppressWarnings("unchecked") - private Class getMessageClass() { - Type type = TypeUtil.getTypeArgument(getClass(), 0); - if (type == null) { - throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName())); - } - return (Class) type; - } - - private void consumeMessageBefore(AbstractRedisMessage message) { - assert redisMQTemplate != null; - List interceptors = redisMQTemplate.getInterceptors(); - // 正序 - interceptors.forEach(interceptor -> interceptor.consumeMessageBefore(message)); - } - - private void consumeMessageAfter(AbstractRedisMessage message) { - assert redisMQTemplate != null; - List interceptors = redisMQTemplate.getInterceptors(); - // 倒序 - for (int i = interceptors.size() - 1; i >= 0; i--) { - interceptors.get(i).consumeMessageAfter(message); - } - } - -} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractRedisChannelMessage.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractRedisChannelMessage.java new file mode 100644 index 0000000..d5ea5b9 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractRedisChannelMessage.java @@ -0,0 +1,23 @@ +package cn.iocoder.yudao.framework.mq.redis.core.pubsub; + +import cn.iocoder.yudao.framework.mq.redis.core.message.AbstractRedisMessage; +import com.fasterxml.jackson.annotation.JsonIgnore; + +/** + * Redis Channel Message 抽象类 + * + * @author 芋道源码 + */ +public abstract class AbstractRedisChannelMessage extends AbstractRedisMessage { + + /** + * 获得 Redis Channel，默认使用类名 + * + * @return Channel + */ + @JsonIgnore // 避免序列化。原因是，Redis 发布 Channel 消息的时候，已经会指定。 + public String getChannel() { + return getClass().getSimpleName(); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractRedisChannelMessageListener.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractRedisChannelMessageListener.java new file mode 100644 index 0000000..fd7c910 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractRedisChannelMessageListener.java @@ -0,0 +1,103 @@ +package cn.iocoder.yudao.framework.mq.redis.core.pubsub; + +import cn.hutool.core.util.TypeUtil; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; +import cn.iocoder.yudao.framework.mq.redis.core.interceptor.RedisMessageInterceptor; +import cn.iocoder.yudao.framework.mq.redis.core.message.AbstractRedisMessage; +import lombok.Setter; +import lombok.SneakyThrows; +import org.springframework.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; + +import java.lang.reflect.Type; +import java.util.List; + +/** + * Redis Pub/Sub 监听器抽象类，用于实现广播消费 + * + * @param 消息类型。一定要填写噢，不然会报错 + * + * @author 芋道源码 + */ +public abstract class AbstractRedisChannelMessageListener implements MessageListener { + + /** + * 消息类型 + */ + private final Class messageType; + /** + * Redis Channel + */ + private final String channel; + /** + * RedisMQTemplate + */ + @Setter + private RedisMQTemplate redisMQTemplate; + + @SneakyThrows + protected AbstractRedisChannelMessageListener() { + this.messageType = getMessageClass(); + this.channel = messageType.getDeclaredConstructor().newInstance().getChannel(); + } + + /** + * 获得 Sub 订阅的 Redis Channel 通道 + * + * @return channel + */ + public final String getChannel() { + return channel; + } + + @Override + public final void onMessage(Message message, byte[] bytes) { + T messageObj = JsonUtils.parseObject(message.getBody(), messageType); + try { + consumeMessageBefore(messageObj); + // 消费消息 + this.onMessage(messageObj); + } finally { + consumeMessageAfter(messageObj); + } + } + + /** + * 处理消息 + * + * @param message 消息 + */ + public abstract void onMessage(T message); + + /** + * 通过解析类上的泛型，获得消息类型 + * + * @return 消息类型 + */ + @SuppressWarnings("unchecked") + private Class getMessageClass() { + Type type = TypeUtil.getTypeArgument(getClass(), 0); + if (type == null) { + throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName())); + } + return (Class) type; + } + + private void consumeMessageBefore(AbstractRedisMessage message) { + assert redisMQTemplate != null; + List interceptors = redisMQTemplate.getInterceptors(); + // 正序 + interceptors.forEach(interceptor -> interceptor.consumeMessageBefore(message)); + } + + private void consumeMessageAfter(AbstractRedisMessage message) { + assert redisMQTemplate != null; + List interceptors = redisMQTemplate.getInterceptors(); + // 倒序 + for (int i = interceptors.size() - 1; i >= 0; i--) { + interceptors.get(i).consumeMessageAfter(message); + } + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractRedisStreamMessage.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractRedisStreamMessage.java new file mode 100644 index 0000000..9017e08 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractRedisStreamMessage.java @@ -0,0 +1,23 @@ +package cn.iocoder.yudao.framework.mq.redis.core.stream; + +import cn.iocoder.yudao.framework.mq.redis.core.message.AbstractRedisMessage; +import com.fasterxml.jackson.annotation.JsonIgnore; + +/** + * Redis Stream Message 抽象类 + * + * @author 芋道源码 + */ +public abstract class AbstractRedisStreamMessage extends AbstractRedisMessage { + + /** + * 获得 Redis Stream Key，默认使用类名 + * + * @return Channel + */ + @JsonIgnore // 避免序列化 + public String getStreamKey() { + return getClass().getSimpleName(); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractRedisStreamMessageListener.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractRedisStreamMessageListener.java new file mode 100644 index 0000000..3e656af --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractRedisStreamMessageListener.java @@ -0,0 +1,113 @@ +package cn.iocoder.yudao.framework.mq.redis.core.stream; + +import cn.hutool.core.util.TypeUtil; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; +import cn.iocoder.yudao.framework.mq.redis.core.interceptor.RedisMessageInterceptor; +import cn.iocoder.yudao.framework.mq.redis.core.message.AbstractRedisMessage; +import lombok.Getter; +import lombok.Setter; +import lombok.SneakyThrows; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.data.redis.connection.stream.ObjectRecord; +import org.springframework.data.redis.stream.StreamListener; + +import java.lang.reflect.Type; +import java.util.List; + +/** + * Redis Stream 监听器抽象类，用于实现集群消费 + * + * @param 消息类型。一定要填写噢，不然会报错 + * + * @author 芋道源码 + */ +public abstract class AbstractRedisStreamMessageListener + implements StreamListener> { + + /** + * 消息类型 + */ + private final Class messageType; + /** + * Redis Channel + */ + @Getter + private final String streamKey; + + /** + * Redis 消费者分组，默认使用 spring.application.name 名字 + */ + @Value("${spring.application.name}") + @Getter + private String group; + /** + * RedisMQTemplate + */ + @Setter + private RedisMQTemplate redisMQTemplate; + + @SneakyThrows + protected AbstractRedisStreamMessageListener() { + this.messageType = getMessageClass(); + this.streamKey = messageType.getDeclaredConstructor().newInstance().getStreamKey(); + } + + @Override + public void onMessage(ObjectRecord message) { + // 消费消息 + T messageObj = JsonUtils.parseObject(message.getValue(), messageType); + try { + consumeMessageBefore(messageObj); + // 消费消息 + this.onMessage(messageObj); + // ack 消息消费完成 + redisMQTemplate.getRedisTemplate().opsForStream().acknowledge(group, message); + // TODO 芋艿：需要额外考虑以下几个点： + // 1. 处理异常的情况 + // 2. 发送日志；以及事务的结合 + // 3. 消费日志；以及通用的幂等性 + // 4. 消费失败的重试，https://zhuanlan.zhihu.com/p/60501638 + } finally { + consumeMessageAfter(messageObj); + } + } + + /** + * 处理消息 + * + * @param message 消息 + */ + public abstract void onMessage(T message); + + /** + * 通过解析类上的泛型，获得消息类型 + * + * @return 消息类型 + */ + @SuppressWarnings("unchecked") + private Class getMessageClass() { + Type type = TypeUtil.getTypeArgument(getClass(), 0); + if (type == null) { + throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName())); + } + return (Class) type; + } + + private void consumeMessageBefore(AbstractRedisMessage message) { + assert redisMQTemplate != null; + List interceptors = redisMQTemplate.getInterceptors(); + // 正序 + interceptors.forEach(interceptor -> interceptor.consumeMessageBefore(message)); + } + + private void consumeMessageAfter(AbstractRedisMessage message) { + assert redisMQTemplate != null; + List interceptors = redisMQTemplate.getInterceptors(); + // 倒序 + for (int i = interceptors.size() - 1; i >= 0; i--) { + interceptors.get(i).consumeMessageAfter(message); + } + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessage.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessage.java deleted file mode 100644 index 72c6aac..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessage.java +++ /dev/null @@ -1,23 +0,0 @@ -package cn.iocoder.yudao.framework.mq.redis.core.stream; - -import cn.iocoder.yudao.framework.mq.redis.core.message.AbstractRedisMessage; -import com.fasterxml.jackson.annotation.JsonIgnore; - -/** - * Redis Stream Message 抽象类 - * - * @author 芋道源码 - */ -public abstract class AbstractStreamMessage extends AbstractRedisMessage { - - /** - * 获得 Redis Stream Key，默认使用类名 - * - * @return Channel - */ - @JsonIgnore // 避免序列化 - public String getStreamKey() { - return getClass().getSimpleName(); - } - -} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessageListener.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessageListener.java deleted file mode 100644 index 4a4ea00..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessageListener.java +++ /dev/null @@ -1,113 +0,0 @@ -package cn.iocoder.yudao.framework.mq.redis.core.stream; - -import cn.hutool.core.util.TypeUtil; -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; -import cn.iocoder.yudao.framework.mq.redis.core.interceptor.RedisMessageInterceptor; -import cn.iocoder.yudao.framework.mq.redis.core.message.AbstractRedisMessage; -import lombok.Getter; -import lombok.Setter; -import lombok.SneakyThrows; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.data.redis.connection.stream.ObjectRecord; -import org.springframework.data.redis.stream.StreamListener; - -import java.lang.reflect.Type; -import java.util.List; - -/** - * Redis Stream 监听器抽象类，用于实现集群消费 - * - * @param 消息类型。一定要填写噢，不然会报错 - * - * @author 芋道源码 - */ -public abstract class AbstractStreamMessageListener - implements StreamListener> { - - /** - * 消息类型 - */ - private final Class messageType; - /** - * Redis Channel - */ - @Getter - private final String streamKey; - - /** - * Redis 消费者分组，默认使用 spring.application.name 名字 - */ - @Value("${spring.application.name}") - @Getter - private String group; - /** - * RedisMQTemplate - */ - @Setter - private RedisMQTemplate redisMQTemplate; - - @SneakyThrows - protected AbstractStreamMessageListener() { - this.messageType = getMessageClass(); - this.streamKey = messageType.getDeclaredConstructor().newInstance().getStreamKey(); - } - - @Override - public void onMessage(ObjectRecord message) { - // 消费消息 - T messageObj = JsonUtils.parseObject(message.getValue(), messageType); - try { - consumeMessageBefore(messageObj); - // 消费消息 - this.onMessage(messageObj); - // ack 消息消费完成 - redisMQTemplate.getRedisTemplate().opsForStream().acknowledge(group, message); - // TODO 芋艿：需要额外考虑以下几个点： - // 1. 处理异常的情况 - // 2. 发送日志；以及事务的结合 - // 3. 消费日志；以及通用的幂等性 - // 4. 消费失败的重试，https://zhuanlan.zhihu.com/p/60501638 - } finally { - consumeMessageAfter(messageObj); - } - } - - /** - * 处理消息 - * - * @param message 消息 - */ - public abstract void onMessage(T message); - - /** - * 通过解析类上的泛型，获得消息类型 - * - * @return 消息类型 - */ - @SuppressWarnings("unchecked") - private Class getMessageClass() { - Type type = TypeUtil.getTypeArgument(getClass(), 0); - if (type == null) { - throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName())); - } - return (Class) type; - } - - private void consumeMessageBefore(AbstractRedisMessage message) { - assert redisMQTemplate != null; - List interceptors = redisMQTemplate.getInterceptors(); - // 正序 - interceptors.forEach(interceptor -> interceptor.consumeMessageBefore(message)); - } - - private void consumeMessageAfter(AbstractRedisMessage message) { - assert redisMQTemplate != null; - List interceptors = redisMQTemplate.getInterceptors(); - // 倒序 - for (int i = interceptors.size() - 1; i >= 0; i--) { - interceptors.get(i).consumeMessageAfter(message); - } - } - -} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/《芋道 Spring Boot 事件机制 Event 入门》.md b/yudao-framework/yudao-spring-boot-starter-mq/《芋道 Spring Boot 事件机制 Event 入门》.md new file mode 100644 index 0000000..a3661ee --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/《芋道 Spring Boot 事件机制 Event 入门》.md @@ -0,0 +1 @@ + -- libgit2 0.21.4