快捷導(dǎo)航

RocketMq同組消費(fèi)者如何自動(dòng)設(shè)置InstanceName

更新時(shí)間：2024年06月18日 08:55:46 作者：曾令勝

這篇文章主要介紹了RocketMq同組消費(fèi)者如何自動(dòng)設(shè)置InstanceName問題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

一、背景

同組多于1個(gè)消費(fèi)者，如果沒單獨(dú)設(shè)置instanceName,默認(rèn)為DEFAULT。

啟動(dòng)時(shí)會(huì)報(bào)如下錯(cuò)誤：

org.apache.rocketmq.client.exception.MQClientException: The consumer group[group_03] has been created before, specify another name please.

二、處理方法

創(chuàng)建MqBeanPost，利用后置處理器獲取到想要設(shè)置的bean，把instanceName設(shè)置成隨機(jī)數(shù)。

@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啟動(dòng)流程：

（1）SpringBootApplication

（2）@EnableAutoConfiguration

（3）AutoConfigurationImportSelector實(shí)現(xiàn)了ImportSelector 接口，所以執(zhí)行selectImports方法

->getAutoConfigurationEntry
->getCandidateConfigurations
->SpringFactoriesLoader.loadFactoryNames
->loadSpringFactories此方法會(huì)讀取所有META-INF/spring.factories文件,轉(zhuǎn)成Map<String, List>,最后getOrDefault(factoryTypeName, Collections.emptyList())獲取key 為org.springframework.boot.autoconfigure.EnableAutoConfiguration的值為需要加載到容器類的全類名的集合。

（4）rocketmq和springboot整合jar中spring.factories位置。

（5）RocketMQAutoConfiguration中@import注入ListenerContainerConfiguration。

ListenerContainerConfiguration 實(shí)現(xiàn)了SmartInitializingSingleton類所以當(dāng)spring容器創(chuàng)建ListenerContainerConfiguration是會(huì)進(jìn)入afterSingletonsInstantiated方法。

（6）此方法中，獲取帶有RocketMQMessageListener注解類的集合，遍歷執(zhí)行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）重點(diǎn)分析一下畫紅的代碼。createRocketMQListenerContainer方法是獲取注解中的屬性，創(chuàng)建出DefaultRocketMQListenerContainer對(duì)象。最后注冊(cè)到容器。

(8)DefaultRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName,DefaultRocketMQListenerContainer.class);

spring容器創(chuàng)建DefaultRocketMQListenerContainer對(duì)象。

創(chuàng)建對(duì)象的流程不在贅述（可以翻看我以前博客創(chuàng)建對(duì)象流程）。

主要分析后置處理器。使用后置處理器來(lái)處理instanceName。

（9）invokeInitMethods方法中，執(zhí)行afterPropertiesSet初始化方法。

invokeInitMethods方法執(zhí)行前會(huì)調(diào)用applyBeanPostProcessorsBeforeInitialization，方法執(zhí)行后會(huì)調(diào)用applyBeanPostProcessorsAfterInitialization。

（10）DefaultRocketMQListenerContainer 實(shí)現(xiàn)了InitializingBean所以在執(zhí)行初始化方法時(shí)，調(diào)用afterPropertiesSet，然后繼續(xù)調(diào)用initRocketMQPushConsumer

（11）initRocketMQPushConsumer 方法會(huì)創(chuàng)建DefaultMQPushConsumer對(duì)象，默認(rèn)的instanceName就是在此創(chuàng)建。

所以如果想給DefaultMQPushConsumer設(shè)置instanceName，就可以在applyBeanPostProcessorsAfterInitialization中設(shè)置。

為何不能在applyBeanPostProcessorsBeforeInitialization執(zhí)行的時(shí)候，因?yàn)镈efaultMQPushConsumer還未創(chuàng)建。