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。

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

最新评论