认识 Redis 的发布订阅功能

关于 Redis 发布订阅的功能介绍可以参考：订阅与发布。下面我来介绍下 Redis 发布订阅功能的特性和适用场景。

Redis 发布订阅功能的特性

• 消息的发送者与接收者之间通过 channel 绑定：channel 可以是确定的字符串，也可以基于模式匹配
• 客户端可以订阅任意多个 channel
• 发送者发送的消息无法持久化，所以可能会造成消息丢失
• 由于消息无法持久化，所以，消费者无法收到在订阅 channel 之间发送的消息
• 发送者与客户端之间的消息发送与接收不存在 ACK 机制

Redis 发布订阅功能的适用场景

由于没有消息持久化与 ACK 的保证，所以，Redis 的发布订阅功能并不可靠。这也就导致了它的应用场景很有限，建议用于实时与可靠性要求不高的场景。例如：

• 消息推送
• 内网环境的消息通知

总之，Redis 发布订阅功能足够简单，如果没有过多的要求，且不想搭建 Kafka、RabbitMQ 这样的可靠型消息系统时，可以考虑尝试使用 Redis。

Redis 发布订阅功能在 SpringBoot 中的关键类

Spring Data Redis 实现发布订阅功能非常简单，只有这样的几个类：Topic、MessageListener、RedisMessageListenerContainer。下面对它们进行解释：

• org.springframework.data.redis.listener.Topic： 消息发送者与接收者之间的 channel 定义，有两个实现类
  • org.springframework.data.redis.listener.ChannelTopic：一个确定的字符串
  • org.springframework.data.redis.listener.PatternTopic：基于模式匹配
• org.springframework.data.redis.connection.MessageListener： 一个回调接口，消息监听器，用于接收发送到 channel 的消息，接口定义如下：

public interface MessageListener {/*** Callback for processing received objects through Redis.* * @param message message* @param pattern pattern matching the channel (if specified) - can be null*/void onMessage(Message message, byte[] pattern);
}

• org.springframework.data.redis.listener.RedisMessageListenerContainer： 用于消息监听，需要将 Topic 和 MessageListener 注册到 RedisMessageListenerContainer 中。这样，当 Topic 上有消息时，由 RedisMessageListenerContainer 通知 MessageListener，客户端通过 onMessage 拿到消息后，自行处理。

Redis 发布订阅功能在 SpringBoot 中的实践

maven依赖包

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

实体对象

@Data
@NoArgsConstructor
@AllArgsConstructor
@ToString
public class User {private String userName;private String password;private int age;private String email;private Date createDate;
}

实现方式1

redis配置类

@Configuration
public class RedisConfig {// 默认情况下RedisTemplate模板只能支持字符串，我们自定义一个RedisTemplate，设置序列化器，这样我们可以很方便的操作实例对象。@Beanpublic RedisTemplate<String, Serializable> redisTemplate(LettuceConnectionFactory connectionFactory) {RedisTemplate<String, Serializable> redisTemplate = new RedisTemplate<>();redisTemplate.setKeySerializer(new StringRedisSerializer());redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());// 解决存储hash时key&value乱码问题redisTemplate.setHashKeySerializer(new StringRedisSerializer());redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());redisTemplate.setConnectionFactory(connectionFactory);return redisTemplate;}// 配置用户注册消息监听器@Beanpublic UserRegisterMessageListener userRegisterMessageListener() {return new UserRegisterMessageListener();}// 配置用户注销消息监听器@Beanpublic UserLogoutMessageListener userLogoutMessageListener() {return new UserLogoutMessageListener();}// 配置用户消息(包括注册、注销等)监听器@Beanpublic UserMessageListener userMessageListener() {return new UserMessageListener();}// 将消息监听器绑定到消息容器@Beanpublic RedisMessageListenerContainer messageListenerContainer(LettuceConnectionFactory lettuceConnectionFactory) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(lettuceConnectionFactory);UserRegisterMessageListener userRegisterMessageListener = userRegisterMessageListener();// 订阅用户注册消息话题container.addMessageListener(userRegisterMessageListener, userRegisterMessageListener.topic());UserLogoutMessageListener userLogoutMessageListener = userLogoutMessageListener();// 订阅用户注销消息话题container.addMessageListener(userLogoutMessageListener, userLogoutMessageListener.topic());UserMessageListener userMessageListener = userMessageListener();// 订阅用户消息(包括注册、注销等)话题container.addMessageListener(userMessageListener, userMessageListener.topic());return container;}
}

消息订阅者

public interface MessageEventListener extends MessageListener {/*** 订阅者订阅的话题** @return topic*/Topic topic();
}

订阅用户注册消息

@Slf4j
public class UserRegisterMessageListener implements MessageEventListener {public static final String TOPIC_NAME = "jaemon:user:register";@Autowiredprivate RedisTemplate redisTemplate;@Overridepublic void onMessage(Message message, byte[] pattern) {String body = new String(message.getBody());String channel = new String(message.getChannel());// 反序列化消息对象User user = (User) redisTemplate.getValueSerializer().deserialize(message.getBody());// 如果是 ChannelTopic, 则 channel 和 new String(pattern) 的值是相同的log.info("用户注册事件: body=[{}], channel=[{}], pattern=[{}]", body, channel, new String(pattern));}@Overridepublic Topic topic() {return new ChannelTopic(TOPIC_NAME);}
}

订阅用户注销消息

@Slf4j
public class UserLogoutMessageListener implements MessageEventListener {public static final String TOPIC_NAME = "jaemon:user:logout";@Overridepublic void onMessage(Message message, byte[] pattern) {String body = new String(message.getBody());String channel = new String(message.getChannel());// 如果是 ChannelTopic, 则 channel 和 new String(pattern) 的值是相同的log.info("用户注销事件: body=[{}], channel=[{}], pattern=[{}]", body, channel, new String(pattern));}@Overridepublic Topic topic() {return new ChannelTopic(TOPIC_NAME);}
}

订阅用户消息(包括注册、注销等)

@Slf4j
public class UserMessageListener implements MessageEventListener {public static final String TOPIC_NAME = "jaemon:user:*";@Overridepublic void onMessage(Message message, byte[] pattern) {String body = new String(message.getBody());String channel = new String(message.getChannel());// 如果是 ChannelTopic, 则 channel 和 new String(pattern) 的值是相同的log.info("用户事件: body=[{}], channel=[{}], pattern=[{}]", body, channel, new String(pattern));}@Overridepublic Topic topic() {return new PatternTopic(TOPIC_NAME);}
}

测试用例

@RunWith(SpringRunner.class)
@SpringBootTest(classes = {AssetApplication.class}, webEnvironment = SpringBootTest.WebEnvironment.NONE)
public class RedisPubSubTest {@Autowiredprivate RedisTemplate redisTemplate;@Testpublic void doJob() {// 发布用户注册消息redisTemplate.convertAndSend(UserRegisterMessageListener.TOPIC_NAME,new User("jaemon", "123456", 20, "answer_ljm@163.com", new Date()));// 发布用户注销消息redisTemplate.convertAndSend(UserLogoutMessageListener.TOPIC_NAME,new User("jaemon", "7654321", 20, "answer_ljm@163.com", new Date()));}
}

运行结果

用户注册事件: body=[{"@class":"com.jaemon.entity.User","userName":"jaemon","password":"123456","age":20,"email":"answer_ljm@163.com","createDate":["java.util.Date",1596709760919]}], channel=[jaemon:user:register], pattern=[jaemon:user:register]
用户事件: body=[{"@class":"com.jaemon.entity.User","userName":"jaemon","password":"123456","age":20,"email":"answer_ljm@163.com","createDate":["java.util.Date",1596709760919]}], channel=[jaemon:user:register], pattern=[jaemon:user:*]
用户注销事件: body=[{"@class":"com.jaemon.entity.User","userName":"jaemon","password":"7654321","age":20,"email":"answer_ljm@163.com","createDate":["java.util.Date",1596709761023]}], channel=[jaemon:user:logout], pattern=[jaemon:user:logout]
用户事件: body=[{"@class":"com.jaemon.entity.User","userName":"jaemon","password":"7654321","age":20,"email":"answer_ljm@163.com","createDate":["java.util.Date",1596709761023]}], channel=[jaemon:user:logout], pattern=[jaemon:user:*]

实现方式2

redis配置类

@Configuration
public class RedisConfig {// 默认情况下RedisTemplate模板只能支持字符串，我们自定义一个RedisTemplate，设置序列化器，这样我们可以很方便的操作实例对象。@Beanpublic RedisTemplate<String, Serializable> redisTemplate(LettuceConnectionFactory connectionFactory) {RedisTemplate<String, Serializable> redisTemplate = new RedisTemplate<>();redisTemplate.setKeySerializer(new StringRedisSerializer());redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());redisTemplate.setConnectionFactory(connectionFactory);return redisTemplate;}// 配置用户注册消息监听器@Bean(name = "userRegisterMessageListenerAdapter")public MessageListenerAdapter userRegisterMessageListenerAdapter(UserRegisterMessageListener userRegisterMessageListener) {return new MessageListenerAdapter(userRegisterMessageListener);}// 配置用户注销消息监听器@Bean(name = "userLogoutMessageListenerAdapter")public MessageListenerAdapter userLogoutMessageListenerAdapter(UserLogoutMessageListener userLogoutMessageListener) {return new MessageListenerAdapter(userLogoutMessageListener);}// 配置用户消息(包括注册、注销等)监听器@Bean(name = "userMessageListenerAdapter")public MessageListenerAdapter userMessageListenerAdapter(UserMessageListener userMessageListener) {// 可指定订阅者接受消息的方法return new MessageListenerAdapter(userMessageListener, "receiveMessage");}// 将消息监听器绑定到消息容器@Beanpublic RedisMessageListenerContainer messageListenerContainer(LettuceConnectionFactory lettuceConnectionFactory,MessageListenerAdapter userRegisterMessageListenerAdapter,MessageListenerAdapter userLogoutMessageListenerAdapter,MessageListenerAdapter userMessageListenerAdapter) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(lettuceConnectionFactory);// 订阅用户注册消息话题container.addMessageListener(userRegisterMessageListenerAdapter, new ChannelTopic(UserRegisterMessageListener.TOPIC_NAME));// 订阅用户注销消息话题container.addMessageListener(userLogoutMessageListenerAdapter, new ChannelTopic(UserLogoutMessageListener.TOPIC_NAME));// 订阅用户消息(包括注册、注销等)话题container.addMessageListener(userMessageListenerAdapter, new PatternTopic(UserMessageListener.TOPIC_NAME));return container;}
}

消息订阅者

订阅用户注册消息

@Slf4j
@Component
public class UserRegisterMessageListener {public static final String TOPIC_NAME = "jaemon:user:register";public void handleMessage(String message) {log.info("用户注册事件: message=[{}]", message);}}

订阅用户注销消息

@Slf4j
@Component
public class UserLogoutMessageListener {public static final String TOPIC_NAME = "jaemon:user:logout";public void handleMessage(String message) {log.info("用户注销事件: message=[{}]", message);}}

订阅用户消息(包括注册、注销等)

@Slf4j
@Component
public class UserMessageListener {public static final String TOPIC_NAME = "jaemon:user:*";public void receiveMessage(String message) {log.info("用户事件: message=[{}]", message);}
}

测试用例

@RunWith(SpringRunner.class)
@SpringBootTest(classes = {AssetApplication.class}, webEnvironment = SpringBootTest.WebEnvironment.NONE)
public class RedisPubSubTest {@Autowiredprivate RedisTemplate redisTemplate;@Testpublic void doJob() {// 发布用户注册消息redisTemplate.convertAndSend(UserRegisterMessageListener.TOPIC_NAME,new User("jaemon", "123456", 20, "answer_ljm@163.com", new Date()));// 发布用户注销消息redisTemplate.convertAndSend(UserLogoutMessageListener.TOPIC_NAME,new User("jaemon", "7654321", 20, "answer_ljm@163.com", new Date()));}
}

运行结果

用户注册事件: message=[{"@class":"com.jaemon.entity.User","userName":"jaemon","password":"123456","age":20,"email":"answer_ljm@163.com","createDate":["java.util.Date",1596765582710]}]
用户事件: message=[{"@class":"com.jaemon.entity.User","userName":"jaemon","password":"123456","age":20,"email":"answer_ljm@163.com","createDate":["java.util.Date",1596765582710]}]
用户注销事件: message=[{"@class":"com.jaemon.entity.User","userName":"jaemon","password":"7654321","age":20,"email":"answer_ljm@163.com","createDate":["java.util.Date",1596765582773]}]
用户事件: message=[{"@class":"com.jaemon.entity.User","userName":"jaemon","password":"7654321","age":20,"email":"answer_ljm@163.com","createDate":["java.util.Date",1596765582773]}]

我在哪些业务场景使用Redis发布订阅？

异步消息通知

比如渠道在调支付平台的时候，我们可以用回调的方式给支付平台一个我们的回调接口来通知我们支付状态，还可以利用Redis的发布订阅来实现。比如我们发起支付的同时订阅频道pay_notice_ + wk (假如我们的渠道标识是wk，不能让其他渠道也订阅这个频道)，当支付平台处理完成后，支付平台往该频道发布消息，告诉频道的订阅者该订单的支付信息及状态。收到消息后，根据消息内容更新订单信息及后续操作。

当很多人都调用支付平台时，支付时都去订阅同一个频道会有问题。比如用户A支付完订阅频道pay_notice_wk，在支付平台未处理完时，用户B支付完也订阅了pay_notice_wk，当A收到通知后，接着B的支付通知也发布了，这时渠道收不到第二次消息发布。因为同一个频道收到消息后，订阅自动取消，也就是订阅是一次性的。

所以我们订阅的订单支付状态的频道就得唯一，一个订单一个频道，我们可以在频道上加上订单号pay_notice_wk+orderNo保证频道唯一。这样我们可以把频道号在支付时当做参数一并传过去，支付平台处理完就可以用此频道发布消息给我们了。（实际大多接口用回调通知，因为用Redis发布订阅限制条件苛刻，系统间必须共用一套Redis）

任务通知

比如通过跑批系统通知应用系统做一些事（跑批系统无法拿到用户数据，且应用系统又不能做定时任务的情况下）。

如每天凌晨3点提前加载一些用户的用户数据到Redis，应用系统不能做定时任务，可以通过系统公共的Redis来由跑批系统发布任务给应用系统，应用系统收到指令，去做相应的操作。

这里需要注意的是在线上集群部署的情况下，所有服务实例都会收到通知，都要做同样的操作吗？完全没必要。可以用Redis实现锁机制，其中一台实例拿到锁后执行任务。另外如果任务比较耗时，可以不用锁，可以考虑一下任务分片执行。当然这不在本文的讨论范畴，这里不在赘述。

参数刷新加载

众所周知，我们用Redis无非就是将系统中不怎么变的、查询又比较频繁的数据缓存起来，例如我们系统首页的轮播图啊，页面的动态链接啊，一些系统参数啊，公共数据啊都加载到Redis，然后有个后台管理系统去配置修改这些数据。

打个比方我们首页的轮播图要再增加一个图，那我们就在后管系统加上，加上就完事了吗？当然没有，因为Redis里还是老数据。那你会说不是有过期时间吗？是的，但有的过期时间设置的较长如24小时并且我们想立即生效怎么办？这时候我们就可以利用Redis的发布订阅机制来实现数据的实时刷新。当我们修改完数据后，点击刷新按钮，通过发布订阅机制，订阅者接收到消息后调用重新加载的方法即可。

References

• Redis 的发布订阅功能在 SpringBoot 中的应用
• 我在生产项目里是如何使用Redis发布订阅的？（一）业务场景