RocketMQ：深入了解路由中心（NameServer）

一、架構設計

我們知道，消息中間件的設計思路一般基于主題的訂閱釋出機制，消息生産者（Produer）發送某一主題的消息到伺服器，消息伺服器負責該消息的持久化存儲，消息消費者（Consumer）訂閱感興趣的主題，消息伺服器根據訂閱消息（路由消息）将消息推送到消費者（Push模式）或者消息消費者主動向消息伺服器拉取消息（Pull模式），進而實作消息生産者與消息消費者解耦。

那麼，

消息生産者如何知道消息要發送哪台消息伺服器呢？
消息消費者如何知道要從哪台消息伺服器消費消費呢？
若某一太消息伺服器當機了，生産者/消費者如何在不重新開機伺服器的情況下感覺呢？

NameServer就是為解決這些問題而設計的，看RocketMQ的邏輯部署圖：

RocketMQ：深入了解路由中心（NameServer）

NameServer的設計追求簡單高效，摒棄了業界常使用的Zookeeper充當資訊管理的“注冊中心”。

從實際需求出發，自身實作中繼資料的路由管理（Topic路由資訊等）。

Broker在啟動時，向所有NameServer注冊，生産者/消費者在發送/消費消息時，先從NameServer擷取Broker伺服器位址清單。

NameServer與Broker保持長連接配接，并間隔30s檢測Broker是否存活，若檢測到Broker當機，則從路由系統資料庫中将其移除，且路由變化不會馬上通知消息生産者。

NameServer彼此之間互不通信，Topic路由資訊無需在叢集之間保持強一緻性，追求最終一緻性，并且能容忍分鐘級的不一緻。

二、源碼實作

1. NameServer啟動流程

從源碼的角度窺探一下 Names巳rver 啟動流程，重點關注 NameServer 相關啟動參數。

NameServer 啟動類： org.apache.rocketmq.namesrv.NamesrvStartup。

1) 首先解析配置檔案，需要填充NameServerConfig、NettyServerConfig屬性值。

org.apache.rocketmq.namesrv.NamesrvStartup#createNamesrvController代碼片段

//建立NameServer的業務參數配置
final NamesrvConfig namesrvConfig = new NamesrvConfig();
 
//建立NameServer的網絡參數配置
final NettyServerConfig nettyServerConfig = new NettyServerConfig(); 
nettyServerConfig.setListenPort(9876); //網絡監聽端口
if (commandLine.hasOption('c')) { //有指定配置檔案路徑
    String file = commandLine.getOptionValue('c');
    if (file != null) {
        InputStream in = new BufferedInputStream(new FileInputStream(file));
        properties = new Properties();
        //加載配置參數檔案
        properties.load(in);
        MixAll.properties2Object(properties, namesrvConfig);
        MixAll.properties2Object(properties, nettyServerConfig);

        namesrvConfig.setConfigStorePath(file);

        System.out.printf("load config properties file OK, %s%n", file);
        in.close();
    }
}

if (commandLine.hasOption('p')) {
    InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
    MixAll.printObjectProperties(console, namesrvConfig);
    MixAll.printObjectProperties(console, nettyServerConfig);
    System.exit(0);
}

MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);

從代碼我們可以知道先建立NameServerConfig（NameServer業務參數）、NettyServerConfig（NameServer網絡參數），然後解析啟動時把指定的配置檔案或啟動指令中的選項值，填充到nameServerConfig、nettyServerConfig對象。參數來源有如下兩種方式：

-c configFile 通過 -c 指令指定配置檔案的路徑；
使用“–屬性名屬性值”，例如 --listPort 9876；

2) 根據啟動屬性建立NameServer核心控制器執行個體，并初始化

org.apache.rocketmq.namesrv.NamesrvController#initialize代碼片段

//加載KV配置
this.kvConfigManager.load();
//建立NettyServer網絡處理對象
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);

this.remotingExecutor =
    Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
//注冊NameServer接收、響應請求的處理器
this.registerProcessor();
//啟動定時任務，掃描、移除失效的Broker（10s/次）
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

    @Override
    public void run() {
        NamesrvController.this.routeInfoManager.scanNotActiveBroker();
    }
}, 5, 10, TimeUnit.SECONDS);
//啟動定時任務，列印KV配置資訊（10m/次）
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

    @Override
    public void run() {
        NamesrvController.this.kvConfigManager.printAllPeriodically();
    }
}, 1, 10, TimeUnit.MINUTES);

加載KV配置，建立NettyServer網絡處理對象，然後開啟兩個定時任務，在RocketMQ中此類定時任務統稱為心跳檢測。

定時任務1：NameServer每個10s掃描一次Broker，移除處于不激活狀态的Broker；
定時任務2：NameServer每個10分鐘列印一次KV配置；

2. NameServer核心業務邏輯處理

當NameServer的網絡通信服務子產品收到請求後，就調用這個 Processor 來處理。

org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#processRequest代碼片段

public RemotingCommand processRequest(ChannelHandlerContext ctx,
    RemotingCommand request) throws RemotingCommandException {

    if (ctx != null) {
        log.debug("receive request, {} {} {}",
            request.getCode(),
            RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
            request);
    }

    switch (request.getCode()) {
        case RequestCode.PUT_KV_CONFIG:
            return this.putKVConfig(ctx, request);
        case RequestCode.GET_KV_CONFIG:
            return this.getKVConfig(ctx, request);
        case RequestCode.DELETE_KV_CONFIG:
            return this.deleteKVConfig(ctx, request);
        case RequestCode.QUERY_DATA_VERSION:
            return queryBrokerTopicConfig(ctx, request);
        case RequestCode.REGISTER_BROKER: //broker注冊請求
            Version brokerVersion = MQVersion.value2Version(request.getVersion());
            if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
                return this.registerBrokerWithFilterServer(ctx, request);
            } else {
                return this.registerBroker(ctx, request);
            }
        case RequestCode.UNREGISTER_BROKER:
            return this.unregisterBroker(ctx, request);
        case RequestCode.GET_ROUTEINTO_BY_TOPIC: //用戶端路由發現
            return this.getRouteInfoByTopic(ctx, request);
        case RequestCode.GET_BROKER_CLUSTER_INFO:
            return this.getBrokerClusterInfo(ctx, request);
        case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
            return this.wipeWritePermOfBroker(ctx, request);
        case RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER:
            return getAllTopicListFromNameserver(ctx, request);
        case RequestCode.DELETE_TOPIC_IN_NAMESRV:
            return deleteTopicInNamesrv(ctx, request);
        case RequestCode.GET_KVLIST_BY_NAMESPACE:
            return this.getKVListByNamespace(ctx, request);
        case RequestCode.GET_TOPICS_BY_CLUSTER:
            return this.getTopicsByCluster(ctx, request);
        case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS:
            return this.getSystemTopicListFromNs(ctx, request);
        case RequestCode.GET_UNIT_TOPIC_LIST:
            return this.getUnitTopicList(ctx, request);
        case RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST:
            return this.getHasUnitSubTopicList(ctx, request);
        case RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:
            return this.getHasUnitSubUnUnitTopicList(ctx, request);
        case RequestCode.UPDATE_NAMESRV_CONFIG:
            return this.updateConfig(ctx, request);
        case RequestCode.GET_NAMESRV_CONFIG:
            return this.getConfig(ctx, request);
        default:
            break;
    }
    return null;
}

3. NameServer路由注冊、故障剔除

我們知道，NameServer的設計目的主要是提供Broker注冊，為消息生産者/消費者提供關于Topic的路由資訊。

那麼NameServer需要存儲路由的基礎資訊，還要能夠管理Broker節點，包括路由注冊、路由剔除等功能。

1) 路由元資訊

在了解路由注冊之前，我們首先看一下 NameServer 到底存儲哪些資訊。

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager代碼片段

private final HashMap<String/* topic */, List<QueueData>> topicQueueTable; //topic消息隊列路由
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable; //broker基礎資訊
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable; //broker叢集資訊
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable; //broker狀态資訊
private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable; //broker，FilterServer清單

topicQueueTable: Topic 消息隊列路由資訊，消息發送時根據路由表進行負載均衡。
brokerAddrTable : Broker 基礎資訊，包含 brokerName 、所屬叢集名稱、主備 Broker位址。
clusterAddrTable: Broker 叢集資訊，存儲叢集中所有 Broker 名稱。
brokerLiveTable: Broker 狀态資訊，NameServer 每次收到心跳包時會替換該資訊。
filterServerTable : Broker 上的 FilterServer 清單。

2) 路由注冊處理

RocketMQ路由注冊是通過Broker與NameServer的心跳功能實作的。

Broker啟動時，向叢集中所有NameServer發送心跳，每個30s向叢集中所有NameServer發送心跳包，NameServer收到Broker心跳包時會更新brokerLiveTable緩存中BrokerLiveInfo的lastUpdateTimestamp，然後NameServer每隔10s掃描brokerLiveTable，如果連續120s沒有收到心跳包，NameServer将剔除該Broker的路由資訊同時關閉Socket連接配接。

org.apache.rocketmq.broker.BrokerController#start代碼片段

//先注冊一次broker，後續開啟任務定時注冊
this.registerBrokerAll(true, false, true);

//開啟定時任務，注冊broker
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

    @Override
    public void run() {
        try {
            BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
        } catch (Throwable e) {
            log.error("registerBrokerAll Exception", e);
        }
    }
}, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);

Broker定時發送心跳包，上報NameServer自身的存活狀态。

org.apache.rocketmq.broker.out.BrokerOuterAPI#registerBrokerAll代碼片段

//擷取namesrv位址清單
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList(); 

for (final String namesrvAddr : nameServerAddressList) {
    brokerOuterExecutor.execute(new Runnable() {
        @Override
        public void run() {
            try {
            	//注冊broker
                RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
                if (result != null) {
                    registerBrokerResultList.add(result);
                }

                log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
            } catch (Exception e) {
                log.warn("registerBroker Exception, {}", namesrvAddr, e);
            } finally {
                countDownLatch.countDown();
            }
        }
    });
}

周遊NameServer清單，Broker消息伺服器依次向NameServer發送心跳包。

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#registerBroker代碼片段

//2. 維護broker資訊，擷取broker資訊，不存在，新建立後，加入集合中
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
if (null == brokerData) {
    registerFirst = true;
    brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
    this.brokerAddrTable.put(brokerName, brokerData);
}
Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();
//Switch slave to master: first remove <1, IP:PORT> in namesrv, then add <0, IP:PORT>
//The same IP:PORT must only have one record in brokerAddrTable
Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();
while (it.hasNext()) {
    Entry<Long, String> item = it.next();
    //新注冊的broker與舊的不一緻，則删除舊的broker資訊
    if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {
        it.remove();
    }
}

String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr); //緩存broker位址
registerFirst = registerFirst || (null == oldAddr); //判斷是否第一次注冊
//3. broker為master，且初次注冊或者topic配置資訊發生變化，則建立或更新topic的路由中繼資料
if (null != topicConfigWrapper
    && MixAll.MASTER_ID == brokerId) {
    if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
        || registerFirst) {
        ConcurrentMap<String, TopicConfig> tcTable =
            topicConfigWrapper.getTopicConfigTable();
        if (tcTable != null) {
            for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
                this.createAndUpdateQueueData(brokerName, entry.getValue());
            }
        }
    }
}
//4. 更新broker存活資訊
BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,
    new BrokerLiveInfo(
        System.currentTimeMillis(), //Broker存活的最新更新時間
        topicConfigWrapper.getDataVersion(),
        channel,
        haServerAddr));
if (null == prevBrokerLiveInfo) {
    log.info("new broker registered, {} HAServer: {}", brokerAddr, haServerAddr);
}
//5. 注冊broker的過濾器
if (filterServerList != null) {
    if (filterServerList.isEmpty()) {
        this.filterServerTable.remove(brokerAddr);
    } else {
        this.filterServerTable.put(brokerAddr, filterServerList);
    }
}
//6. broker非master，擷取master的broker位址傳回
if (MixAll.MASTER_ID != brokerId) {
    String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
    if (masterAddr != null) {
        BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);
        if (brokerLiveInfo != null) {
            result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
            result.setMasterAddr(masterAddr);
        }
    }
}

3) 路由剔除

根據前面的分析，我們知道Broker每隔30s向NameServer發送一個心跳包，上報自己的存活資訊。

那麼如果Broker當機，NameServer無法收到心跳包，此時NameServer如何來剔除這些失效的Broker呢？

NamerServer會每隔10s掃描brokerLiveTable狀态表，如果BrokerLive的lastUpdateTimestamp的時間戳距目前時間超過120s，則認為Broker失效，移除該Broker，關閉與Broker的連接配接，并同時更新topicQueueTable、brokerAddrTable、brokerLiveTable、filterServerTable。

RocktMQ 有兩個觸發點來觸發路由删除：

NameServer 定時掃描 brokerLiveTable 檢測上次心跳包與目前系統時間的時間差，如果時間戳大于 120s ，則需要移除該 Broker 資訊。
Broker 在正常被關閉的情況下，會執行 unregisterBroker 指令。

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#scanNotActiveBroker代碼片段

public void scanNotActiveBroker() {
    Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
    while (it.hasNext()) {
        Entry<String, BrokerLiveInfo> next = it.next();
        long last = next.getValue().getLastUpdateTimestamp();
        //判斷broker更新的最近時間戳，距目前事件超過120s
        //則認為broker失效，移除該broker，關閉與broker的連接配接
        if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
            RemotingUtil.closeChannel(next.getValue().getChannel());
            it.remove();
            log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);
            this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
        }
    }
}

4) 路由發現

RocketMQ路由發現是非實時的，當Topic路由出現變化後，NameServer并不主動推送給用戶端，而是由用戶端定時拉取Topic最新的路由。

org.apache.rocketmq.client.impl.factory.MQClientInstance#startScheduledTask代碼片段

private void startScheduledTask() {
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            try {
                MQClientInstance.this.updateTopicRouteInfoFromNameServer();
            } catch (Exception e) {
                log.error("ScheduledTask updateTopicRouteInfoFromNameServer exception", e);
            }
        }
    }, 10, this.clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);

    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            try {
                MQClientInstance.this.cleanOfflineBroker();
                MQClientInstance.this.sendHeartbeatToAllBrokerWithLock();
            } catch (Exception e) {
                log.error("ScheduledTask sendHeartbeatToAllBroker exception", e);
            }
        }
    }, 1000, this.clientConfig.getHeartbeatBrokerInterval(), TimeUnit.MILLISECONDS);
}

Producer/Consumer啟動後，開啟定時任務從NameServer更新Topic路由資訊。

org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#getRouteInfoByTopic代碼片段

public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,
    RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final GetRouteInfoRequestHeader requestHeader =
        (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
    //1. 擷取topic的路由資訊
    TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());

    if (topicRouteData != null) {
        //2. 若topic為順序消息，則從KV中擷取順序消息填充路由
        if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {
            String orderTopicConf =
                this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,
                    requestHeader.getTopic());
            topicRouteData.setOrderTopicConf(orderTopicConf);
        }

        byte[] content = topicRouteData.encode();
        response.setBody(content);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

    response.setCode(ResponseCode.TOPIC_NOT_EXIST);
    response.setRemark("No topic route info in name server for the topic: " + requestHeader.getTopic()
        + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
    return response;
}

通過上述分析，NameServer其核心在于Broker路由注冊及路由查詢，但有一個關鍵點，Broker失效後，NameServer是定時查詢失效Broker并删除路由資訊的，也就是說，在某一個時刻，Broker已經當機了，但是NameServer中還維持有該Broker路由資訊，這樣Producer在查詢路由資訊時，就可能從NameServer查詢到一個已經失效的路由資訊，這不是違背了高可用的原則了？

實際上MQ在消息發送/消費時，針對這種情況做了容錯處理。

引用
《RocketMQ技術内幕》

RocketMQ：深入了解路由中心（NameServer）

一、架構設計

二、源碼實作

1. NameServer啟動流程

2. NameServer核心業務邏輯處理

3. NameServer路由注冊、故障剔除

引用

繼續閱讀

單片機mcu—time定時器詳解

Camera安卓源碼-高通mm_camera架構剖析

自然語言了解（NLU）相關微信小程式大全

jquery3.0源碼解讀（四）Callbacks

使用Socket實作最簡單的聊天功能

Linux指令行源碼查找方式

實驗吧WEBWP(一）

hadoop實戰之單表關聯STjoin源碼

ROS Navigation之amcl源碼解析（完全詳解）0. 寫在最前面1. amcl是幹什麼的2. 總體情況3. amcl_node.cpp參考

DOS源碼相關資料

聊聊 RocketMQ 名字服務

Mybatis源碼閱讀（一）：Mybatis初始化1.1 解析properties、settings前言入口引申結語

MyBatis源碼解析(一)——MyBatis初始化過程解MyBatis源碼解析(一)——MyBatis初始化過程解1. 準備工作2. MyBatis初始化過程

《資料結構》（嚴蔚敏,吳偉民版）課本源碼+習題集解析使用說明先附上文檔歸類目錄：部落客有話說：(已遷移到部落格園 ☛☛☛ 新部落格連結)

一步一步解析集合架構ArrayList源碼（2）

如何提高個人開源網站源碼開發使用率