赞
踩
最近开发了一个内部消息组件,逻辑大体是通过定义注解 @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的发布订阅模式写法一共三步
- typescript复制代码@Bean
- public MessageListenerAdapter smsExpirationListener(TestSubscriber messageListener) {
- return new MessageListenerAdapter(messageListener, "onMessage");
- }
- java复制代码
- @Component
- public class TestSubscriber implements MessageListener {
-
- @Override
- public void onMessage(Message message, byte[] pattern) {
- log.info("get data :{}", msg);
- }
-
- }
-
- 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 的方法里
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。