SpringBoot+Redis自定义注解实现发布订阅
程
东哥约 5 分钟
东哥约 5 分钟1. 引言
@MessageHub 来管理消息消费。2. 常规的 Redis 发布订阅模式
**创建消息监听器 MessageListenerAdapter**：这是 Spring 中用于处理 Redis 消息的标准方式，基本上就是将一个消息处理方法适配成 Redis 可以识别的监听器。**创建订阅器 MessageListener**：订阅器会监听 Redis 频道上的消息，接收到消息后调用onMessage方法进行处理。**配置 RedisMessageListenerContainer**：通过这个容器来管理所有的消息监听器，并且把它们绑定到 Redis 上。
3. 问题分析：如何将发布订阅模式融入内部消息流转中间件
4. 消息组件的设计
生产消息：这个方法负责将消息发送到 Redis 频道，使用 StringRedisTemplate.convertAndSend()实现消息的生产。生产者就像是发送者，负责把消息推送到 Redis 上。消费消息：这个方法是比较复杂的，因为我们需要创建一个常驻的消费线程来处理从 Redis 中接收到的消息。我们必须考虑如何动态地注册这些消费线程，并确保能够回调到具体的方法。
5. 基于 @MessageHub 注解的动态实现方案
用 @MessageHub注解来标记消费消息的方法。Spring 启动时，扫描这些注解并生成相应的 Redis 消息监听器。 使用动态代理，将这些方法回调和消息处理结合起来。
步骤：
扫描 Bean 和方法：我们使用 BeanPostProcessor#postProcessAfterInitialization来扫描 Spring 容器中的所有 Bean，寻找那些标注了@MessageHub注解的方法。动态构建监听器：根据方法信息，动态创建一个 MessageListenerAdapter，并将它注册到RedisMessageListenerContainer中。回调方法设置：为了确保监听器能正常工作，我们需要手动调用 afterPropertiesSet()方法来初始化监听器的属性，确保消息能正确被消费。
6. Redis 发布订阅处理器实现
RedisPubSubProcessor。这个处理器主要负责消息的生产和消费。生产消息：通过 stringRedisTemplate.convertAndSend()发送消息到指定的 Redis 频道。消费消息：首先，动态创建 MessageListenerAdapter，然后将其注册到RedisMessageListenerContainer中。每当消息被发布到频道时，MessageListenerAdapter就会触发回调方法，执行我们定义好的消费逻辑。
// 动态创建消息监听器
MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(delegateObject, "processMessage");
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.addMessageListener(listenerAdapter, new ChannelTopic("myTopic"));
MessageListenerAdapter，我们可以让 Redis 消息直接触发到我们定义的方法上。7. 关键技术实现
MessageListenerAdapter 构造器
MessageListenerAdapter 需要两个参数，一个是 delegate 对象，另一个是方法名。这个构造器让我们能够指定哪个方法来处理收到的消息。// 构造一个消息监听器，指定处理方法
MessageListenerAdapter adapter = new MessageListenerAdapter(myBean, "handleMessage");
RedisMessageListenerContainer 的配置
RedisMessageListenerContainer 是 Spring 提供的一个容器，用于管理 Redis 消息的监听和分发。它需要我们将监听器和 Redis 频道绑定。// 配置 Redis 消息监听容器
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.addMessageListener(listenerAdapter, new ChannelTopic("myChannel"));
生命周期管理
afterPropertiesSet() 方法非常有用，它能保证监听器在实际使用之前就已经完成初始化。// 手动初始化监听器的相关属性
listenerAdapter.afterPropertiesSet();
8. 总结与扩展
以上，就是今天的分享了，看完文章记得右下角给何老师点赞，也欢迎在评论区写下你的留言。