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集群。

三、解决

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

  1. rocketMQTemplate.syncSend();

  2. producer.send(rocketMsg, timeout);

  3. this.defaultMQProducerImpl.send(msg, timeout);

  4. this.sendDefaultImpl(msg, CommunicationMode.SYNC, null, timeout);

  5. this.tryToFindTopicPublishInfo(msg.getTopic());

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

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

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

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

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

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

  5. 在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使用错乱。

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

本文使用 mdnice
排版

原文 

https://juejin.im/post/5ef9c650f265da22ea0a1c6d

本站部分文章源于互联网,本着传播知识、有益学习和研究的目的进行的转载,为网友免费提供。如有著作权人或出版方提出异议,本站将立即删除。如果您对文章转载有任何疑问请告之我们,以便我们及时纠正。

PS:推荐一个微信公众号: askHarries 或者qq群:474807195,里面会分享一些资深架构师录制的视频录像:有Spring,MyBatis,Netty源码分析,高并发、高性能、分布式、微服务架构的原理,JVM性能优化这些成为架构师必备的知识体系。还能领取免费的学习资源,目前受益良多

转载请注明原文出处:Harries Blog™ » Spring Boot RocketMQ 多集群客户端使用小坑记录

赞 (0)
分享到:更多 ()

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址