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

时间：2025-02-18 09:22:42

原文地址

一、前情

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

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

public class RocketMqConfig {

    @Value("${-server}")
    private String mallServer;
    
    @Value("${}")
    private String producerGroup;

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

    @Bean("mallMQTemplate")
    public RocketMQTemplate mallMQTemplate( ObjectMapper rocketMQMessageObjectMapper) {
        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        (liveMQProducer());
        (rocketMQMessageObjectMapper);
        return rocketMQTemplate;
    }
}

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

二、问题

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

Caused by: : Send [3] times, still failed, cost [14]ms, Topic: SELL_xxx_TOPIC, BrokersSent: [broker-a, broker-a, broker-a]
See /docs/faq/ for further details.
        at (:638)
        at (:1310)
        at (:358)
        at (:188)
        ... 36 common frames omitted
Caused by: : CODE: 1  DESC: : No accessKey is configured, (:371)
For more information, please visit the url, /docs/faq/
        at (:671)
        at (:467)
        at (:449)
        at (:403)
        at (:831)
        at (:557)
        ... 39 common frames omitted

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

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

三、解决

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

  1. ();

  2. (rocketMsg, timeout)；

  3. (msg, timeout)；

  4. (msg, , null, timeout);

  5. (());

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

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

  1. 首先我们一眼就看到 mQClientFactory 是 DefaultMQProducerImpl 的属性。

  2. 类的的依赖关系 RocketMQTemplate -> DefaultMQProducer -> DefaultMQProducerImpl -> mQClientFactory

  3. 在我们进行创建 RocketMQTemplate 的时候，因为其实现了 InitializingBean，所以 afterPropertiesSet 方法会执行.

  4. 这个时候就会调用 ()。在 DefaultMQProducer 内又会调用 ();

  5. 在 DefaultMQProducerImpl start 方法内就会发现 mQClientFactory 的创建过程了。

().getOrCreateMQClientInstance(, rpcHook);

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

 public String buildMQClientId() {
        StringBuilder sb = new StringBuilder();
        (());

        ("@");
        (());
        if (!()) {
            ("@");
            ();
        }

        return ();
    }

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

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

@Configuration
public class RocketMqConfig {

    @Value("${-server}")
    private String mallServer;
    @Value("${}")
    private String producerGroup;
  
    public DefaultMQProducer mallMQProducer() {
        DefaultMQProducer producer;
        producer = new DefaultMQProducer(producerGroup);
        ("mall");
        (mallServer);

        return producer;
    }

    @Bean("mallMQTemplate")
    public RocketMQTemplate mallMQTemplate( ObjectMapper rocketMQMessageObjectMapper) {
        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        (mallMQProducer());
        (rocketMQMessageObjectMapper);
        return rocketMQTemplate;
    }
}

四、总结

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

不光是生产者，消费者也要设置 unitName，不然也会导致只使用同一个集群的问题

相关文章