转载

Spring Boot RocketMQ 多集群客户端使用小坑记录

一、前情

今儿听说业务小伙伴需要在项目中使用多个RocketMQ集群，当前业务有一个集群做canal消费使用（此MQ集群开启了ACL），需要在增加一个MQ集群做业务数据发送，项目使用了Spring Boot组件。

好了，问题描述完了，概括下，就是当前有个MQ集群在进行数据消费，需要在像另一个MQ集群发送数据。整明白需求，搞起来，这不是分分钟的事儿吗，嗖嗖嗖，我就写了下面的Config。

public class RocketMqConfig {

    @Value("${rocketmq.mall.name-server}")
    private String mallServer;
    
    @Value("${rocketmq.mall.producer.group}")
    private String producerGroup;

    public DefaultMQProducer liveMQProducer() {
        DefaultMQProducer producer;
        producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(mallServer);
        return producer;
    }

    @Bean("mallMQTemplate")
    public RocketMQTemplate mallMQTemplate( ObjectMapper rocketMQMessageObjectMapper) {
        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        rocketMQTemplate.setProducer(liveMQProducer());
        rocketMQTemplate.setObjectMapper(rocketMQMessageObjectMapper);
        return rocketMQTemplate;
    }
}
复制代码

看看，分分钟搞定，使用的时候直接注入mallMQTemplate就可以了，交付完成后我就飘走了。

二、问题

然而，天有不测风云，业务小伙伴紧急来电，测试环境报错了，这玩意不好使啊，WTF？不能够啊。赶紧跑过去看了下异常。。。

Caused by: org.apache.rocketmq.client.exception.MQClientException: Send [3] times, still failed, cost [14]ms, Topic: SELL_xxx_TOPIC, BrokersSent: [broker-a, broker-a, broker-a]
See http://rocketmq.apache.org/docs/faq/ for further details.
        at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:638)
        at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1310)
        at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:358)
        at org.apache.rocketmq.spring.core.RocketMQTemplate.syncSend(RocketMQTemplate.java:188)
        ... 36 common frames omitted
Caused by: org.apache.rocketmq.client.exception.MQBrokerException: CODE: 1  DESC: org.apache.rocketmq.acl.common.AclException: No accessKey is configured, org.apache.rocketmq.acl.plain.PlainPermissionManager.validate(PlainPermissionManager.java:371)
For more information, please visit the url, http://rocketmq.apache.org/docs/faq/
        at org.apache.rocketmq.client.impl.MQClientAPIImpl.processSendResponse(MQClientAPIImpl.java:671)
        at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessageSync(MQClientAPIImpl.java:467)
        at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessage(MQClientAPIImpl.java:449)
        at org.apache.rocketmq.client.impl.MQClientAPIImpl.sendMessage(MQClientAPIImpl.java:403)
        at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendKernelImpl(DefaultMQProducerImpl.java:831)
        at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:557)
        ... 39 common frames omitted

复制代码

关键点 No accessKey is configured，嗯？这玩意我业务集群没开ACL啊，设置个毛线。但冥冥中感觉那里少配置啥了，但开发环境又没有问题。

经过我这大脑一顿分析和测试，发现这发送的消费根本就没到达测试环境的MQ业务集群（这里有个自身问题就是我们测试环境业务和canal MQ集群是分开的，开发是在一起的）。马上切换到开发环境测试一把，发现不管怎么配置最后都会发送到canal集群。

三、解决

复现了问题，那就来解决吧，翻了翻源代码进行查看消息发送流程，把关键点标注下。

rocketMQTemplate.syncSend();
producer.send(rocketMsg, timeout)；
this.defaultMQProducerImpl.send(msg, timeout)；
this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);
this.tryToFindTopicPublishInfo(msg.getTopic());

当调用到第5步的时候，问题出现了，这货返回的根本就不是我配置的MQ业务集群，而是canal的分区信息。呵呵，麻麦皮。进入 tryToFindTopicPublishInfo 方法，看了下关键点在于mQClientFactory 这个对象，居然是canal创建的对象，而不是我业务集群创建的对象。

所以，问题就在于mQClientFactory，那就来看下这货是怎么创建的就可以了。

首先我们一眼就看到 mQClientFactory 是DefaultMQProducerImpl的属性。
类的的依赖关系 RocketMQTemplate -> DefaultMQProducer -> DefaultMQProducerImpl -> mQClientFactory
在我们进行创建RocketMQTemplate的时候，因为其实现了InitializingBean，所以afterPropertiesSet方法会执行.
这个时候就会调用DefaultMQProducer.start()。在DefaultMQProducer内又会调用 DefaultMQProducerImpl.start();
在DefaultMQProducerImpl start方法内就会发现mQClientFactory 的创建过程了。

MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
复制代码

通过 getOrCreateMQClientInstance(final ClientConfig clientConfig, RPCHook rpcHook);方法得知，这货搞了个单例把我们DefaultMQProducer都给缓存起来了。而其中关键代码如下：

 public String buildMQClientId() {
        StringBuilder sb = new StringBuilder();
        sb.append(this.getClientIP());

        sb.append("@");
        sb.append(this.getInstanceName());
        if (!UtilAll.isBlank(this.unitName)) {
            sb.append("@");
            sb.append(this.unitName);
        }

        return sb.toString();
    }
复制代码

这就是获取key的方式，就是我们的IP加上ClientConfig 的属性unitName得到的。所以如果我们没有设置unitName，就算你再怎么创建DefaultMQProducer，都只会获得相同的一个。

所以，最后只需要加上一行代码 producer.setUnitName("mall")，就完美解决了这个问题，完整如下：


@Configuration
public class RocketMqConfig {

    @Value("${rocketmq.mall.name-server}")
    private String mallServer;
    @Value("${rocketmq.mall.producer.group}")
    private String producerGroup;
  
    public DefaultMQProducer mallMQProducer() {
        DefaultMQProducer producer;
        producer = new DefaultMQProducer(producerGroup);
        producer.setUnitName("mall");
        producer.setNamesrvAddr(mallServer);

        return producer;
    }

    @Bean("mallMQTemplate")
    public RocketMQTemplate mallMQTemplate( ObjectMapper rocketMQMessageObjectMapper) {
        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        rocketMQTemplate.setProducer(mallMQProducer());
        rocketMQTemplate.setObjectMapper(rocketMQMessageObjectMapper);
        return rocketMQTemplate;
    }
}
复制代码

四、总结

在使用Spring Boot RocketMQTemplate 多集群发送消息时，因为DefaultMQProducerImpl内部会通过MQClientManager维护一个defaultMQProducer的缓存，而key是IP加unitName拼接的，所以一定要设置unitName，防止defaultMQProducer使用错乱。