From c066ea46f9881e8ccf7b7abf3b2add8a38f9ea84 Mon Sep 17 00:00:00 2001 From: YunaiV Date: Mon, 30 Oct 2023 21:15:22 +0800 Subject: [PATCH] mq 重构：默认的 redis 实现： 1）默认 channel 和 stream key 2）移除 enabled 开关，通过 listener 是否存在来实现 3）调整包名，为接入 rocketmq 作为 mq 实现做准备 --- yudao-framework/yudao-spring-boot-starter-biz-tenant/src/main/java/cn/iocoder/yudao/framework/tenant/core/mq/TenantRedisMessageInterceptor.java | 4 ++-- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java | 170 -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/RedisMQTemplate.java | 87 --------------------------------------------------------------------------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/interceptor/RedisMessageInterceptor.java | 26 -------------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/message/AbstractRedisMessage.java | 29 ----------------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessage.java | 21 --------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessageListener.java | 103 ------------------------------------------------------------------------------------------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessage.java | 21 --------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessageListener.java | 113 ----------------------------------------------------------------------------------------------------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java | 100 ---------------------------------------------------------------------------------------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/package-info.java | 1 + yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQAutoConfiguration.java | 164 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/RedisMQTemplate.java | 87 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/interceptor/RedisMessageInterceptor.java | 26 ++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java | 100 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/message/AbstractRedisMessage.java | 29 +++++++++++++++++++++++++++++ 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/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/src/main/java/cn/iocoder/yudao/framework/mq/redis/package-info.java | 6 ++++++ yudao-framework/yudao-spring-boot-starter-mq/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainerX.java | 62 -------------------------------------------------------------- yudao-framework/yudao-spring-boot-starter-mq/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports | 2 +- yudao-server/src/main/resources/application.yaml | 6 ------ 24 files changed, 678 insertions(+), 741 deletions(-) delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/RedisMQTemplate.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/interceptor/RedisMessageInterceptor.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/message/AbstractRedisMessage.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessage.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessageListener.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessage.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessageListener.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQAutoConfiguration.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/RedisMQTemplate.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/interceptor/RedisMessageInterceptor.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/message/AbstractRedisMessage.java create mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessage.java create 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/stream/AbstractStreamMessage.java create 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/src/main/java/cn/iocoder/yudao/framework/mq/redis/package-info.java delete mode 100644 yudao-framework/yudao-spring-boot-starter-mq/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainerX.java diff --git a/yudao-framework/yudao-spring-boot-starter-biz-tenant/src/main/java/cn/iocoder/yudao/framework/tenant/core/mq/TenantRedisMessageInterceptor.java b/yudao-framework/yudao-spring-boot-starter-biz-tenant/src/main/java/cn/iocoder/yudao/framework/tenant/core/mq/TenantRedisMessageInterceptor.java index c493b41..e7abb67 100644 --- a/yudao-framework/yudao-spring-boot-starter-biz-tenant/src/main/java/cn/iocoder/yudao/framework/tenant/core/mq/TenantRedisMessageInterceptor.java +++ b/yudao-framework/yudao-spring-boot-starter-biz-tenant/src/main/java/cn/iocoder/yudao/framework/tenant/core/mq/TenantRedisMessageInterceptor.java @@ -1,8 +1,8 @@ package cn.iocoder.yudao.framework.tenant.core.mq; import cn.hutool.core.util.StrUtil; -import cn.iocoder.yudao.framework.mq.core.interceptor.RedisMessageInterceptor; -import cn.iocoder.yudao.framework.mq.core.message.AbstractRedisMessage; +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.tenant.core.context.TenantContextHolder; import static cn.iocoder.yudao.framework.web.core.util.WebFrameworkUtils.HEADER_TENANT_ID; diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java deleted file mode 100644 index e300b1a..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/config/YudaoMQAutoConfiguration.java +++ /dev/null @@ -1,170 +0,0 @@ -package cn.iocoder.yudao.framework.mq.config; - -import cn.hutool.core.map.MapUtil; -import cn.hutool.core.util.StrUtil; -import cn.hutool.system.SystemUtil; -import cn.iocoder.yudao.framework.common.enums.DocumentEnum; -import cn.iocoder.yudao.framework.mq.core.RedisMQTemplate; -import cn.iocoder.yudao.framework.mq.core.interceptor.RedisMessageInterceptor; -import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessageListener; -import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessageListener; -import cn.iocoder.yudao.framework.mq.job.RedisPendingMessageResendJob; -import cn.iocoder.yudao.framework.redis.config.YudaoRedisAutoConfiguration; -import lombok.extern.slf4j.Slf4j; -import org.redisson.api.RedissonClient; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.AutoConfiguration; -import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.context.annotation.Bean; -import org.springframework.data.redis.connection.RedisServerCommands; -import org.springframework.data.redis.connection.stream.Consumer; -import org.springframework.data.redis.connection.stream.ObjectRecord; -import org.springframework.data.redis.connection.stream.ReadOffset; -import org.springframework.data.redis.connection.stream.StreamOffset; -import org.springframework.data.redis.core.RedisCallback; -import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.data.redis.core.StringRedisTemplate; -import org.springframework.data.redis.listener.ChannelTopic; -import org.springframework.data.redis.listener.RedisMessageListenerContainer; -import org.springframework.data.redis.stream.DefaultStreamMessageListenerContainerX; -import org.springframework.data.redis.stream.StreamMessageListenerContainer; -import org.springframework.scheduling.annotation.EnableScheduling; - -import java.util.List; -import java.util.Properties; - -/** - * 消息队列配置类 - * - * @author 芋道源码 - */ -@Slf4j -@EnableScheduling // 启用定时任务，用于 RedisPendingMessageResendJob 重发消息 -@AutoConfiguration(after = YudaoRedisAutoConfiguration.class) -public class YudaoMQAutoConfiguration { - - @Bean - public RedisMQTemplate redisMQTemplate(StringRedisTemplate redisTemplate, - List interceptors) { - RedisMQTemplate redisMQTemplate = new RedisMQTemplate(redisTemplate); - // 添加拦截器 - interceptors.forEach(redisMQTemplate::addInterceptor); - return redisMQTemplate; - } - - // ========== 消费者相关 ========== - - /** - * 创建 Redis Pub/Sub 广播消费的容器 - */ - @Bean(initMethod = "start", destroyMethod = "stop") - @ConditionalOnBean(AbstractChannelMessageListener.class) // 只有 AbstractChannelMessageListener 存在的时候，才需要注册 Redis pubsub 监听 - @ConditionalOnProperty(prefix = "yudao.mq.redis.pubsub", value = "enable", matchIfMissing = true) // 允许使用 yudao.mq.redis.pubsub.enable=false 禁用多租户 - public RedisMessageListenerContainer redisMessageListenerContainer( - RedisMQTemplate redisMQTemplate, List> listeners) { - // 创建 RedisMessageListenerContainer 对象 - RedisMessageListenerContainer container = new RedisMessageListenerContainer(); - // 设置 RedisConnection 工厂。 - container.setConnectionFactory(redisMQTemplate.getRedisTemplate().getRequiredConnectionFactory()); - // 添加监听器 - listeners.forEach(listener -> { - listener.setRedisMQTemplate(redisMQTemplate); - container.addMessageListener(listener, new ChannelTopic(listener.getChannel())); - log.info("[redisMessageListenerContainer][注册 Channel({}) 对应的监听器({})]", - listener.getChannel(), listener.getClass().getName()); - }); - return container; - } - - /** - * 创建 Redis Stream 重新消费的任务 - */ - @Bean - @ConditionalOnBean(AbstractStreamMessageListener.class) // 只有 AbstractStreamMessageListener 存在的时候，才需要注册 Redis pubsub 监听 - @ConditionalOnProperty(prefix = "yudao.mq.redis.stream", value = "enable", matchIfMissing = true) // 允许使用 yudao.mq.redis.stream.enable=false 禁用多租户 - public RedisPendingMessageResendJob redisPendingMessageResendJob(List> listeners, - RedisMQTemplate redisTemplate, - @Value("${spring.application.name}") String groupName, - RedissonClient redissonClient) { - return new RedisPendingMessageResendJob(listeners, redisTemplate, groupName, redissonClient); - } - - /** - * 创建 Redis Stream 集群消费的容器 - *

- * Redis Stream 的 xreadgroup 命令：https://www.geek-book.com/src/docs/redis/redis/redis.io/commands/xreadgroup.html - */ - @Bean(initMethod = "start", destroyMethod = "stop") - @ConditionalOnBean(AbstractStreamMessageListener.class) // 只有 AbstractStreamMessageListener 存在的时候，才需要注册 Redis pubsub 监听 - @ConditionalOnProperty(prefix = "yudao.mq.redis.stream", value = "enable", matchIfMissing = true) // 允许使用 yudao.mq.redis.stream.enable=false 禁用多租户 - public StreamMessageListenerContainer> redisStreamMessageListenerContainer( - RedisMQTemplate redisMQTemplate, List> listeners) { - RedisTemplate redisTemplate = redisMQTemplate.getRedisTemplate(); - checkRedisVersion(redisTemplate); - // 第一步，创建 StreamMessageListenerContainer 容器 - // 创建 options 配置 - StreamMessageListenerContainer.StreamMessageListenerContainerOptions> containerOptions = - StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder() - .batchSize(10) // 一次性最多拉取多少条消息 - .targetType(String.class) // 目标类型。统一使用 String，通过自己封装的 AbstractStreamMessageListener 去反序列化 - .build(); - // 创建 container 对象 - StreamMessageListenerContainer> container = -// StreamMessageListenerContainer.create(redisTemplate.getRequiredConnectionFactory(), containerOptions); - DefaultStreamMessageListenerContainerX.create(redisMQTemplate.getRedisTemplate().getRequiredConnectionFactory(), containerOptions); - - // 第二步，注册监听器，消费对应的 Stream 主题 - String consumerName = buildConsumerName(); - listeners.parallelStream().forEach(listener -> { - log.info("[redisStreamMessageListenerContainer][开始注册 StreamKey({}) 对应的监听器({})]", - listener.getStreamKey(), listener.getClass().getName()); - // 创建 listener 对应的消费者分组 - try { - redisTemplate.opsForStream().createGroup(listener.getStreamKey(), listener.getGroup()); - } catch (Exception ignore) { - } - // 设置 listener 对应的 redisTemplate - listener.setRedisMQTemplate(redisMQTemplate); - // 创建 Consumer 对象 - Consumer consumer = Consumer.from(listener.getGroup(), consumerName); - // 设置 Consumer 消费进度，以最小消费进度为准 - StreamOffset streamOffset = StreamOffset.create(listener.getStreamKey(), ReadOffset.lastConsumed()); - // 设置 Consumer 监听 - StreamMessageListenerContainer.StreamReadRequestBuilder builder = StreamMessageListenerContainer.StreamReadRequest - .builder(streamOffset).consumer(consumer) - .autoAcknowledge(false) // 不自动 ack - .cancelOnError(throwable -> false); // 默认配置，发生异常就取消消费，显然不符合预期；因此，我们设置为 false - container.register(builder.build(), listener); - log.info("[redisStreamMessageListenerContainer][完成注册 StreamKey({}) 对应的监听器({})]", - listener.getStreamKey(), listener.getClass().getName()); - }); - return container; - } - - /** - * 构建消费者名字，使用本地 IP + 进程编号的方式。 - * 参考自 RocketMQ clientId 的实现 - * - * @return 消费者名字 - */ - private static String buildConsumerName() { - return String.format("%s@%d", SystemUtil.getHostInfo().getAddress(), SystemUtil.getCurrentPID()); - } - - /** - * 校验 Redis 版本号，是否满足最低的版本号要求！ - */ - private static void checkRedisVersion(RedisTemplate redisTemplate) { - // 获得 Redis 版本 - Properties info = redisTemplate.execute((RedisCallback) RedisServerCommands::info); - String version = MapUtil.getStr(info, "redis_version"); - // 校验最低版本必须大于等于 5.0.0 - int majorVersion = Integer.parseInt(StrUtil.subBefore(version, '.', false)); - if (majorVersion < 5) { - throw new IllegalStateException(StrUtil.format("您当前的 Redis 版本为 {}，小于最低要求的 5.0.0 版本！" + - "请参考 {} 文档进行安装。", version, DocumentEnum.REDIS_INSTALL.getUrl())); - } - } - -} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/RedisMQTemplate.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/RedisMQTemplate.java deleted file mode 100644 index 8a31fed..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/RedisMQTemplate.java +++ /dev/null @@ -1,87 +0,0 @@ -package cn.iocoder.yudao.framework.mq.core; - -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import cn.iocoder.yudao.framework.mq.core.interceptor.RedisMessageInterceptor; -import cn.iocoder.yudao.framework.mq.core.message.AbstractRedisMessage; -import cn.iocoder.yudao.framework.mq.core.pubsub.AbstractChannelMessage; -import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessage; -import lombok.AllArgsConstructor; -import lombok.Getter; -import org.springframework.data.redis.connection.stream.RecordId; -import org.springframework.data.redis.connection.stream.StreamRecords; -import org.springframework.data.redis.core.RedisTemplate; - -import java.util.ArrayList; -import java.util.List; - -/** - * Redis MQ 操作模板类 - * - * @author 芋道源码 - */ -@AllArgsConstructor -public class RedisMQTemplate { - - @Getter - private final RedisTemplate redisTemplate; - /** - * 拦截器数组 - */ - @Getter - private final List interceptors = new ArrayList<>(); - - /** - * 发送 Redis 消息，基于 Redis pub/sub 实现 - * - * @param message 消息 - */ - public void send(T message) { - try { - sendMessageBefore(message); - // 发送消息 - redisTemplate.convertAndSend(message.getChannel(), JsonUtils.toJsonString(message)); - } finally { - sendMessageAfter(message); - } - } - - /** - * 发送 Redis 消息，基于 Redis Stream 实现 - * - * @param message 消息 - * @return 消息记录的编号对象 - */ - public RecordId send(T message) { - try { - sendMessageBefore(message); - // 发送消息 - return redisTemplate.opsForStream().add(StreamRecords.newRecord() - .ofObject(JsonUtils.toJsonString(message)) // 设置内容 - .withStreamKey(message.getStreamKey())); // 设置 stream key - } finally { - sendMessageAfter(message); - } - } - - /** - * 添加拦截器 - * - * @param interceptor 拦截器 - */ - public void addInterceptor(RedisMessageInterceptor interceptor) { - interceptors.add(interceptor); - } - - private void sendMessageBefore(AbstractRedisMessage message) { - // 正序 - interceptors.forEach(interceptor -> interceptor.sendMessageBefore(message)); - } - - private void sendMessageAfter(AbstractRedisMessage message) { - // 倒序 - for (int i = interceptors.size() - 1; i >= 0; i--) { - interceptors.get(i).sendMessageAfter(message); - } - } - -} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/interceptor/RedisMessageInterceptor.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/interceptor/RedisMessageInterceptor.java deleted file mode 100644 index 11d8e13..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/interceptor/RedisMessageInterceptor.java +++ /dev/null @@ -1,26 +0,0 @@ -package cn.iocoder.yudao.framework.mq.core.interceptor; - -import cn.iocoder.yudao.framework.mq.core.message.AbstractRedisMessage; - -/** - * {@link AbstractRedisMessage} 消息拦截器 - * 通过拦截器，作为插件机制，实现拓展。 - * 例如说，多租户场景下的 MQ 消息处理 - * - * @author 芋道源码 - */ -public interface RedisMessageInterceptor { - - default void sendMessageBefore(AbstractRedisMessage message) { - } - - default void sendMessageAfter(AbstractRedisMessage message) { - } - - default void consumeMessageBefore(AbstractRedisMessage message) { - } - - default void consumeMessageAfter(AbstractRedisMessage message) { - } - -} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/message/AbstractRedisMessage.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/message/AbstractRedisMessage.java deleted file mode 100644 index f02e89d..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/message/AbstractRedisMessage.java +++ /dev/null @@ -1,29 +0,0 @@ -package cn.iocoder.yudao.framework.mq.core.message; - -import lombok.Data; - -import java.util.HashMap; -import java.util.Map; - -/** - * Redis 消息抽象基类 - * - * @author 芋道源码 - */ -@Data -public abstract class AbstractRedisMessage { - - /** - * 头 - */ - private Map headers = new HashMap<>(); - - public String getHeader(String key) { - return headers.get(key); - } - - public void addHeader(String key, String value) { - headers.put(key, value); - } - -} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessage.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessage.java deleted file mode 100644 index fbc2a28..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessage.java +++ /dev/null @@ -1,21 +0,0 @@ -package cn.iocoder.yudao.framework.mq.core.pubsub; - -import cn.iocoder.yudao.framework.mq.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 abstract String getChannel(); - -} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessageListener.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessageListener.java deleted file mode 100644 index e7d737d..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/pubsub/AbstractChannelMessageListener.java +++ /dev/null @@ -1,103 +0,0 @@ -package cn.iocoder.yudao.framework.mq.core.pubsub; - -import cn.hutool.core.util.TypeUtil; -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import cn.iocoder.yudao.framework.mq.core.RedisMQTemplate; -import cn.iocoder.yudao.framework.mq.core.interceptor.RedisMessageInterceptor; -import cn.iocoder.yudao.framework.mq.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/core/stream/AbstractStreamMessage.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessage.java deleted file mode 100644 index 29ea833..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessage.java +++ /dev/null @@ -1,21 +0,0 @@ -package cn.iocoder.yudao.framework.mq.core.stream; - -import cn.iocoder.yudao.framework.mq.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 abstract String getStreamKey(); - -} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessageListener.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessageListener.java deleted file mode 100644 index f216885..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/core/stream/AbstractStreamMessageListener.java +++ /dev/null @@ -1,113 +0,0 @@ -package cn.iocoder.yudao.framework.mq.core.stream; - -import cn.hutool.core.util.TypeUtil; -import cn.iocoder.yudao.framework.common.util.json.JsonUtils; -import cn.iocoder.yudao.framework.mq.core.RedisMQTemplate; -import cn.iocoder.yudao.framework.mq.core.interceptor.RedisMessageInterceptor; -import cn.iocoder.yudao.framework.mq.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/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java deleted file mode 100644 index ea0f53d..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/job/RedisPendingMessageResendJob.java +++ /dev/null @@ -1,100 +0,0 @@ -package cn.iocoder.yudao.framework.mq.job; - -import cn.hutool.core.collection.CollUtil; -import cn.iocoder.yudao.framework.mq.core.RedisMQTemplate; -import cn.iocoder.yudao.framework.mq.core.stream.AbstractStreamMessageListener; -import lombok.AllArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.redisson.api.RLock; -import org.redisson.api.RedissonClient; -import org.springframework.data.domain.Range; -import org.springframework.data.redis.connection.stream.*; -import org.springframework.data.redis.core.StreamOperations; -import org.springframework.scheduling.annotation.Scheduled; - -import java.util.List; -import java.util.Map; -import java.util.Objects; - -/** - * 这个任务用于处理，crash 之后的消费者未消费完的消息 - */ -@Slf4j -@AllArgsConstructor -public class RedisPendingMessageResendJob { - - private static final String LOCK_KEY = "redis:pending:msg:lock"; - - /** - * 消息超时时间，默认 5 分钟 - * - * 1. 超时的消息才会被重新投递 - * 2. 由于定时任务 1 分钟一次，消息超时后不会被立即重投，极端情况下消息5分钟过期后，再等 1 分钟才会被扫瞄到 - */ - private static final int EXPIRE_TIME = 5 * 60; - - private final List> listeners; - private final RedisMQTemplate redisTemplate; - private final String groupName; - private final RedissonClient redissonClient; - - /** - * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题 - */ - @Scheduled(cron = "35 * * * * ?") - public void messageResend() { - RLock lock = redissonClient.getLock(LOCK_KEY); - // 尝试加锁 - if (lock.tryLock()) { - try { - execute(); - } catch (Exception ex) { - log.error("[messageResend][执行异常]", ex); - } finally { - lock.unlock(); - } - } - } - - /** - * 执行清理逻辑 - * - * @see 讨论 - */ - private void execute() { - StreamOperations ops = redisTemplate.getRedisTemplate().opsForStream(); - listeners.forEach(listener -> { - PendingMessagesSummary pendingMessagesSummary = Objects.requireNonNull(ops.pending(listener.getStreamKey(), groupName)); - // 每个消费者的 pending 队列消息数量 - Map pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer(); - pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> { - log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount); - // 每个消费者的 pending消息的详情信息 - PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(groupName, consumerName), Range.unbounded(), pendingMessageCount); - if (pendingMessages.isEmpty()) { - return; - } - pendingMessages.forEach(pendingMessage -> { - // 获取消息上一次传递到 consumer 的时间, - long lastDelivery = pendingMessage.getElapsedTimeSinceLastDelivery().getSeconds(); - if (lastDelivery < EXPIRE_TIME){ - return; - } - // 获取指定 id 的消息体 - List> records = ops.range(listener.getStreamKey(), - Range.of(Range.Bound.inclusive(pendingMessage.getIdAsString()), Range.Bound.inclusive(pendingMessage.getIdAsString()))); - if (CollUtil.isEmpty(records)) { - return; - } - // 重新投递消息 - redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord() - .ofObject(records.get(0).getValue()) // 设置内容 - .withStreamKey(listener.getStreamKey())); - // ack 消息消费完成 - redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, records.get(0)); - log.info("[processPendingMessage][消息({})重新投递成功]", records.get(0).getId()); - }); - }); - }); - } -} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/package-info.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/package-info.java index 48eaf23..62e7ce4 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/package-info.java +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/package-info.java @@ -1,4 +1,5 @@ /** + * TODO 芋艿：调整注释 * 消息队列，基于 Redis 提供： * 1. 基于 Pub/Sub 实现广播消费 * 2. 基于 Stream 实现集群消费 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 new file mode 100644 index 0000000..7ecd879 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQAutoConfiguration.java @@ -0,0 +1,164 @@ +package cn.iocoder.yudao.framework.mq.redis.config; + +import cn.hutool.core.map.MapUtil; +import cn.hutool.core.util.StrUtil; +import cn.hutool.system.SystemUtil; +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.redis.config.YudaoRedisAutoConfiguration; +import lombok.extern.slf4j.Slf4j; +import org.redisson.api.RedissonClient; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.context.annotation.Bean; +import org.springframework.data.redis.connection.RedisServerCommands; +import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.ObjectRecord; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.core.RedisCallback; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.listener.ChannelTopic; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; +import org.springframework.data.redis.stream.StreamMessageListenerContainer; +import org.springframework.scheduling.annotation.EnableScheduling; + +import java.util.List; +import java.util.Properties; + +/** + * 消息队列配置类 + * + * @author 芋道源码 + */ +@Slf4j +@EnableScheduling // 启用定时任务，用于 RedisPendingMessageResendJob 重发消息 +@AutoConfiguration(after = YudaoRedisAutoConfiguration.class) +public class YudaoRedisMQAutoConfiguration { + + @Bean + public RedisMQTemplate redisMQTemplate(StringRedisTemplate redisTemplate, + List interceptors) { + RedisMQTemplate redisMQTemplate = new RedisMQTemplate(redisTemplate); + // 添加拦截器 + interceptors.forEach(redisMQTemplate::addInterceptor); + return redisMQTemplate; + } + + // ========== 消费者相关 ========== + + /** + * 创建 Redis Pub/Sub 广播消费的容器 + */ + @Bean(initMethod = "start", destroyMethod = "stop") + @ConditionalOnBean(AbstractChannelMessageListener.class) // 只有 AbstractChannelMessageListener 存在的时候，才需要注册 Redis pubsub 监听 + public RedisMessageListenerContainer redisMessageListenerContainer( + RedisMQTemplate redisMQTemplate, List> listeners) { + // 创建 RedisMessageListenerContainer 对象 + RedisMessageListenerContainer container = new RedisMessageListenerContainer(); + // 设置 RedisConnection 工厂。 + container.setConnectionFactory(redisMQTemplate.getRedisTemplate().getRequiredConnectionFactory()); + // 添加监听器 + listeners.forEach(listener -> { + listener.setRedisMQTemplate(redisMQTemplate); + container.addMessageListener(listener, new ChannelTopic(listener.getChannel())); + log.info("[redisMessageListenerContainer][注册 Channel({}) 对应的监听器({})]", + listener.getChannel(), listener.getClass().getName()); + }); + return container; + } + + /** + * 创建 Redis Stream 重新消费的任务 + */ + @Bean + @ConditionalOnBean(AbstractStreamMessageListener.class) // 只有 AbstractStreamMessageListener 存在的时候，才需要注册 Redis pubsub 监听 + public RedisPendingMessageResendJob redisPendingMessageResendJob(List> listeners, + RedisMQTemplate redisTemplate, + @Value("${spring.application.name}") String groupName, + RedissonClient redissonClient) { + return new RedisPendingMessageResendJob(listeners, redisTemplate, groupName, redissonClient); + } + + /** + * 创建 Redis Stream 集群消费的容器 + * + * 基础知识：Redis Stream 的 xreadgroup 命令 + */ + @Bean(initMethod = "start", destroyMethod = "stop") + @ConditionalOnBean(AbstractStreamMessageListener.class) // 只有 AbstractStreamMessageListener 存在的时候，才需要注册 Redis pubsub 监听 + public StreamMessageListenerContainer> redisStreamMessageListenerContainer( + RedisMQTemplate redisMQTemplate, List> listeners) { + RedisTemplate redisTemplate = redisMQTemplate.getRedisTemplate(); + checkRedisVersion(redisTemplate); + // 第一步，创建 StreamMessageListenerContainer 容器 + // 创建 options 配置 + StreamMessageListenerContainer.StreamMessageListenerContainerOptions> containerOptions = + StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder() + .batchSize(10) // 一次性最多拉取多少条消息 + .targetType(String.class) // 目标类型。统一使用 String，通过自己封装的 AbstractStreamMessageListener 去反序列化 + .build(); + // 创建 container 对象 + StreamMessageListenerContainer> container = + StreamMessageListenerContainer.create(redisMQTemplate.getRedisTemplate().getRequiredConnectionFactory(), containerOptions); + + // 第二步，注册监听器，消费对应的 Stream 主题 + String consumerName = buildConsumerName(); + listeners.parallelStream().forEach(listener -> { + log.info("[redisStreamMessageListenerContainer][开始注册 StreamKey({}) 对应的监听器({})]", + listener.getStreamKey(), listener.getClass().getName()); + // 创建 listener 对应的消费者分组 + try { + redisTemplate.opsForStream().createGroup(listener.getStreamKey(), listener.getGroup()); + } catch (Exception ignore) { + } + // 设置 listener 对应的 redisTemplate + listener.setRedisMQTemplate(redisMQTemplate); + // 创建 Consumer 对象 + Consumer consumer = Consumer.from(listener.getGroup(), consumerName); + // 设置 Consumer 消费进度，以最小消费进度为准 + StreamOffset streamOffset = StreamOffset.create(listener.getStreamKey(), ReadOffset.lastConsumed()); + // 设置 Consumer 监听 + StreamMessageListenerContainer.StreamReadRequestBuilder builder = StreamMessageListenerContainer.StreamReadRequest + .builder(streamOffset).consumer(consumer) + .autoAcknowledge(false) // 不自动 ack + .cancelOnError(throwable -> false); // 默认配置，发生异常就取消消费，显然不符合预期；因此，我们设置为 false + container.register(builder.build(), listener); + log.info("[redisStreamMessageListenerContainer][完成注册 StreamKey({}) 对应的监听器({})]", + listener.getStreamKey(), listener.getClass().getName()); + }); + return container; + } + + /** + * 构建消费者名字，使用本地 IP + 进程编号的方式。 + * 参考自 RocketMQ clientId 的实现 + * + * @return 消费者名字 + */ + private static String buildConsumerName() { + return String.format("%s@%d", SystemUtil.getHostInfo().getAddress(), SystemUtil.getCurrentPID()); + } + + /** + * 校验 Redis 版本号，是否满足最低的版本号要求！ + */ + private static void checkRedisVersion(RedisTemplate redisTemplate) { + // 获得 Redis 版本 + Properties info = redisTemplate.execute((RedisCallback) RedisServerCommands::info); + String version = MapUtil.getStr(info, "redis_version"); + // 校验最低版本必须大于等于 5.0.0 + int majorVersion = Integer.parseInt(StrUtil.subBefore(version, '.', false)); + if (majorVersion < 5) { + throw new IllegalStateException(StrUtil.format("您当前的 Redis 版本为 {}，小于最低要求的 5.0.0 版本！" + + "请参考 {} 文档进行安装。", version, DocumentEnum.REDIS_INSTALL.getUrl())); + } + } + +} 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 new file mode 100644 index 0000000..92a0f77 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/RedisMQTemplate.java @@ -0,0 +1,87 @@ +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 lombok.AllArgsConstructor; +import lombok.Getter; +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.connection.stream.StreamRecords; +import org.springframework.data.redis.core.RedisTemplate; + +import java.util.ArrayList; +import java.util.List; + +/** + * Redis MQ 操作模板类 + * + * @author 芋道源码 + */ +@AllArgsConstructor +public class RedisMQTemplate { + + @Getter + private final RedisTemplate redisTemplate; + /** + * 拦截器数组 + */ + @Getter + private final List interceptors = new ArrayList<>(); + + /** + * 发送 Redis 消息，基于 Redis pub/sub 实现 + * + * @param message 消息 + */ + public void send(T message) { + try { + sendMessageBefore(message); + // 发送消息 + redisTemplate.convertAndSend(message.getChannel(), JsonUtils.toJsonString(message)); + } finally { + sendMessageAfter(message); + } + } + + /** + * 发送 Redis 消息，基于 Redis Stream 实现 + * + * @param message 消息 + * @return 消息记录的编号对象 + */ + public RecordId send(T message) { + try { + sendMessageBefore(message); + // 发送消息 + return redisTemplate.opsForStream().add(StreamRecords.newRecord() + .ofObject(JsonUtils.toJsonString(message)) // 设置内容 + .withStreamKey(message.getStreamKey())); // 设置 stream key + } finally { + sendMessageAfter(message); + } + } + + /** + * 添加拦截器 + * + * @param interceptor 拦截器 + */ + public void addInterceptor(RedisMessageInterceptor interceptor) { + interceptors.add(interceptor); + } + + private void sendMessageBefore(AbstractRedisMessage message) { + // 正序 + interceptors.forEach(interceptor -> interceptor.sendMessageBefore(message)); + } + + private void sendMessageAfter(AbstractRedisMessage message) { + // 倒序 + for (int i = interceptors.size() - 1; i >= 0; i--) { + interceptors.get(i).sendMessageAfter(message); + } + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/interceptor/RedisMessageInterceptor.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/interceptor/RedisMessageInterceptor.java new file mode 100644 index 0000000..dbcee7f --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/interceptor/RedisMessageInterceptor.java @@ -0,0 +1,26 @@ +package cn.iocoder.yudao.framework.mq.redis.core.interceptor; + +import cn.iocoder.yudao.framework.mq.redis.core.message.AbstractRedisMessage; + +/** + * {@link AbstractRedisMessage} 消息拦截器 + * 通过拦截器，作为插件机制，实现拓展。 + * 例如说，多租户场景下的 MQ 消息处理 + * + * @author 芋道源码 + */ +public interface RedisMessageInterceptor { + + default void sendMessageBefore(AbstractRedisMessage message) { + } + + default void sendMessageAfter(AbstractRedisMessage message) { + } + + default void consumeMessageBefore(AbstractRedisMessage message) { + } + + default void consumeMessageAfter(AbstractRedisMessage 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 new file mode 100644 index 0000000..02ede12 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/job/RedisPendingMessageResendJob.java @@ -0,0 +1,100 @@ +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 lombok.AllArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.redisson.api.RLock; +import org.redisson.api.RedissonClient; +import org.springframework.data.domain.Range; +import org.springframework.data.redis.connection.stream.*; +import org.springframework.data.redis.core.StreamOperations; +import org.springframework.scheduling.annotation.Scheduled; + +import java.util.List; +import java.util.Map; +import java.util.Objects; + +/** + * 这个任务用于处理，crash 之后的消费者未消费完的消息 + */ +@Slf4j +@AllArgsConstructor +public class RedisPendingMessageResendJob { + + private static final String LOCK_KEY = "redis:pending:msg:lock"; + + /** + * 消息超时时间，默认 5 分钟 + * + * 1. 超时的消息才会被重新投递 + * 2. 由于定时任务 1 分钟一次，消息超时后不会被立即重投，极端情况下消息5分钟过期后，再等 1 分钟才会被扫瞄到 + */ + private static final int EXPIRE_TIME = 5 * 60; + + private final List> listeners; + private final RedisMQTemplate redisTemplate; + private final String groupName; + private final RedissonClient redissonClient; + + /** + * 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题 + */ + @Scheduled(cron = "35 * * * * ?") + public void messageResend() { + RLock lock = redissonClient.getLock(LOCK_KEY); + // 尝试加锁 + if (lock.tryLock()) { + try { + execute(); + } catch (Exception ex) { + log.error("[messageResend][执行异常]", ex); + } finally { + lock.unlock(); + } + } + } + + /** + * 执行清理逻辑 + * + * @see 讨论 + */ + private void execute() { + StreamOperations ops = redisTemplate.getRedisTemplate().opsForStream(); + listeners.forEach(listener -> { + PendingMessagesSummary pendingMessagesSummary = Objects.requireNonNull(ops.pending(listener.getStreamKey(), groupName)); + // 每个消费者的 pending 队列消息数量 + Map pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer(); + pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> { + log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount); + // 每个消费者的 pending消息的详情信息 + PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(groupName, consumerName), Range.unbounded(), pendingMessageCount); + if (pendingMessages.isEmpty()) { + return; + } + pendingMessages.forEach(pendingMessage -> { + // 获取消息上一次传递到 consumer 的时间, + long lastDelivery = pendingMessage.getElapsedTimeSinceLastDelivery().getSeconds(); + if (lastDelivery < EXPIRE_TIME){ + return; + } + // 获取指定 id 的消息体 + List> records = ops.range(listener.getStreamKey(), + Range.of(Range.Bound.inclusive(pendingMessage.getIdAsString()), Range.Bound.inclusive(pendingMessage.getIdAsString()))); + if (CollUtil.isEmpty(records)) { + return; + } + // 重新投递消息 + redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord() + .ofObject(records.get(0).getValue()) // 设置内容 + .withStreamKey(listener.getStreamKey())); + // ack 消息消费完成 + redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, records.get(0)); + log.info("[processPendingMessage][消息({})重新投递成功]", records.get(0).getId()); + }); + }); + }); + } +} diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/message/AbstractRedisMessage.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/message/AbstractRedisMessage.java new file mode 100644 index 0000000..ee40814 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/message/AbstractRedisMessage.java @@ -0,0 +1,29 @@ +package cn.iocoder.yudao.framework.mq.redis.core.message; + +import lombok.Data; + +import java.util.HashMap; +import java.util.Map; + +/** + * Redis 消息抽象基类 + * + * @author 芋道源码 + */ +@Data +public abstract class AbstractRedisMessage { + + /** + * 头 + */ + private Map headers = new HashMap<>(); + + public String getHeader(String key) { + return headers.get(key); + } + + public void addHeader(String key, String value) { + headers.put(key, value); + } + +} 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 new file mode 100644 index 0000000..ce03efc --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessage.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 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 new file mode 100644 index 0000000..bae90a9 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/pubsub/AbstractChannelMessageListener.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 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/stream/AbstractStreamMessage.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessage.java new file mode 100644 index 0000000..72c6aac --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessage.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 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 new file mode 100644 index 0000000..4a4ea00 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/core/stream/AbstractStreamMessageListener.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 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/src/main/java/cn/iocoder/yudao/framework/mq/redis/package-info.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/package-info.java new file mode 100644 index 0000000..6621fc1 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/package-info.java @@ -0,0 +1,6 @@ +/** + * 消息队列，基于 Redis 提供： + * 1. 基于 Pub/Sub 实现广播消费 + * 2. 基于 Stream 实现集群消费 + */ +package cn.iocoder.yudao.framework.mq.redis; diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainerX.java b/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainerX.java deleted file mode 100644 index b4cf4c5..0000000 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainerX.java +++ /dev/null @@ -1,62 +0,0 @@ -package org.springframework.data.redis.stream; - -import cn.hutool.core.util.ReflectUtil; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.connection.stream.ByteRecord; -import org.springframework.data.redis.connection.stream.ReadOffset; -import org.springframework.data.redis.connection.stream.Record; -import org.springframework.util.Assert; - -import java.util.Collections; -import java.util.List; -import java.util.function.Function; - -/** - * 拓展 DefaultStreamMessageListenerContainer 实现，解决 Spring Data Redis + Redisson 结合使用时，Redisson 在 Stream 获得不到数据时，返回 null 而不是空 List，导致 NPE 异常。 - * 对应 issue：https://github.com/spring-projects/spring-data-redis/issues/2147 和 https://github.com/redisson/redisson/issues/4006 - * 目前看下来 Spring Data Redis 不肯加 null 判断，Redisson 暂时也没改返回 null 到空 List 的打算，所以暂时只能自己改，哽咽！ - * - * @author 芋道源码 - */ -public class DefaultStreamMessageListenerContainerX> extends DefaultStreamMessageListenerContainer { - - /** - * 参考 {@link StreamMessageListenerContainer#create(RedisConnectionFactory, StreamMessageListenerContainerOptions)} 的实现 - */ - public static > StreamMessageListenerContainer create(RedisConnectionFactory connectionFactory, StreamMessageListenerContainer.StreamMessageListenerContainerOptions options) { - Assert.notNull(connectionFactory, "RedisConnectionFactory must not be null!"); - Assert.notNull(options, "StreamMessageListenerContainerOptions must not be null!"); - return new DefaultStreamMessageListenerContainerX<>(connectionFactory, options); - } - - public DefaultStreamMessageListenerContainerX(RedisConnectionFactory connectionFactory, StreamMessageListenerContainerOptions containerOptions) { - super(connectionFactory, containerOptions); - } - - /** - * 参考 {@link DefaultStreamMessageListenerContainer#register(StreamReadRequest, StreamListener)} 的实现 - */ - @Override - public Subscription register(StreamReadRequest streamRequest, StreamListener listener) { - return this.doRegisterX(getReadTaskX(streamRequest, listener)); - } - - @SuppressWarnings("unchecked") - private StreamPollTask getReadTaskX(StreamReadRequest streamRequest, StreamListener listener) { - StreamPollTask task = ReflectUtil.invoke(this, "getReadTask", streamRequest, listener); - // 修改 readFunction 方法 - Function> readFunction = (Function>) ReflectUtil.getFieldValue(task, "readFunction"); - ReflectUtil.setFieldValue(task, "readFunction", (Function>) readOffset -> { - List records = readFunction.apply(readOffset); - //【重点】保证 records 不是空，避免 NPE 的问题！！！ - return records != null ? records : Collections.emptyList(); - }); - return task; - } - - private Subscription doRegisterX(Task task) { - return ReflectUtil.invoke(this, "doRegister", task); - } - -} - diff --git a/yudao-framework/yudao-spring-boot-starter-mq/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/yudao-framework/yudao-spring-boot-starter-mq/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index c47aa4d..29def72 100644 --- a/yudao-framework/yudao-spring-boot-starter-mq/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/yudao-framework/yudao-spring-boot-starter-mq/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1 +1 @@ -cn.iocoder.yudao.framework.mq.config.YudaoMQAutoConfiguration \ No newline at end of file +cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQAutoConfiguration diff --git a/yudao-server/src/main/resources/application.yaml b/yudao-server/src/main/resources/application.yaml index 1e507c8..ef195f3 100644 --- a/yudao-server/src/main/resources/application.yaml +++ b/yudao-server/src/main/resources/application.yaml @@ -145,12 +145,6 @@ yudao: - cn.iocoder.yudao.module.pay.enums.ErrorCodeConstants - cn.iocoder.yudao.module.system.enums.ErrorCodeConstants - cn.iocoder.yudao.module.mp.enums.ErrorCodeConstants - mq: - redis: - pubsub: - enable: false # 是否开启 Redis pubsub 广播消费，默认为 true。这里设置成 false，可以按需开启 - stream: - enable: false # 是否开启 Redis stream 集群消费，默认为 true。这里设置成 false，可以按需开启 tenant: # 多租户相关配置项 enable: true ignore-urls: -- libgit2 0.21.4