当前位置: 首页>后端>正文

RocketMQ—Producer(二)路由动态更新

一、Producer路由信息

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


RocketMQ—Producer(二)路由动态更新,第1张

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

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 getPublishTopicList
getPublishTopicList 方法分析:


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;
        }
    }

分析

  1. 如果生产者中缓存了 topic 的路由信息,如果该路由信息中包含了消息队列,则直接返回该路由信息,

  2. 如果没有缓存或没有包含消息队列, 则向 NameServer查询该 topic 的路由信息。

  3. 如果最终未找到路由信息,则抛出异常 : 无法找到主题相关路由信息异常 。

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;
}

备注:

  1. 判断:TopicRouteData 是否改变,topicRouteDataIsChange(old, topicRouteData); 代码很简单,直接分析如下:

    (1) 判断 olddata 或 nowdata 是否为空

    (2) TopicRouteData的equals方法比较

  2. 继续判断是否需要更新topic路由信息isNeedUpdateTopicRouteInfo(topic);

    最终调用的代码为:

    DefaultMQProducerImpl#isPublishTopicNeedUpdate(主要逻辑判断是 TopicPublishInfo是否存在,或者 TopicPublishInfo的messageQueueList是否为空)

  3. 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 上海社区”联合创始人,曾就职于拼多多、德邦等公司,现任上市快递公司架构负责人,主要负责开发框架的搭建、中间件相关技术的二次开发和运维管理、混合云及基础服务平台的建设。


https://www.xamrdz.com/backend/3jb1942681.html

相关文章: