RedisMessageListenerContainer默认单线程串行处理消息，必须显式配置专用ThreadPoolTaskExecutor并设setTaskExecutor()，否则@Async无效、SimpleAsyncTaskExecutor易OOM；需设合理核心线程数、队列容量及CallerRunsPolicy拒绝策略。
RedisMessageListenerContainer 默认用的是单线程轮询，订阅端一旦消息量上来或处理逻辑稍重，就会卡住后续消息。直接用 Spring Boot 默认线程池（比如 @Async 的那个）也不行——它不归 RedisMessageListenerContainer 管，你加了 @Async 也没用。
必须显式给 RedisMessageListenerContainer 配一个独立的、可调优的线程池，否则所有监听器都在同一个线程里串行执行，发布再多消息也白搭。
为什么不能靠 @Async 解决订阅端并发
很多人试过在 onMessage 方法上加 @Async，发现没效果。原因很简单：RedisMessageListenerContainer 底层是靠自己的 TaskExecutor 拉取消息并分发，它根本不会走 Spring 的 AOP 代理链，@Async 注解压根不生效。
真正起作用的是容器自己持有的 taskExecutor 属性，它决定“谁来调用你的 onMessage”。不设这个，就默认用 SimpleAsyncTaskExecutor——每次新建线程，无复用、无限制、易 OOM。
-
SimpleAsyncTaskExecutor不是线程池，只是“每次都 new Thread()” - 消息积压时会疯狂创建线程，JVM 很快报
java.lang.OutOfMemoryError: unable to create new native thread - 无法控制并发数、队列长度、拒绝策略等关键参数
如何配置专用的 ThreadPoolTaskExecutor 给容器
你需要手动定义一个 ThreadPoolTaskExecutor Bean，并把它塞进 RedisMessageListenerContainer 实例里。别用 @Bean 直接覆盖 Spring Boot 自动装配的那个容器——它可能已被其他组件依赖，改了会出问题。稳妥做法是：自己 new 一个容器，配好线程池，再注册为 Bean。
- 线程名前缀建议带
redis-listener-，方便在堆栈和监控里识别 -
corePoolSize建议设为 2～4，除非你有多个频道且每个频道消息都很密集 -
maxPoolSize不宜过大，超过 10 就要警惕上下文切换开销 - 拒绝策略推荐
CallerRunsPolicy：让主线程（即消息拉取线程）自己执行一条，起到自然限流作用
@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory factory,
MessageListenerAdapter listenerAdapter) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(factory);
container.addMessageListener(listenerAdapter, new PatternTopic("orders.*"));
// ⬇️ 关键：注入自定义线程池
container.setTaskExecutor(redisListenerExecutor());
return container;
}
@Bean
public ThreadPoolTaskExecutor redisListenerExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(3);
executor.setMaxPoolSize(6);
executor.setQueueCapacity(100);
executor.setThreadNamePrefix("redis-listener-");
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
频道模式匹配与线程分配的关系
PatternTopic 和 ChannelTopic 对线程使用方式没区别：一个线程池服务所有注册的 topic。但要注意，如果你同时监听 PatternTopic("orders.*") 和 ChannelTopic("system.alert")，它们共享同一个线程池，彼此会争抢线程资源。
- 高频小消息（如心跳）和低频大消息（如订单快照）混在一个池子里，容易互相拖慢
- 真有隔离需求，就得拆成两个
RedisMessageListenerContainerBean，各自配独立线程池 - 不要指望 Lettuce 的
EventLoopGroup能帮你分担——那是网络 I/O 层，和业务消息分发无关
别忽略连接断开后的重订阅线程安全
Lettuce 在连接断开后会自动重连，但重连成功后重新 SUBSCRIBE 是同步操作，如果此时线程池已满、正在执行大量 onMessage，这个重订阅动作可能被阻塞，导致“看起来消息收不到了”。
- 确保线程池
queueCapacity足够容纳重连期间积压的“重订阅任务” - 不要在
onMessage里做耗时 IO（如远程 HTTP 调用），否则会堵死整个池子 - 用
logging.level.org.springframework.data.redis.listener=DEBUG可看到重订阅日志，用于排查是否卡在这一步
线程池不是配了就完事。真正容易被忽略的是：当你的 onMessage 方法里抛出未捕获异常时，Lettuce 默认会静默吞掉它，既不重试也不打日志，线程池里的那个线程就“看似正常”地继续空转——而你的业务逻辑其实已经停摆了。务必在监听器里包一层 try-catch，至少打 ERROR 日志。
