springboot+redis自定义注解实现发布订阅

布满荆棘的人生 2023-10-14 18:26 46阅读 0赞

## 前言 ##

最近开发了一个内部消息组件，逻辑大体是通过定义注解 **@MessageHub**，在启动时扫描全部bean中有使用了该注解的方法后台创建一个常驻线程代理消费数据，当线程消费到数据就回写到对应加了注解的方法里。

less复制代码@Slf4j
    @Service
    public class RedisConsumerDemo {
        @MessageHub(topic = "${uptown.topic}", type = "REDIS_PUBSUB")
        public void consumer(Object message) {
            log.info("pubsub info {} ", message);
        }   
    }

实现redis的队列、stream方式实现都很简单，唯独发布订阅方式，网上的demo全都是一个固定套路，通过redis容器注入监听器，而且回写非常死板。

**常规写法**

常规实现reids的发布订阅模式写法一共三步

1.  创建消息监听器

typescript复制代码@Bean 
    public MessageListenerAdapter smsExpirationListener(TestSubscriber messageListener) {
        return new MessageListenerAdapter(messageListener, "onMessage");
    }

1.  创建订阅器

java复制代码
    @Component
    public class TestSubscriber implements MessageListener {
     
        @Override
        public void onMessage(Message message, byte[] pattern) {
            log.info("get data :{}", msg);
        }
     
    }

1.  向redis容器中添加**消息监听器**

java复制代码@Configuration
    public class RedisConfig {
    
        @Bean
        public RedisMessageListenerContainer container(
            RedisConnectionFactory redisConnectionFactory,
            MessageListenerAdapter smsExpirationListener) {
        
            RedisMessageListenerContainer container = new RedisMessageListenerContainer();
            container.setConnectionFactory(redisConnectionFactory);
            container.addMessageListener(smsExpirationListener, new PatternTopic("test"));
            return container;
        }
    }

这样定义非常简单明了，但是有个问题是太代码僵硬了，创建监听者很不灵活，只能指定内部的onMessage方法，那么怎么才能融入到我们的内部消息流转中间件里呢。

## 自定义注解实现 ##

我们内部组件抽象了两个方法，生产和消费，但这两个方法逻辑截然不同，**生产方法是暴露给serverice层接口调用**，调用方在调用生产方法后能直接知道生产了几条数据和成功与否。而消费方法是配合Spring生命周期函数服务启动时建立常驻消费线程的。

scss复制代码
    /**
     * 生产消息
     */
    Integer producer(MessageForm messageForm);
    
    /**
     * 消费消息
     */
    void consumer(ConsumerAdapterForm adapterForm);

生产消息当然很容易实现，只需要调用已经封装好的convertAndSend方法。

less复制代码stringRedisTemplate.convertAndSend(messageForm.getTopic(), messageForm.getMessage());

消费方法就有说法了，动态生成监听者的场景下使用redis容器用代码挨个注册已经满足不了了，但仔细过一遍源代码就会发现，监听类的构造方法的入参只有两个，第一个需要回调的代理类，第二个消费到数据后回调的方法。

vbnet复制代码/**
     * Create a new {@link MessageListenerAdapter} for the given delegate.
     *
     * @param delegate the delegate object
     * @param defaultListenerMethod method to call when a message comes
     * @see #getListenerMethodName
     */
    public MessageListenerAdapter(Object delegate, String defaultListenerMethod) {
       this(delegate);
       setDefaultListenerMethod(defaultListenerMethod);
    }

那么好了好了，方案有了，本质上就是把**RedisMessageListenerContainer**注入进来之后，扫描项目里所有加了 **@MessageHub** 的bean，包装成监听类加载到容器里就完事了。怎么扫描的代码就不再赘述了，实现Spring的生命周期函数**BeanPostProcessor\#postProcessAfterInitialization**，在这里用AnnotationUtils判断是否标注了注解。

ini复制代码MessageHub annotation = AnnotationUtils.findAnnotation(method, MessageHub.class);
    if (annotation == null) {
        continue;
    }

标注了后判断如果是发布订阅，进入发布订阅的实现类。

java复制代码@Scope(proxyMode = ScopedProxyMode.TARGET_CLASS)
    @Service("redisPubSubProcessor")
    public class RedisPubSubProcessor extends MessageHubServiceImpl {
    
        @Resource
        RedisMessageListenerContainer redisPubSubContainer;
    
    
    
        @Override
        public void produce(ProducerAdapterForm producerAdapterForm) {
            stringRedisTemplate.convertAndSend(producerAdapterForm.getTopic(), producerAdapterForm.getMessage());
        }
    
        @Override
        public void consume(ConsumerAdapterForm messageForm) {
            MessageListenerAdapter adapter = new MessageListenerAdapter(messageForm.getBean(), messageForm.getInvokeMethod().getName());
            adapter.afterPropertiesSet();
            redisPubSubContainer.addMessageListener(adapter, new PatternTopic(messageForm.getTopic()));
        }
    
    
        @Bean
        public RedisMessageListenerContainer redisPubSubContainer(RedisConnectionFactory connectionFactory) {
            RedisMessageListenerContainer container = new RedisMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            return container;
        }
    }

首先先将RedisMessageListenerContainer注入到Spring容器里，produce方法只需要调用下现程的api。consume方法由于上一步我们获取了bean和对应的method，直接用**MessageListenerAdapter**的构造器创建出监听器来，这里有个坑，需要手动调用adapter.afterPropertiesSet()设置一些必要的属性，这个在常规写法里框架帮忙做了。如果不调用的话会出一些空指针之类的bug。

随后把监听器add到容器就实现了方法代理，背后的线程监听到数据会回调到标注了 **@MessageHub** 的方法里