RocketMQ—Producer（二）路由动态更新

2022 年 4 月 04 日
本文字数：6575 字
阅读完需：约 22 分钟

一、Producer 路由信息

从 NameServer 章节分析得知，路由信息存储在 NameServer，生产端和消费端定时向 NameServer 获取 topic 相关的路由信息；从生产者启动流程得知：路由信息的动态更新源码在 MQClientInstance#startScheduledTask 定时任务里面具体方法：updateTopicRouteInfoFromNameServer 下图为路由更新流程

接下来我们着重解析此段源码:

1 定时任务：频率-30s


this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {  @Override  public void run() {    try {        //从nameServer更新路由信息 -定时任务：30s一次        MQClientInstance.this.updateTopicRouteInfoFromNameServer();    } catch (Exception e) {        log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);    }  }}, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);

复制代码

2 updateTopicRouteInfoFromNameServer


public void updateTopicRouteInfoFromNameServer() {    Set<String> topicList = new HashSet<String>();    {   // Consumer 消费端，后续再分析      ...省略....    }    {   // Producer 生产端        Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();        while (it.hasNext()) {            Entry<String, MQProducerInner> entry = it.next();            MQProducerInner impl = entry.getValue();            if (impl != null) {                Set<String> lst = impl.getPublishTopicList();//1>获取所有 topic-list                topicList.addAll(lst);            }        }    }    //2>更新路由信息    for (String topic : topicList) {        this.updateTopicRouteInfoFromNameServer(topic);    }}

复制代码

分析如下:

1.1 getPublishTopicListgetPublishTopicList 方法分析：


public Set<String> getPublishTopicList() {    Set<String> topicList = new HashSet<String>();    for (String key : this.topicPublishInfoTable.keySet()) {        topicList.add(key);    }    return topicList;}

复制代码

备注：

细心的你可能发现从启动流程中得知：

topicPublishInfoTable(ConcurrentHashMap)只会默认注册 topic=TBW102 的信息,那正常业务发送的 topic 是如何注册进去的呢，建议直接观看理解以下代码，在发送流程中会体现出如何注册到 topicPublishInfoTable 中；topicPublishInfoTable 数据的初始化（value:第一次默认都是 new TopicPublishInfo()）

 //查找主题的路由信息的方法    private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {        TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);        if (null == topicPublishInfo || !topicPublishInfo.ok()) {            this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());            this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic); //从NameServer更新topic路由信息            topicPublishInfo = this.topicPublishInfoTable.get(topic);        }
        if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {            return topicPublishInfo;        } else {            this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer); //从NameServer更新topic路由信息            topicPublishInfo = this.topicPublishInfoTable.get(topic);            return topicPublishInfo;        }    }

复制代码

分析

如果生产者中缓存了 topic 的路由信息，如果该路由信息中包含了消息队列，则直接返回该路由信息，
如果没有缓存或没有包含消息队列，则向 NameServer 查询该 topic 的路由信息。
如果最终未找到路由信息，则抛出异常 : 无法找到主题相关路由信息异常。

1.2 updateTopicRouteInfoFromNameServer 从 NameServer 更新 topic 路由信息

在分析之前，可先简单分析 MQClientInstance 核心属性：


public class MQClientInstance {    ...省略...    //key:group,    value: 生产者    private final ConcurrentMap<String/* group */, MQProducerInner> producerTable = new ConcurrentHashMap<String, MQProducerInner>();    ...省略..   //topic-路由信息    private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<String, TopicRouteData>();    private final Lock lockNamesrv = new ReentrantLock();   //更新路由使用    private final Lock lockHeartbeat = new ReentrantLock(); //发送心跳使用    //broker信息，key:Broker Name, value:-key:brokerId,value:address    private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable =        new ConcurrentHashMap<String, HashMap<Long, String>>();    //broker版本信息，key:Broker Name, value:-key:address,value:version    private final ConcurrentMap<String/* Broker Name */, HashMap<String/* address */, Integer>> brokerVersionTable =        new ConcurrentHashMap<String, HashMap<String, Integer>>();    ...省略...}

复制代码

备注：此处列出的属性仅跟生产端相关，其他的属性和方法大都我们会在消费端分析

接下来着重分析：updateTopicRouteInfoFromNameServer


/** * 向-NameServer查询该 topic 的路由信息 * @param topic 主题 * @param isDefault 是否默认主题 * @param defaultMQProducer 默认MQProducer * @return */public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,    DefaultMQProducer defaultMQProducer) {    try {        if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) { //获取锁:3s            try {                TopicRouteData topicRouteData;                if (isDefault && defaultMQProducer != null) { //默认主题-'TBW102'，从NameServer查询-topicRouteData                    topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(), 1000 * 3);                    if (topicRouteData != null) {                        for (QueueData data : topicRouteData.getQueueDatas()) { //读写队列取最小值，getDefaultTopicQueueNums=4，getReadQueueNums=16                            int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());                            data.setReadQueueNums(queueNums);                            data.setWriteQueueNums(queueNums);                        }                    }                } else {//非默认主题，从NameServer查询-topicRouteData                    topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);                }                if (topicRouteData != null) {                    TopicRouteData old = this.topicRouteTable.get(topic);                    boolean changed = topicRouteDataIsChange(old, topicRouteData);//1> 判断：TopicRouteData 是否改变                    if (!changed) { //未改变，                        changed = this.isNeedUpdateTopicRouteInfo(topic);//2>继续判断是否需要更新：topic-路由信息                    } else {                        log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);                    }
                    if (changed) { // 需要更新                        TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData();
                        for (BrokerData bd : topicRouteData.getBrokerDatas()) {                            this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());//维护brokerAddrTable地址信息                        }
                        // Update Pub info                        {   //topicRouteData 转换 TopicPublishInfo(isWriteable)-生产需要的数据                            TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData); // 3>数据转换                            publishInfo.setHaveTopicRouterInfo(true);                            Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();//迭代-producerTable                            while (it.hasNext()) {                                Entry<String, MQProducerInner> entry = it.next();                                MQProducerInner impl = entry.getValue();                                if (impl != null) {                                    impl.updateTopicPublishInfo(topic, publishInfo);//4-更新-路由发布信息                                }                            }                        }
                        // Update sub info                        { //消费端--后续消费端讲解（ topicRouteData 转换 TopicPublishInfo(isReadable) 队列消息）                            Set<MessageQueue> subscribeInfo = topicRouteData2TopicSubscribeInfo(topic, topicRouteData);//构建队列信息                            Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();                            while (it.hasNext()) {                                Entry<String, MQConsumerInner> entry = it.next();                                MQConsumerInner impl = entry.getValue();                                if (impl != null) {                                    impl.updateTopicSubscribeInfo(topic, subscribeInfo); //更新-消费端：队列信息                                }                            }                        }                        log.info("topicRouteTable.put. Topic = {}, TopicRouteData[{}]", topic, cloneTopicRouteData);                        this.topicRouteTable.put(topic, cloneTopicRouteData);  // 维护：topicRouteTable                        return true;                    }                } else {                    log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}", topic);                }            } catch (Exception e) { //异常吃掉了                if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) && !topic.equals(MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC)) {                    log.warn("updateTopicRouteInfoFromNameServer Exception", e);//topic 非 RETRY 和 TBW102 失                }            } finally {                this.lockNamesrv.unlock();  // 释放锁            }        } else {            log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms", LOCK_TIMEOUT_MILLIS);        }    } catch (InterruptedException e) {        log.warn("updateTopicRouteInfoFromNameServer Exception", e);    }    return false;}

复制代码

备注：

判断：TopicRouteData 是否改变，topicRouteDataIsChange(old, topicRouteData); 代码很简单，直接分析如下：
(1) 判断 olddata 或 nowdata 是否为空
(2) TopicRouteData 的 equals 方法比较
继续判断是否需要更新 topic 路由信息 isNeedUpdateTopicRouteInfo(topic);
最终调用的代码为：
DefaultMQProducerImpl#isPublishTopicNeedUpdate(主要逻辑判断是 TopicPublishInfo 是否存在，或者 TopicPublishInfo 的 messageQueueList 是否为空)
topicRouteData2TopicPublishInfo 数据转换，你一定感兴趣，内容相当简单，

分析如下：

public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route) {    TopicPublishInfo info = new TopicPublishInfo();    info.setTopicRouteData(route);    if (route.getOrderTopicConf() != null && route.getOrderTopicConf().length() > 0) {  // 此处可忽略，针对顺序消息        String[] brokers = route.getOrderTopicConf().split(";");        for (String broker : brokers) {            String[] item = broker.split(":");            int nums = Integer.parseInt(item[1]);            for (int i = 0; i < nums; i++) {                MessageQueue mq = new MessageQueue(topic, item[0], i);                info.getMessageQueueList().add(mq);            }        }        info.setOrderTopic(true);    } else {        List<QueueData> qds = route.getQueueDatas();        Collections.sort(qds);        for (QueueData qd : qds) {            if (PermName.isWriteable(qd.getPerm())) { // 写权限判断                BrokerData brokerData = null;                for (BrokerData bd : route.getBrokerDatas()) {                    if (bd.getBrokerName().equals(qd.getBrokerName())) {                        brokerData = bd;                        break;                    }                }                if (null == brokerData) {                    continue;                }                if (!brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) { // 不包含masterId，过滤                    continue;                }                for (int i = 0; i < qd.getWriteQueueNums(); i++) {                    MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i);                    info.getMessageQueueList().add(mq); //根据写队列个数，根据 topic+序号创建 MessageQueue，填充topicPublishlnfo 的 List<QuueMessage>。完成消息发送的路由查找 。                }            }        }        info.setOrderTopic(false);    }    return info;}

复制代码

4.更新-路由发布信息:updateTopicPublishInfo(topic, publishInfo);调用的代码为：DefaultMQProducerImpl#updateTopicPublishInfo，本质就是维护 Map-topicPublishInfoTable

二、结论

路由更新虽然相对简单，但对于生产者来说至关重要，生产端需要知道路由信息才能进行计算选择将消息发送到哪台 broker；但从源码分析中，可以看出更新路由信息以 topic 为维度，组装更新数据，本质还是维护 Map(topicRouteTable、brokerAddrTable、topicPublishInfoTable)等，但是要注意是：ConcurrentHashMap。

程序员的核心竞争力其实还是技术，因此对技术还是要不断的学习，关注 “IT 巅峰技术” 公众号 ，该公众号内容定位：中高级开发、架构师、中层管理人员等中高端岗位服务的，除了技术交流外还有很多架构思想和实战案例，作者是《消息中间件 RocketMQ 技术内幕》一书作者，同时也是 “RocketMQ 上海社区”联合创始人，曾就职于拼多多、德邦等公司，现任上市快递公司架构负责人，主要负责开发框架的搭建、中间件相关技术的二次开发和运维管理、混合云及基础服务平台的建设。

发布于: 刚刚阅读数: 2

原文链接:【http://xie.infoq.cn/article/b44f73c0a528ba3f7b1e1e2cc】。

IT巅峰技术

关注

一线架构师、二线开发、三线管理 2021.12.07 加入

Redis6.X、ES7.X、Kafka3.X、RocketMQ5.0、Flink1.X、ClickHouse20.X、SpringCloud、Netty5等热门技术分享；架构设计方法论与实践；作者热销新书《RocketMQ技术内幕》；

发布

暂无评论

创作场景

RocketMQ—Producer（二）路由动态更新

一、Producer 路由信息

1 定时任务：频率-30s

2 updateTopicRouteInfoFromNameServer

二、结论

IT巅峰技术

评论