当前位置 博文首页 > 盛夏温暖流年：SpringBoot + Redis 实现发布订阅功能
pub/sub 是目前广泛使用的通信模型，它采用事件作为基本的通信机制，提供大规模系统所要求的松散耦合的交互模式：
订阅者（如客户端）以事件订阅的方式表达出它有兴趣接收的一个事件或一类事件；发布者（如服务器）可将订阅者感兴趣的事件随时通知相关订阅者。
同样，Redis 的 pub/sub 是一种消息通信模式，主要目的是解除消息发布者和消息订阅者之间的耦合， Redis 作为一个 pub/sub 的 server, 在订阅者和发布者之间起到了消息路由的功能。
当前的应用场景是：我们需要建立一个独立的监控系统来监控另外一个系统的接口运行状态，需要做的工作主要分为两部分：被监测系统中的消息发布和监测系统中的消息订阅和消费。
由于需要传输的信息较多，我们使用 json 格式进行消息交互。
消息发布端
消息发布端是被监测的系统，新建 RedisPublisher 类进行消息发布
public class RedisPublisher {
private final static Logger logger = LoggerFactory.getLogger(RedisPublisher.class);
private final JedisPool jedisPool;
public RedisPublisher(JedisPool jedisPool) {
this.jedisPool = jedisPool;
}
/**
* 发布一个消息
*
* @param channel
* @param message
*/
public void publishMsg(String channel, String message) {
Jedis jedis = null;
logger.info("发布消息，频道：" + channel + ";内容：" + message);
try {
jedis = jedisPool.getResource();
jedis.publish(channel, message);
} catch (Exception e) {
logger.error("第一次publish发布信息失败,channel:" + channel + "失败原因:",e);
// 再次尝试
try {
if (jedis == null) {
jedis = jedisPool.getResource();
}
jedis.publish(channel, message);
} catch (Exception ee) {
logger.error( "再次publish信息失败,channel:" + channel + "失败原因:",ee);
} finally {
if (jedis != null) {
try {
jedis.close();
} catch (Exception e) {
jedis = null;
}
}
}
}
}
}
业务内部调用 RedisPublisher 类中的 publishMsg 方法
JedisPool jedisPool = new JedisPool(new JedisPoolConfig(), "127.0.0.1", 6379, 0, "123456");
RedisPublisher redisPublisher = new RedisPublisher(jedisPool);
// 将需要发送的消息处理为 json 格式
JSONObject jsonObject = new JSONObject();
jsonObject.put("type", 1);
jsonObject.put("region", "北京");
RdeisPublisher.publishMsg("api_channel", jsonObject.toString());
消息接收端
消息接收端是监测系统，当有用户访问被监测系统后，会发布消息，监测系统监听到消息后进行对应的处理。
新建消息订阅业务处理类 BusinessSubService
/**
* 消息订阅业务处理类
*/
@Service
@Slf4j
public class BusinessSubService {
public void receiveMessage(String message) {
try {
// 使用hutool进行json处理
JSONObject jsonObject = new JSONObject(message);
// 具体的业务处理
......
} catch (Exception e) {
log.error("消息订阅处理发生异常", e);
}
}
}
新建消息订阅业务处理配置类 SubscriberConfig
/**
** 消息订阅业务处理配置类
**/
@Configuration
public class SubscriberConfig {
@Autowired
private BusinessSubService businessSubService;
@Bean
public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
MessageListenerAdapter listenerAdapter) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
List<PatternTopic> topicList = new ArrayList<>();
// api_channel 是对应的频道名称
topicList.add(new PatternTopic("api_channel"));
container.addMessageListener(listenerAdapter, topicList);
return container;
}
@Bean
public MessageListenerAdapter listenerAdapter() {
// receiveMessage 是 BusinessSubService 类中的接收信息的方法（名称需要对应起来）
return new MessageListenerAdapter(businessSubService, "receiveMessage");
}
}
Redis 发布订阅的消息并没有持久化机制，属于即发即弃模式，也就是说它们不能像 MQ 中的消息那样保证订阅者不会错过任何消息，无论这些消息订阅者是否在线。
由于本来就是即发即弃的消息模式，所以 Redis 也不需要专门制定消息的备份和恢复机制。
Redis 也没有为发布者和订阅者准备保证消息性能的任何方案，例如大量消息同时到达 Redis 服务时，如果消息订阅者来不及完成消费，就可能导致消息堆积，而 MQ 中有专门针对这种情况的慢消息机制。
如果能容忍这些缺点，Redis 的消息队列是个不错的选择方案。
cs