⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 juejin.cn/post/7265624177775558716 「在下uptown」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

前言

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

@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容器注入监听器，而且回写非常死板。那么如何将这块的逻辑统一呢。

之前总结过消息组件的代码设计，这里贴一下链接：

“

https://juejin.cn/post/7204113113699729463

常规写法

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

1.创建消息监听器

@Bean 
public MessageListenerAdapter smsExpirationListener(TestSubscriber messageListener) {
    return new MessageListenerAdapter(messageListener, "onMessage");
}

2.创建订阅器

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

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

@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生命周期函数服务启动时建立常驻消费线程的。

/**
 * 生产消息
 */
Integer producer(MessageForm messageForm);

/**
 * 消费消息
 */
void consumer(ConsumerAdapterForm adapterForm);

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

stringRedisTemplate.convertAndSend(messageForm.getTopic(), messageForm.getMessage());

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

/**
 * 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判断是否标注了注解。

MessageHub annotation = AnnotationUtils.findAnnotation(method, MessageHub.class);
if (annotation == null) {
    continue;
}

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

@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 的方法里

芋道源码 —— 纯源码解析博客

SpringBoot+Redis自定义注解实现发布订阅

前言

常规写法

自定义注解实现