Commit a597eb17d45c5cdf6d1b85040d49459cc30b1b84
1 parent
0233b092
websocket:修复 RedisWebSocketMessageConsumer 注册失效的问题
Showing
4 changed files
with
38 additions
and
19 deletions
yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQAutoConfiguration.java renamed to yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQConsumerAutoConfiguration.java
| @@ -5,7 +5,6 @@ import cn.hutool.core.util.StrUtil; | @@ -5,7 +5,6 @@ import cn.hutool.core.util.StrUtil; | ||
| 5 | import cn.hutool.system.SystemUtil; | 5 | import cn.hutool.system.SystemUtil; |
| 6 | import cn.iocoder.yudao.framework.common.enums.DocumentEnum; | 6 | import cn.iocoder.yudao.framework.common.enums.DocumentEnum; |
| 7 | import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; | 7 | import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; |
| 8 | -import cn.iocoder.yudao.framework.mq.redis.core.interceptor.RedisMessageInterceptor; | ||
| 9 | import cn.iocoder.yudao.framework.mq.redis.core.job.RedisPendingMessageResendJob; | 8 | import cn.iocoder.yudao.framework.mq.redis.core.job.RedisPendingMessageResendJob; |
| 10 | import cn.iocoder.yudao.framework.mq.redis.core.pubsub.AbstractRedisChannelMessageListener; | 9 | import cn.iocoder.yudao.framework.mq.redis.core.pubsub.AbstractRedisChannelMessageListener; |
| 11 | import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessageListener; | 10 | import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessageListener; |
| @@ -23,7 +22,6 @@ import org.springframework.data.redis.connection.stream.ReadOffset; | @@ -23,7 +22,6 @@ import org.springframework.data.redis.connection.stream.ReadOffset; | ||
| 23 | import org.springframework.data.redis.connection.stream.StreamOffset; | 22 | import org.springframework.data.redis.connection.stream.StreamOffset; |
| 24 | import org.springframework.data.redis.core.RedisCallback; | 23 | import org.springframework.data.redis.core.RedisCallback; |
| 25 | import org.springframework.data.redis.core.RedisTemplate; | 24 | import org.springframework.data.redis.core.RedisTemplate; |
| 26 | -import org.springframework.data.redis.core.StringRedisTemplate; | ||
| 27 | import org.springframework.data.redis.listener.ChannelTopic; | 25 | import org.springframework.data.redis.listener.ChannelTopic; |
| 28 | import org.springframework.data.redis.listener.RedisMessageListenerContainer; | 26 | import org.springframework.data.redis.listener.RedisMessageListenerContainer; |
| 29 | import org.springframework.data.redis.stream.StreamMessageListenerContainer; | 27 | import org.springframework.data.redis.stream.StreamMessageListenerContainer; |
| @@ -33,30 +31,19 @@ import java.util.List; | @@ -33,30 +31,19 @@ import java.util.List; | ||
| 33 | import java.util.Properties; | 31 | import java.util.Properties; |
| 34 | 32 | ||
| 35 | /** | 33 | /** |
| 36 | - * 消息队列配置类 | 34 | + * Redis 消息队列 Consumer 配置类 |
| 37 | * | 35 | * |
| 38 | * @author 芋道源码 | 36 | * @author 芋道源码 |
| 39 | */ | 37 | */ |
| 40 | @Slf4j | 38 | @Slf4j |
| 41 | @EnableScheduling // 启用定时任务,用于 RedisPendingMessageResendJob 重发消息 | 39 | @EnableScheduling // 启用定时任务,用于 RedisPendingMessageResendJob 重发消息 |
| 42 | @AutoConfiguration(after = YudaoRedisAutoConfiguration.class) | 40 | @AutoConfiguration(after = YudaoRedisAutoConfiguration.class) |
| 43 | -public class YudaoRedisMQAutoConfiguration { | ||
| 44 | - | ||
| 45 | - @Bean | ||
| 46 | - public RedisMQTemplate redisMQTemplate(StringRedisTemplate redisTemplate, | ||
| 47 | - List<RedisMessageInterceptor> interceptors) { | ||
| 48 | - RedisMQTemplate redisMQTemplate = new RedisMQTemplate(redisTemplate); | ||
| 49 | - // 添加拦截器 | ||
| 50 | - interceptors.forEach(redisMQTemplate::addInterceptor); | ||
| 51 | - return redisMQTemplate; | ||
| 52 | - } | ||
| 53 | - | ||
| 54 | - // ========== 消费者相关 ========== | 41 | +public class YudaoRedisMQConsumerAutoConfiguration { |
| 55 | 42 | ||
| 56 | /** | 43 | /** |
| 57 | * 创建 Redis Pub/Sub 广播消费的容器 | 44 | * 创建 Redis Pub/Sub 广播消费的容器 |
| 58 | */ | 45 | */ |
| 59 | - @Bean(initMethod = "start", destroyMethod = "stop") | 46 | + @Bean |
| 60 | @ConditionalOnBean(AbstractRedisChannelMessageListener.class) // 只有 AbstractChannelMessageListener 存在的时候,才需要注册 Redis pubsub 监听 | 47 | @ConditionalOnBean(AbstractRedisChannelMessageListener.class) // 只有 AbstractChannelMessageListener 存在的时候,才需要注册 Redis pubsub 监听 |
| 61 | public RedisMessageListenerContainer redisMessageListenerContainer( | 48 | public RedisMessageListenerContainer redisMessageListenerContainer( |
| 62 | RedisMQTemplate redisMQTemplate, List<AbstractRedisChannelMessageListener<?>> listeners) { | 49 | RedisMQTemplate redisMQTemplate, List<AbstractRedisChannelMessageListener<?>> listeners) { |
yudao-framework/yudao-spring-boot-starter-mq/src/main/java/cn/iocoder/yudao/framework/mq/redis/config/YudaoRedisMQProducerAutoConfiguration.java
0 → 100644
| 1 | +package cn.iocoder.yudao.framework.mq.redis.config; | ||
| 2 | + | ||
| 3 | +import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; | ||
| 4 | +import cn.iocoder.yudao.framework.mq.redis.core.interceptor.RedisMessageInterceptor; | ||
| 5 | +import cn.iocoder.yudao.framework.redis.config.YudaoRedisAutoConfiguration; | ||
| 6 | +import lombok.extern.slf4j.Slf4j; | ||
| 7 | +import org.springframework.boot.autoconfigure.AutoConfiguration; | ||
| 8 | +import org.springframework.context.annotation.Bean; | ||
| 9 | +import org.springframework.data.redis.core.StringRedisTemplate; | ||
| 10 | + | ||
| 11 | +import java.util.List; | ||
| 12 | + | ||
| 13 | +/** | ||
| 14 | + * Redis 消息队列 Producer 配置类 | ||
| 15 | + * | ||
| 16 | + * @author 芋道源码 | ||
| 17 | + */ | ||
| 18 | +@Slf4j | ||
| 19 | +@AutoConfiguration(after = YudaoRedisAutoConfiguration.class) | ||
| 20 | +public class YudaoRedisMQProducerAutoConfiguration { | ||
| 21 | + | ||
| 22 | + @Bean | ||
| 23 | + public RedisMQTemplate redisMQTemplate(StringRedisTemplate redisTemplate, | ||
| 24 | + List<RedisMessageInterceptor> interceptors) { | ||
| 25 | + RedisMQTemplate redisMQTemplate = new RedisMQTemplate(redisTemplate); | ||
| 26 | + // 添加拦截器 | ||
| 27 | + interceptors.forEach(redisMQTemplate::addInterceptor); | ||
| 28 | + return redisMQTemplate; | ||
| 29 | + } | ||
| 30 | + | ||
| 31 | +} |
yudao-framework/yudao-spring-boot-starter-mq/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
| 1 | -cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQAutoConfiguration | 1 | +cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQProducerAutoConfiguration |
| 2 | +cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQConsumerAutoConfiguration | ||
| 2 | cn.iocoder.yudao.framework.mq.rabbitmq.config.YudaoRabbitMQAutoConfiguration | 3 | cn.iocoder.yudao.framework.mq.rabbitmq.config.YudaoRabbitMQAutoConfiguration |
| 3 | \ No newline at end of file | 4 | \ No newline at end of file |
yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/config/YudaoWebSocketAutoConfiguration.java
| 1 | package cn.iocoder.yudao.framework.websocket.config; | 1 | package cn.iocoder.yudao.framework.websocket.config; |
| 2 | 2 | ||
| 3 | +import cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQConsumerAutoConfiguration; | ||
| 3 | import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; | 4 | import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; |
| 4 | import cn.iocoder.yudao.framework.websocket.core.handler.JsonWebSocketMessageHandler; | 5 | import cn.iocoder.yudao.framework.websocket.core.handler.JsonWebSocketMessageHandler; |
| 5 | import cn.iocoder.yudao.framework.websocket.core.listener.WebSocketMessageListener; | 6 | import cn.iocoder.yudao.framework.websocket.core.listener.WebSocketMessageListener; |
| @@ -38,7 +39,7 @@ import java.util.List; | @@ -38,7 +39,7 @@ import java.util.List; | ||
| 38 | * | 39 | * |
| 39 | * @author xingyu4j | 40 | * @author xingyu4j |
| 40 | */ | 41 | */ |
| 41 | -@AutoConfiguration | 42 | +@AutoConfiguration(before = YudaoRedisMQConsumerAutoConfiguration.class) // before YudaoRedisMQConsumerAutoConfiguration 的原因是,需要保证 RedisWebSocketMessageConsumer 先创建,才能创建 RedisMessageListenerContainer |
| 42 | @EnableWebSocket // 开启 websocket | 43 | @EnableWebSocket // 开启 websocket |
| 43 | @ConditionalOnProperty(prefix = "yudao.websocket", value = "enable", matchIfMissing = true) // 允许使用 yudao.websocket.enable=false 禁用 websocket | 44 | @ConditionalOnProperty(prefix = "yudao.websocket", value = "enable", matchIfMissing = true) // 允许使用 yudao.websocket.enable=false 禁用 websocket |
| 44 | @EnableConfigurationProperties(WebSocketProperties.class) | 45 | @EnableConfigurationProperties(WebSocketProperties.class) |
| @@ -98,7 +99,6 @@ public class YudaoWebSocketAutoConfiguration { | @@ -98,7 +99,6 @@ public class YudaoWebSocketAutoConfiguration { | ||
| 98 | return new RedisWebSocketMessageSender(sessionManager, redisMQTemplate); | 99 | return new RedisWebSocketMessageSender(sessionManager, redisMQTemplate); |
| 99 | } | 100 | } |
| 100 | 101 | ||
| 101 | - // TODO 芋艿:需要额外删除 YudaoRedisMQAutoConfiguration 的 RedisMessageListenerContainer Bean 上的 @ConditionalOnBean 注解。可能是 spring boot 的 bug! | ||
| 102 | @Bean | 102 | @Bean |
| 103 | public RedisWebSocketMessageConsumer redisWebSocketMessageConsumer( | 103 | public RedisWebSocketMessageConsumer redisWebSocketMessageConsumer( |
| 104 | RedisWebSocketMessageSender redisWebSocketMessageSender) { | 104 | RedisWebSocketMessageSender redisWebSocketMessageSender) { |