RocketMq同组消费者如何自动设置InstanceName
作者:曾令胜
一、背景
同组多于1个消费者,如果没单独设置instanceName,默认为DEFAULT。
启动时会报如下错误:
org.apache.rocketmq.client.exception.MQClientException: The consumer group[group_03] has been created before, specify another name please.
二、处理方法
创建MqBeanPost,利用后置处理器获取到想要设置的bean,把instanceName设置成随机数。
@Component public class MqBeanPost implements BeanPostProcessor { @Autowired MqJudgePacsConfig mqJudgePacsConfig; @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { if(bean instanceof DefaultRocketMQListenerContainer){ DefaultRocketMQListenerContainer container = (DefaultRocketMQListenerContainer) bean; String topic = container.getTopic(); if(topic.equals(mqJudgePacsConfig.getTopic())){ DefaultMQPushConsumer consumer = container.getConsumer(); consumer.setInstanceName(UUID.fastUUID().toString()); } } return bean; } }
三、源码分析
一、springboot整合rocketmq启动流程:
(1)SpringBootApplication
(2)@EnableAutoConfiguration
(3)AutoConfigurationImportSelector实现了ImportSelector 接口,所以执行selectImports方法
- ->getAutoConfigurationEntry
- ->getCandidateConfigurations
- ->SpringFactoriesLoader.loadFactoryNames
- ->loadSpringFactories此方法会读取所有META-INF/spring.factories文件,转成Map<String, List>,最后getOrDefault(factoryTypeName, Collections.emptyList())获取key 为org.springframework.boot.autoconfigure.EnableAutoConfiguration的值为需要加载到容器类的全类名的集合。
(4)rocketmq和springboot整合jar中spring.factories位置。
(5)RocketMQAutoConfiguration中@import注入ListenerContainerConfiguration。
ListenerContainerConfiguration 实现了SmartInitializingSingleton类所以当spring容器创建ListenerContainerConfiguration是会进入afterSingletonsInstantiated方法。
(6)此方法中,获取带有RocketMQMessageListener注解类的集合,遍历执行registerContainer。
public void afterSingletonsInstantiated() { Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class) .entrySet().stream().filter(entry -> !ScopedProxyUtils.isScopedTarget(entry.getKey())) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); beans.forEach(this::registerContainer); }
(7)重点分析一下画红的代码。createRocketMQListenerContainer方法是获取注解中的属性,创建出DefaultRocketMQListenerContainer对象。最后注册到容器。
(8)DefaultRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName,DefaultRocketMQListenerContainer.class);
spring容器创建DefaultRocketMQListenerContainer对象。
创建对象的流程不在赘述(可以翻看我以前博客创建对象流程)。
主要分析后置处理器。使用后置处理器来处理instanceName。
(9)invokeInitMethods方法中,执行afterPropertiesSet初始化方法。
invokeInitMethods方法执行前会调用applyBeanPostProcessorsBeforeInitialization,方法执行后会调用applyBeanPostProcessorsAfterInitialization。
(10)DefaultRocketMQListenerContainer 实现了InitializingBean所以在执行初始化方法时,调用afterPropertiesSet,然后继续调用initRocketMQPushConsumer
(11)initRocketMQPushConsumer 方法会创建DefaultMQPushConsumer对象,默认的instanceName就是在此创建。
所以如果想给DefaultMQPushConsumer设置instanceName,就可以在applyBeanPostProcessorsAfterInitialization中设置。
为何不能在applyBeanPostProcessorsBeforeInitialization执行的时候,因为DefaultMQPushConsumer还未创建。
(12)拓展:initRocketMQPushConsumer 中画红的地方。
如果消费端实现了RocketMQPushConsumerLifecycleListener或RocketMQPushConsumerLifecycleListener类的话,可以重写prepareStart方法。
在prepareStart方法中设置instanceName。
但是这种方法如果有多个消费端的话,要写多次。
四、总结
此方法的切入点是DefaultRocketMQListenerContainer类创建过程中,使用后置处理器设置instanceName。
以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。