Spring Boot集成Redis发布订阅实战：从原理到代码避坑指南

做 Spring Boot 项目的时候，Redis 的发布订阅功能我经常用来做轻量级的消息通知，比如订单状态变更广播、缓存失效通知、在线用户心跳提醒。Spring Boot 里默认的 Redis 客户端就是 Lettuce，集成发布订阅其实只需要几张配置类和监听器，但里面的序列化、频道模式、断线重连这些坑，文档不会给你写全。这篇文章就以一个可运行的 Spring Boot + Lettuce 发布订阅 Demo 为主线，把核心配置、代码实现、测试方法和排查思路一起梳理一遍。

如果你只是想快速跑通一个“生产者发消息、消费者收消息”的样例，百度一搜一大把。但真实项目里总会遇到“为什么别人能收到我收不到”“RedisTemplate 发出去以后 listener 拿到一堆二进制乱码”“服务重启以后消息丢了怎么办”之类的问题。这篇文章不会只给代码，还会把背后的原理和我实际踩过的坑一起讲透。

1. 项目整体思路与选型拆解

1.1 为什么用 Redis 发布订阅而不是消息队列

先要搞清楚一个核心问题：你需要的到底是“广播通知”还是“任务分发”。Redis 发布订阅，也就是 Pub/Sub，是一种很轻量的实时广播机制。一个消息发布到频道，所有订阅这个频道的客户端都会实时收到，消息不会存储在 Redis 里。它和消息队列最大的区别就是：不存在“消费者拉取”和“消息积压”的概念。

假设你现在要做一个订单系统，订单创建成功后需要通知积分服务、消息推送服务、库存服务。用 Pub/Sub 就很顺手：订单服务往orders.created频道发一条消息，所有监听这个频道的服务同时收到，各自做自己的事情。这个场景核心是“广播”，不是“谁抢到谁处理”。反过来，如果是一个邮件发送任务，希望多台机器消费同一个任务队列，一份任务只能被一台机器处理，那就不该用 Pub/Sub，应该用 Redis Stream 或者专业的消息队列。

我见过新手一上来就用发布订阅做异步任务队列，消息发出去没有消费者在线，任务直接就丢了。后来改成 Redis Stream 才解决问题。所以选型前先问自己一句：消息没了能不能接受？不能接受就换方案。

1.2 为什么 Spring Boot 默认选择 Lettuce

Spring Boot 的spring-boot-starter-data-redis默认集成的是 Lettuce 而不是 Jedis，这是很多人忽略的事实。Jedis 是阻塞式 IO，一个连接同时只能处理一个命令，多线程要使用连接池。Lettuce 基于 Netty，把连接复用做得很好，多个线程可以共享一个连接而不互相干扰，对异步操作、响应式编程支持也更友好。

对于发布订阅场景，Lettuce 的优势尤其明显。Redis 发布订阅要求订阅连接长期存在，专门负责监听频道消息。如果用 Jedis，你得单独维护一条长期存活的连接，并且不能在这个连接上执行普通命令，否则会破坏订阅状态。Lettuce 在底层已经把这层事情封装好了，Spring Data Redis 只需要通过RedisMessageListenerContainer去管理订阅生命周期即可。

我用过的项目中，Spring Boot 2.x 和 3.x 默认都是 Lettuce，不需要额外改依赖。如果你以前用过 Jedis，建议直接适应 Lettuce，Spring 官方围绕 Lettuce 做的集成最完整。

1.3 适合什么场景，不适合什么场景

适合用 Redis 发布订阅的场景，我梳理了几类：

• 配置变更通知：比如后台改了开关配置，通过频道广播通知各服务刷新本地缓存。
• 缓存失效通知：应用内存缓存过期后，广播给集群里的其他节点，避免脏数据。
• 实时在线推送：聊天室、站内信、用户状态变更这类低延迟、可容忍丢失的场景。
• 服务间轻量事件通知：订单创建、用户注册等事件，多个下游服务并行感知。

不适合的场景，也很明确：

• 要求消息不丢失，必须有持久化。Pub/Sub 消息是“发了就没了”。
• 要求消息能够回溯、重放。Pub/Sub 做不到。
• 要求一条消息只被一个消费者消费，需要消费组。Pub/Sub 广播是发给所有人的。
• 要求可靠的消息确认和重试。Pub/Sub 没有 ACK 机制，接收方处理失败也不知道。

这些边界想清楚，后面才不会在设计上踩大坑。

2. 快速搭建 Spring Boot 集成环境

2.1 依赖引入与版本建议

我用的是 Spring Boot 2.7.x，Redis 用的是本机或者 Docker 起的 6.x 以上。只需要引入一个 starter 就够：

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency>

如果你打算给 Lettuce 连接池设置参数，还得引入连接池依赖：

<dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-pool2</artifactId> </dependency>

Spring Boot 的自动配置会把 LettuceConnectionFactory、RedisTemplate、StringRedisTemplate 这些核心 Bean 都创建好，不需要手动注册。我们要做的只是在配置类里补充定制化的 Bean，比如消息监听容器。

2.2 配置文件：连接 Redis 与线程池参数

application.yml里基本的 Redis 配置如下：

spring: redis: host: 127.0.0.1 port: 6379 password: database: 0 timeout: 3000ms lettuce: pool: max-active: 16 max-idle: 8 min-idle: 2 max-wait: 3000ms

在发布订阅场景里，连接池的作用主要是给普通的 Redis 命令操作提供连接。订阅连接本身由RedisMessageListenerContainer管理，一般不会从连接池里占用一个连接长期不放。不过连接池参数还是建议设置，防止高并发下 RedisTemplate 操作连接不足。

有一个容易被忽视的配置是timeout。如果一个 Redis 服务端或网络环境不太好，订阅连接长时间空闲可能被中间层断开，timeout太短会加剧连接不稳定。我一般会把它设置为 3 到 5 秒，既要能快速暴露问题，也不要频繁超时。

2.3 定义频道常量与消息结构

频道名最好统一管理，不要到处写字符串。我习惯建一个常量类：

public final class RedisChannel { private RedisChannel() { } public static final String ORDER_CREATED = "orders.created"; public static final String ORDER_PAID = "orders.paid"; public static final String NOTICE = "notice"; }

如果一对一业务，写死字符串也能跑，但项目一旦有多个频道、多个模块，字符串散落在各处，第二天修改频道名时你会非常痛苦。

对于消息体，我建议频道里传输的永远是字符串，最好是 JSON 字符串。这样一是不受 Java 序列化影响，二是其他语言的服务也能消费。比如订单消息结构，可以直接定义一个 DTO：

public class OrderEvent { private String orderId; private String status; private Long timestamp; public OrderEvent() { } public OrderEvent(String orderId, String status) { this.orderId = orderId; this.status = status; this.timestamp = System.currentTimeMillis(); } // getter/setter 省略 }

通过 Jackson 手动序列化成 JSON 字符串再发出去，比直接塞 Java 对象靠谱得多。后面讲序列化坑的时候，你会明白为什么我这么坚持。

3. 核心功能实现：发布端与订阅端

3.1 消息监听容器与 MessageListener

Spring Data Redis 里专门有一个RedisMessageListenerContainer，负责维护订阅连接、把收到的消息分发给监听器。我们可以把它理解成一个“订阅管理器”：容器启动时去 Redis 执行 subscribe/psubscribe，消息到达后转交给对应的MessageListener。

实现一个监听器，最简单直接的方式是实现MessageListener接口：

@Component public class OrderMessageListener implements MessageListener { private static final Logger log = LoggerFactory.getLogger(OrderMessageListener.class); @Override public void onMessage(Message message, byte[] pattern) { String channel = new String(message.getChannel(), StandardCharsets.UTF_8); String body = new String(message.getBody(), StandardCharsets.UTF_8); log.info("收到频道 [{}] 的消息: {}", channel, body); // 这里可以直接根据频道名做不同处理 if (RedisChannel.ORDER_CREATED.equals(channel)) { handleOrderCreated(body); } } private void handleOrderCreated(String body) { // 解析 JSON，写业务逻辑 } }

然后在配置类里把这个监听器注册进容器：

@Configuration public class RedisPubSubConfig { @Bean public RedisMessageListenerContainer redisMessageListenerContainer( RedisConnectionFactory connectionFactory, OrderMessageListener orderMessageListener) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); // 监听精确频道 + 通配符频道 container.addMessageListener(orderMessageListener, Arrays.asList( new ChannelTopic(RedisChannel.ORDER_CREATED), new PatternTopic("orders.*") )); container.setRecoveryInterval(3000L); return container; } }

这里注意一个细节：addMessageListener的第二个参数可以传单个Topic，也可以传一个Topic列表。ChannelTopic表示精确频道，对应 Redis 命令SUBSCRIBE；PatternTopic表示频道模式，对应PSUBSCRIBE。如果你同时注册了orders.created精确频道和orders.*模式，发送到orders.created的消息会重复进入同一个监听器。我在编码时一般只选择一种方式，避免重复处理。

3.2 发布消息的两种方式与封装

发布消息最推荐的是StringRedisTemplate.convertAndSend()，因为消息体是 String，绕开了 RedisTemplate 默认序列化的坑。

@Service public class OrderMessagePublisher { private final StringRedisTemplate stringRedisTemplate; private final ObjectMapper objectMapper = new ObjectMapper(); public OrderMessagePublisher(StringRedisTemplate stringRedisTemplate) { this.stringRedisTemplate = stringRedisTemplate; } public void publishOrderCreated(String orderId) { OrderEvent event = new OrderEvent(orderId, "CREATED"); try { String json = objectMapper.writeValueAsString(event); stringRedisTemplate.convertAndSend(RedisChannel.ORDER_CREATED, json); } catch (JsonProcessingException e) { throw new RuntimeException("订单消息序列化失败", e); } } }

如果你一定要用RedisTemplate发送对象，RedisTemplate 的 ValueSerializer 必须和订阅端使用的序列化器对齐，否则监听器拿到的字节数组和预期不一致。同一个 Redis 连接工厂里，StringRedisTemplate和RedisTemplate是两套独立的 Bean，别以为它们底层序列化行为一样。

还有一个小技巧：发布消息时可以在 JSON 里带上一个全局唯一的消息 ID，下游收到后做幂等。发布订阅没有 ACK，收不到也没办法，但至少处理端不会因为网络重发或者其他原因重复执行业务逻辑。

3.3 MessageListenerAdapter 的用法与坑

网上很多教程会推荐MessageListenerAdapter，把监听器写成一个普通的 Bean，然后在配置类里适配：

@Bean MessageListenerAdapter orderListenerAdapter(OrderMessageListener listener) { MessageListenerAdapter adapter = new MessageListenerAdapter(listener); adapter.setSerializer(new StringRedisSerializer()); return adapter; }

这样监听器就不需要实现MessageListener接口，Spring 会通过反射调用默认名为handleMessage的方法。比如：

@Component public class OrderMessageListener { public void handleMessage(String message) { System.out.println(message); } }

这个写法很简洁，但坑在于：MessageListenerAdapter 默认会使用SerializingMessageConverter把消息体从字节数组反序列化成对象。如果你发布端用的是 StringRedisTemplate 发出的 JSON 字符串，而 MessageListenerAdapter 没有设置StringRedisSerializer，监听器方法里拿到的参数就是乱码，或者直接类型转换异常。所以如果使用这个模式，一定要记得调用adapter.setSerializer(new StringRedisSerializer())。

我个人更偏向直接实现MessageListener接口，少一层反射调用，也更透明。但对于那种只想监听某个固定频道、方法名固定的快速开发，MessageListenerAdapter 也够用。

4. 序列化、线程池与消息可靠性：容易被坑的细节

4.1 RedisTemplate 序列化器选择：JSON 还是 String

这是发布订阅功能里最经典的坑。Spring Boot 默认创建的RedisTemplate<String, Object>的 key 序列化器是JdkSerializationRedisSerializer，value 序列化器也是 JDK 序列化。这意味着用默认 RedisTemplate 的convertAndSend发送一个对象时，它会先经过 Java 序列化，带着一堆\xAC\xED\x00\x05的二进制头，再存进 Redis 的频道。

订阅端拿到Message.getBody()后直接转成 String 就是一堆乱码。要解决这个问题，有两条路可以走：

第一条路，也是最省心的，就是用StringRedisTemplate，它内部已经把 key 和 value 的序列化器都设成了StringRedisSerializer。我不需要特意去改任何 Bean 配置，直接把 JSON 字符串发出去，监听器拿到后就是 UTF-8 明文。

第二条路，自定义 RedisTemplate：

@Bean public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) { RedisTemplate<String, Object> template = new RedisTemplate<>(); template.setConnectionFactory(factory); StringRedisSerializer stringSerializer = new StringRedisSerializer(); GenericJackson2JsonRedisSerializer jsonSerializer = new GenericJackson2JsonRedisSerializer(); template.setKeySerializer(stringSerializer); template.setHashKeySerializer(stringSerializer); template.setValueSerializer(jsonSerializer); template.setHashValueSerializer(jsonSerializer); template.afterPropertiesSet(); return template; }

如果消息体是包含时间类型字段的对象，Jackson 默认可能不支持LocalDateTime，需要在 ObjectMapper 中注册 JavaTimeModule。这也是一个隐藏坑。

4.2 消息监听容器的线程模型与并发控制

很多人以为 RedisMessageListenerContainer 只是简单地在某个连接上循环读消息，其实它的内部有独立的订阅任务和消息分发线程。发布订阅消息到达监听器后，如果onMessage方法执行很慢，后面的消息就会被阻塞。原因很简单：同一个连接上，Redis 服务端按顺序推送消息，监听器是串行消费的。

我遇到过一个实际案例：有人在监听器里直接调用了一个外部 HTTP 接口，这个接口最慢耗了 5 秒，结果频道上后续所有消息都排队，最夸张的时候延迟几分钟。解决方式其实很简单，把耗时逻辑丢到线程池异步执行：

private final ExecutorService bizExecutor = Executors.newFixedThreadPool(4); @Override public void onMessage(Message message, byte[] pattern) { String body = new String(message.getBody(), StandardCharsets.UTF_8); bizExecutor.execute(() -> { try { // 耗时的业务处理 } catch (Exception e) { log.error("处理消息异常", e); } }); }

如果你同时订阅了许多频道，并且消息量很大，还可以给容器设置专门的任务执行器：

container.setTaskExecutor(Executors.newFixedThreadPool(8));

但要注意，这并不会把一个频道内的消息并发消费，Redis 频道本身的推送顺序仍然由底层连接持有。它更多是让多个监听器之间的分发不互相阻塞。

4.3 订阅断线重连与监听器存活问题

Redis 服务端重启、网络闪断、防火墙空闲断开，都会让订阅连接失效。如果容器没有重连机制，监听器就悄无声息地“死掉”了。Spring Data Redis 的RedisMessageListenerContainer默认是有恢复机制的，它内部有一个recoveryInterval参数，默认是 5000 毫秒。容器会定期检查订阅状态，如果发现连接断开，会尝试重新注册订阅。

我在配置里一般会显式缩短恢复时间，让业务对断连更敏感：

container.setRecoveryInterval(3000L);

但这里有个边界大家要明白：重连可以恢复“之后”的新消息，重连期间发布出来的消息依然会丢。这是发布订阅的天然限制，不是代码能解决的。生产环境如果对丢失敏感，请用 Redis Stream 或者把关键事件同步落库后由定时任务补偿。

另一点，如果服务是多实例部署的，每个实例都有自己的监听器，广播消息会同时进入所有实例。这种情况下要提前想好“重复消费”的问题。要么接受广播的语义，让各实例自己判断；要么在业务上做幂等，比如根据消息 ID 去重。

5. 测试与故障排查实录

5.1 用 redis-cli 和代码发布消息验证

先把服务跑起来，然后在命令行连接 Redis，往频道发一条消息，观察应用日志：

redis-cli -h 127.0.0.1 -p 6379 PUBLISH orders.created "hello-123"

如果监听器正常，应用日志会打印“收到频道 [orders.created] 的消息: hello-123”。

也可以先单独在 redis-cli 里订阅频道，确认频道名没写错：

SUBSCRIBE orders.created

这条命令会进入阻塞监听模式，此时另开一个 redis-cli 发布消息，能看到订阅输出，说明 Redis 本身没问题，问题可能在应用端。

我建议在开发环境加一个临时接口或者启动时测试发布，确认整个链路可用。比如用一个CommandLineRunner：

@Component public class PubSubStartupTester implements CommandLineRunner { private final OrderMessagePublisher publisher; public PubSubStartupTester(OrderMessagePublisher publisher) { this.publisher = publisher; } @Override public void run(String... args) { publisher.publishOrderCreated("startup-test"); } }

跑完记得删掉，避免每次启动都发测试消息污染环境。

5.2 常见故障速查表：收不到消息、乱码、失去连接

我把实际开发中遇到过的典型问题整理成了一个表格，排查的时候对照着看就行：

5.3 调试技巧：日志级别与监控频道

排查发布订阅问题，第一步是确认消息有没有到达应用这一层。给监听器加好日志是基础，但日志别只打印消息内容，把频道也带上。我之前就遇到过“明明收到了消息，但处理逻辑走了别的频道分支”，就是因为只打印了消息 body 没打印 channel。

第二，可以把 Spring Data Redis 的日志级别调成 DEBUG，观察订阅连接是否建立：

logging: level: org.springframework.data.redis.listener: DEBUG org.springframework.data.redis.connection.lettuce: DEBUG

第三，如果消息量不大，可以临时用一个“垃圾桶”监听器，订阅*模式，把所有发布消息都打出来，看看是不是消息确实能到订阅端。但要注意PSUBSCRIBE *在高频场景下压力很大，只适合临时排查。

6. 顺带聊聊进阶与替代方案

6.1 从 Pub/Sub 迁移到 Redis Stream 的取舍

如果你发现自己开始在 Pub/Sub 上做消费重试、消息持久化、消费组了，那就是它在越权。Redis Stream 才是 Redis 里更适合做消息队列的数据结构。Stream 的消息会持久化在 Redis 中，支持消费组，支持每条消息的 ACK，也支持按时间范围回溯。

两者的选择，不是谁更好，而是谁更匹配：

我在项目里常用一个组合：实时通知走 Pub/Sub，订单任务、积分类事件走 Redis Stream。两个都是 Redis 原生的，不用额外引入中间件，维护成本很低。

6.2 生产中值得增加的改进点

第一个改进是统一封装消息模型。不要直接暴露convertAndSend(channel, message)给业务方，而是封装一个Publisher，内部处理 JSON 序列化、频道命名、消息 ID 生成。这样以后如果换 Stream，业务方不需要感知。

第二个改进是增加简单的监控。比如在监听器里记录每秒处理消息数、最近一条消息到达时间。如果一段时间没有消息进来，可以认为是订阅连接异常，配合告警提前发现。这种方式比等服务完全挂掉再排查高效得多。

第三个改进是消息处理的异常隔离。监听器里永远要有一个统一的 try-catch，否则某个消息反序列化失败会直接中断整个监听流程。我在生产代码里习惯把所有失败的原始消息写到一个专门的“死信频道”或者 Redis List 里，然后由定时任务扫描重新处理。虽然发布订阅本身没有 ACK，但业务层面的兜底可以自己设计。

第四个改进是网络连接参数的调优。高并发场景下，Lettuce 的连接池和超时时间要结合实际压测调整，不要照抄网上的配置。尤其是订阅连接不通过连接池，但普通操作 latency 如果太高，也会间接影响整个 Redis 客户端的稳定性。

最后再分享一个小体会：发布订阅功能看起来简单，但真正碰过线上问题之后，会发现它和消息队列完全是两种思维模式。Simple 从来不代表没边界，理解了它“实时、广播、不持久”这三个特点，你自然就明白什么时候用对了，什么时候该换方案。我自己在多个项目里用这个套路做订单事件广播、配置刷新通知，跑得很稳，希望这份经验对你也有用。