文章目錄

• • NameServer路由管理
  • • 1.前言
    • 2.路由元信息
    • 3.路由注冊
    • • 3.1Broker路由注冊
      • 3.2NameServer處理路由注冊
      • 3.3路由刪除
      • • 3.3.1Broker異常關閉
        • 3.3.2Broker正常關閉
      • 3.4路由發現
      • 3.5總結

NameServer路由管理

1.前言

NameServer主要作用是為消息消費者和消息生產者提供關于主題Topic的路由信息，那么NameServer需要存儲路由的基本信息，還要管理Broker節點，包括路由注冊、路由刪除等。

2.路由元信息

路由元信息主要由RouteInfoManager來進行管理，這個類在上一篇文章中已經介紹過，這里再做一下簡單的介紹。

private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2; private final ReadWriteLock lock = new ReentrantReadWriteLock(); private final HashMap<String/* topic */, List<QueueData>> topicQueueTable; private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable; private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable; private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable; private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

BROKER_CHANNEL_EXPIRED_TIME：NameServer與Broker空閑連接時長，在2minNameServer之內沒有收到Broker的心跳包，則NameServer會關閉與該Broker的連接并刪除Broker的路由信息。

lock：讀寫鎖，用來保護以下用于存儲關鍵信息的非線程安全容器HashMap。

topicQueueTable：Topic消息隊列路由信息，消息發送時根據路由表進行負載均衡

brokerAddrTable：Broker基礎信息，包括brokerName、所屬集群名稱、主備Broker地址

clusterAddrTable：Broker集群信息，存儲集群中所有Broker名稱

brokerLiveTable：Broker狀態信息，NameServer每次收到心跳包會對該表進行信息維護。

filterServerTable：Broker上的FilterServer列表，用于類模式消息過濾。

RocketMQ基于定于發布機制，一個Topic擁有多個消息隊列，一個Broker為每一個主題創建4個讀隊列和4個寫隊列。多個Broker組成一個集群，集群由相同的多臺Broker組成Master-Slave架構，brokerId為0代表Master，大于0為Slave。BrokerLiveInfo中的lastUpdateTimestamp存儲上次收到Broker心跳包的時間。

3.路由注冊

以下是路由注冊的時序圖：

RocketMQ的路由注冊功能是通過Broker與NameServer的心跳機制實現的。Broker啟動時向NameServer集群發送心跳信息，每隔30s向集群所有NameServer發送心跳包，NameServer接收到之后會對brokerLiveTable中的BrokerLiveInfo的lastUpdateTimeStamp屬性值進行更新。此外NameServer會每隔10s掃描brokerLiveTable，如果檢測到有Broker連續120s沒有發送心跳包，則NameServer將移除Broker的路由信息同時關閉Socket連接。

---

源碼

3.1Broker路由注冊

BrokerController#start

//...進行檢查if (!messageStoreConfig.isEnableDLegerCommitLog()) {startProcessorByHa(messageStoreConfig.getBrokerRole());handleSlaveSynchronize(messageStoreConfig.getBrokerRole());//注冊Broker信息this.registerBrokerAll(true, false, true);}//創建定時任務線程池this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {try {//每隔三十秒注冊Broker信息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);if (this.brokerStatsManager != null) {this.brokerStatsManager.start();}if (this.brokerFastFailure != null) {this.brokerFastFailure.start();} }

BrokerController#registerBrokerAll

if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),this.getBrokerAddr(),this.brokerConfig.getBrokerName(),this.brokerConfig.getBrokerId(),this.brokerConfig.getRegisterBrokerTimeoutMills())) {//開始注冊doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper); }

BrokerController#doRegisterBrokerAll

//進入brokerOuterAPI#registerBrokerAll List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(this.brokerConfig.getBrokerClusterName(),this.getBrokerAddr(),this.brokerConfig.getBrokerName(),this.brokerConfig.getBrokerId(),this.getHAServerAddr(),topicConfigWrapper,this.filterServerManager.buildNewFilterServerList(),oneway,this.brokerConfig.getRegisterBrokerTimeoutMills(),this.brokerConfig.isCompressedRegister());

BrokerOuterAPI#registerBrokerAll

final List<RegisterBrokerResult> registerBrokerResultList = new CopyOnWriteArrayList<>(); //獲取nameServer地址信息 List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList(); if (nameServerAddressList != null && nameServerAddressList.size() > 0) {//進行請求頭的封裝final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();requestHeader.setBrokerAddr(brokerAddr);requestHeader.setBrokerId(brokerId);requestHeader.setBrokerName(brokerName);requestHeader.setClusterName(clusterName);requestHeader.setHaServerAddr(haServerAddr);requestHeader.setCompressed(compressed);//封裝注冊請求體RegisterBrokerBody requestBody = new RegisterBrokerBody();requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);requestBody.setFilterServerList(filterServerList);final byte[] body = requestBody.encode(compressed);final int bodyCrc32 = UtilAll.crc32(body);requestHeader.setBodyCrc32(bodyCrc32);final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());//遍歷nameServer地址 ---> 分別向各個nameServer注冊Brokerfor (final String namesrvAddr : nameServerAddressList) {brokerOuterExecutor.execute(new Runnable() {@Overridepublic void run() {try {//向nameServer注冊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();}}});}try {countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);} catch (InterruptedException e) {} }return registerBrokerResultList;

BrokerOuterAPI#registerBroker

//如果是oneway發送 無需關注返回結果 直接返回null if (oneway) {try {this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);} catch (RemotingTooMuchRequestException e) {// Ignorelog.error("RemotingTooMuchRequestException!!!");}return null; } //返回遠程調用的響應結果 RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);

---

3.2NameServer處理路由注冊

以下是NameServer處理路由注冊的時序圖：

請求發送至NameServer，會由默認的網絡處理解析類DefaultRequestProcessor進行處理解析，如果請求類型是REGISTER_BROKER則會將請求轉發至RouteInfoManager進行路由的注冊。

源碼

DefaultRequestProcessor#processRequest

switch (request.getCode()) {case RequestCode.REGISTER_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);} }

DefaultRequestProcessor#registerBrokerWithFilterServer

//轉發到NameServerController中的RouteInfoManager進行路由注冊 RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(requestHeader.getClusterName(),requestHeader.getBrokerAddr(),requestHeader.getBrokerName(),requestHeader.getBrokerId(),requestHeader.getHaServerAddr(),registerBrokerBody.getTopicConfigSerializeWrapper(),registerBrokerBody.getFilterServerList(),ctx.channel());

RouteInfoManager#registerBroker

//加入寫鎖-保護HashTable線程安全 this.lock.writeLock().lockInterruptibly(); //維護clusterAddrTable Set<String> brokerNames = this.clusterAddrTable.get(clusterName); if (null == brokerNames) {brokerNames = new HashSet<String>();this.clusterAddrTable.put(clusterName, brokerNames); } brokerNames.add(brokerName); //維護brokerAddrTable //是否是第一次注冊 默認為false boolean registerFirst = false; BrokerData brokerData = this.brokerAddrTable.get(brokerName); //如果在brokerAddrTable沒有該Broker信息 if (null == brokerData) {//設置為第一次注冊registerFirst = true;//創建對應的BrokerData信息brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());this.brokerAddrTable.put(brokerName, brokerData); } //不是第一次注冊-對Broker信息進行更新 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();if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {it.remove();} }String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr); registerFirst = registerFirst || (null == oldAddr); //維護topicQueueTable 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());}}} } //維護brokerLiveTable BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,new BrokerLiveInfo(System.currentTimeMillis(),topicConfigWrapper.getDataVersion(),channel,haServerAddr)); if (null == prevBrokerLiveInfo) {log.info("new broker registered, {} HAServer: {}", brokerAddr, haServerAddr); } //維護filterServerTable if (filterServerList != null) {if (filterServerList.isEmpty()) {this.filterServerTable.remove(brokerAddr);} else {this.filterServerTable.put(brokerAddr, filterServerList);} }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);}} }

RouteInfoManager#createAndUpdateQueueData

//創建并封裝新的隊列信息 QueueData queueData = new QueueData(); //Broker名稱 queueData.setBrokerName(brokerName); //讀寫隊列個數 queueData.setWriteQueueNums(topicConfig.getWriteQueueNums()); queueData.setReadQueueNums(topicConfig.getReadQueueNums()); //操作權限 queueData.setPerm(topicConfig.getPerm()); //同步復制還是異步復制的標識 queueData.setTopicSysFlag(topicConfig.getTopicSysFlag());//根據Topic從topicQueueTable中獲取隊列集合 List<QueueData> queueDataList = this.topicQueueTable.get(topicConfig.getTopicName()); //如果沒有獲取到 if (null == queueDataList) {//創建并將新的queueData添加到集合中queueDataList = new LinkedList<QueueData>();queueDataList.add(queueData);//放入topicQueueTablethis.topicQueueTable.put(topicConfig.getTopicName(), queueDataList);log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData); } //queueDataList不為空 else {boolean addNewOne = true;Iterator<QueueData> it = queueDataList.iterator();//對該主題已有隊列信息進行遍歷while (it.hasNext()) {QueueData qd = it.next();//如果找到相同主題if (qd.getBrokerName().equals(brokerName)) {//已經存在if (qd.equals(queueData)) {addNewOne = false;} else {//進行更改-先刪除后新增log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), qd,queueData);it.remove();}}}//添加一個新的隊列信息if (addNewOne) {queueDataList.add(queueData);} }

3.3路由刪除

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

刪除路由信息的兩個觸發條件：

• NameServer定期掃描brokerLiveTable檢測上次心跳包發送時間與當前系統的時間差，如果時間超過120s，則需要移除該Broker路由信息
• Broker正常關閉，執行unregisterBroker

3.3.1Broker異常關閉

NamesrvController#initialize

//定時任務線程池--->每隔十秒掃描活躍狀態異常的Broker信息 this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {/*** 對Not Active Broker 進行掃描*/@Overridepublic void run() {//在routeInfoManager中進行路由刪除NamesrvController.this.routeInfoManager.scanNotActiveBroker();} }, 5, 10, TimeUnit.SECONDS);

RouteInfoManager#scanNotActiveBroker

//遍歷brokerLiveTable 查看是否有broker狀態長時間沒有更新 Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator(); //遍歷brokerLiveTable while (it.hasNext()) {Entry<String, BrokerLiveInfo> next = it.next();long last = next.getValue().getLastUpdateTimestamp();//如果收到心跳包的時間間距超過120sif ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {//關閉連接RemotingUtil.closeChannel(next.getValue().getChannel());//移除brokerit.remove();log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);//維護路由表this.onChannelDestroy(next.getKey(), next.getValue().getChannel());} }

RouteInfoManager#onChannelDestroy

//待刪除的BrokerAddress String brokerAddrFound = null; if (channel != null) {try {try {//加入讀鎖this.lock.readLock().lockInterruptibly();Iterator<Entry<String, BrokerLiveInfo>> itBrokerLiveTable =this.brokerLiveTable.entrySet().iterator();//遍歷查找待刪除的Brokerwhile (itBrokerLiveTable.hasNext()) {Entry<String, BrokerLiveInfo> entry = itBrokerLiveTable.next();if (entry.getValue().getChannel() == channel) {brokerAddrFound = entry.getKey();break;}}} finally {//釋放鎖this.lock.readLock().unlock();}} catch (Exception e) {log.error("onChannelDestroy Exception", e);} } //加入寫鎖 //根據brokerAddrFound從brokerLiveTable和filterServerTable移除對應的Broker信息 this.lock.writeLock().lockInterruptibly(); this.brokerLiveTable.remove(brokerAddrFound); this.filterServerTable.remove(brokerAddrFound); //維護brokerAddrTable String brokerNameFound = null; boolean removeBrokerName = false; Iterator<Entry<String, BrokerData>> itBrokerAddrTable =this.brokerAddrTable.entrySet().iterator(); //遍歷brokerAddrTable while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {BrokerData brokerData = itBrokerAddrTable.next().getValue();//遍歷brokerAddressIterator<Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();while (it.hasNext()) {Entry<Long, String> entry = it.next();Long brokerId = entry.getKey();String brokerAddr = entry.getValue();//遍歷到匹配的brokerif (brokerAddr.equals(brokerAddrFound)) {brokerNameFound = brokerData.getBrokerName();//根據broker地址移除brokerit.remove();log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",brokerId, brokerAddr);break;}}//如果當前BrokerData中的Broker列表為空if (brokerData.getBrokerAddrs().isEmpty()) {removeBrokerName = true;//從BrokerAddrTable移除該BrokerDataitBrokerAddrTable.remove();log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",brokerData.getBrokerName());} } //維護clusterAddrTable if (brokerNameFound != null && removeBrokerName) {Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();//遍歷clusterAddrTablewhile (it.hasNext()) {Entry<String, Set<String>> entry = it.next();String clusterName = entry.getKey();Set<String> brokerNames = entry.getValue();//根據brokerName刪除cluster集群中的broker并返回是否刪除成功boolean removed = brokerNames.remove(brokerNameFound);if (removed) {log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",brokerNameFound, clusterName);//如果cluster已經不包含任何broker則移除該集群if (brokerNames.isEmpty()) {log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",clusterName);it.remove();}break;}} } //維護TopicQueueTable if (removeBrokerName) {Iterator<Entry<String, List<QueueData>>> itTopicQueueTable =this.topicQueueTable.entrySet().iterator();//遍歷TopicQueueTablewhile (itTopicQueueTable.hasNext()) {Entry<String, List<QueueData>> entry = itTopicQueueTable.next();//主題String topic = entry.getKey();//主題對應的Broker信息List<QueueData> queueDataList = entry.getValue();Iterator<QueueData> itQueueData = queueDataList.iterator();//遍歷Brokerwhile (itQueueData.hasNext()) {QueueData queueData = itQueueData.next();//通過brokerNameFound在queueData中查找對應的Broker信息if (queueData.getBrokerName().equals(brokerNameFound)) {//移除該BrokeritQueueData.remove();log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",topic, queueData);}}//如果當前主題已經沒有任何Broker，則移除該主題Topicif (queueDataList.isEmpty()) {itTopicQueueTable.remove();log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",topic);}} } //釋放寫鎖 finally {this.lock.writeLock().unlock(); }

3.3.2Broker正常關閉

關閉入口-BrokerController

//BrokerController#shutdown this.unregisterBrokerAll();//BrokerController#unregisterBrokerAll this.brokerOuterAPI.unregisterBrokerAll(this.brokerConfig.getBrokerClusterName(),this.getBrokerAddr(),this.brokerConfig.getBrokerName(),this.brokerConfig.getBrokerId());

BrokerOuterAPI#unregisterBrokerAll

//獲取NameServerAddress列表 List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList(); if (nameServerAddressList != null) {//遍歷NameServerfor (String namesrvAddr : nameServerAddressList) {try {//注銷指定Brokerthis.unregisterBroker(namesrvAddr, clusterName, brokerAddr, brokerName, brokerId);log.info("unregisterBroker OK, NamesrvAddr: {}", namesrvAddr);} catch (Exception e) {log.warn("unregisterBroker Exception, {}", namesrvAddr, e);}} }

BrokerOuterAPI#unregisterBroker

//封裝注銷Broker請求頭 UnRegisterBrokerRequestHeader requestHeader = new UnRegisterBrokerRequestHeader(); requestHeader.setBrokerAddr(brokerAddr); requestHeader.setBrokerId(brokerId); requestHeader.setBrokerName(brokerName); requestHeader.setClusterName(clusterName); //封裝注銷請求 RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_BROKER, requestHeader); //遠程調用并發送請求-返回響應結果 RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, 3000); assert response != null; switch (response.getCode()) {case ResponseCode.SUCCESS: {return;}default:break; }throw new MQBrokerException(response.getCode(), response.getRemark(), brokerAddr);

忽略Netty遠程調用過程

DefaultRequestProcess#processRequest

//根據請求狀態碼-進行不同的處理 switch (request.getCode()) {case RequestCode.UNREGISTER_BROKER:return this.unregisterBroker(ctx, request); }

DefaultRequestProcess#unregisterBroker

final RemotingCommand response = RemotingCommand.createResponseCommand(null); //解碼注銷Broker的請求頭 final UnRegisterBrokerRequestHeader requestHeader =(UnRegisterBrokerRequestHeader) request.decodeCommandCustomHeader(UnRegisterBrokerRequestHeader.class); //轉發至RouteInfoManager注銷Broker this.namesrvController.getRouteInfoManager().unregisterBroker(requestHeader.getClusterName(),requestHeader.getBrokerAddr(),requestHeader.getBrokerName(),requestHeader.getBrokerId());//設置響應狀態碼 response.setCode(ResponseCode.SUCCESS); response.setRemark(null); //返回響應結果 return response;

RouteInfoManager#unregisterBroker

try {try {//加入寫鎖-保護HashMap線程安全this.lock.writeLock().lockInterruptibly();//由于是從BrokerController遠程調用來的請求-故不對BrokerAddr進行檢查-直接進行移除//移除brokerLiveTable中的Broker信息BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.remove(brokerAddr);log.info("unregisterBroker, remove from brokerLiveTable {}, {}",brokerLiveInfo != null ? "OK" : "Failed",brokerAddr);//維護filterServerTablethis.filterServerTable.remove(brokerAddr);boolean removeBrokerName = false;//根據brokername獲取對應的BrokerData信息BrokerData brokerData = this.brokerAddrTable.get(brokerName);if (null != brokerData) {//刪除對應Broker信息String addr = brokerData.getBrokerAddrs().remove(brokerId);log.info("unregisterBroker, remove addr from brokerAddrTable {}, {}",addr != null ? "OK" : "Failed",brokerAddr);//如果當前BrokerData中已經沒有其他Brokerif (brokerData.getBrokerAddrs().isEmpty()) {//從BrokerAddrTable移除相應的Brokerthis.brokerAddrTable.remove(brokerName);log.info("unregisterBroker, remove name from brokerAddrTable OK, {}",brokerName);removeBrokerName = true;}}if (removeBrokerName) {//在clusterAddrTable中通過集群名稱獲取對應的Broker集合Set<String> nameSet = this.clusterAddrTable.get(clusterName);if (nameSet != null) {//移除該集合中的Brokerboolean removed = nameSet.remove(brokerName);log.info("unregisterBroker, remove name from clusterAddrTable {}, {}",removed ? "OK" : "Failed",brokerName);//如果該集群中的Broker集合為空-移除該集群if (nameSet.isEmpty()) {this.clusterAddrTable.remove(clusterName);log.info("unregisterBroker, remove cluster from clusterAddrTable {}",clusterName);}}//移除該Broker下的所有主題Topicthis.removeTopicByBrokerName(brokerName);}} finally {//釋放寫鎖this.lock.writeLock().unlock();} } catch (Exception e) {log.error("unregisterBroker Exception", e); }

---

3.4路由發現

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

DefaultRequestProcessor#processRequest

switch (request.getCode()) {case RequestCode.GET_ROUTEINFO_BY_TOPIC:return this.getRouteInfoByTopic(ctx, request); }

DefaultRequestProcessor#getRouteInfoByTopic

//遠程調用返回的響應結果 final RemotingCommand response = RemotingCommand.createResponseCommand(null); //解析獲取路由信息的請求頭 final GetRouteInfoRequestHeader requestHeader =(GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class); //調用RouteInfoManager中的方法,從路由表topicQueueTable、brokerAddrTable、filterServerTable中 //分別填充TopicRouteData的List<QueueData>、List<BrokerData>、filterServer TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());//如果找到主題對應你的路由信息并且該主題為順序消息 //則從NameServer KVConfig中獲取關于順序消息相關的配置填充路由信息 if (topicRouteData != null) {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;

RouteInfoManager#pickupTopicRouteData

//創建主題路由信息對象 TopicRouteData topicRouteData = new TopicRouteData(); //是否找到對應隊列的標識 boolean foundQueueData = false; //是否找到對應Broker的標識 boolean foundBrokerData = false; Set<String> brokerNameSet = new HashSet<String>(); List<BrokerData> brokerDataList = new LinkedList<BrokerData>(); topicRouteData.setBrokerDatas(brokerDataList);HashMap<String, List<String>> filterServerMap = new HashMap<String, List<String>>(); topicRouteData.setFilterServerTable(filterServerMap);try {try {//加入讀鎖this.lock.readLock().lockInterruptibly();//從topicQueueTable中通過主題topic查找Broker列表List<QueueData> queueDataList = this.topicQueueTable.get(topic);if (queueDataList != null) {//填充Broker信息到topicRouteData.QueueDatastopicRouteData.setQueueDatas(queueDataList);foundQueueData = true;Iterator<QueueData> it = queueDataList.iterator();while (it.hasNext()) {QueueData qd = it.next();brokerNameSet.add(qd.getBrokerName());}//遍歷Broker集合for (String brokerName : brokerNameSet) {//從brokerAddrTable中通過BrokerName進行查找BrokerData brokerData = this.brokerAddrTable.get(brokerName);if (null != brokerData) {BrokerData brokerDataClone = new BrokerData(brokerData.getCluster(), brokerData.getBrokerName(), (HashMap<Long, String>) brokerData.getBrokerAddrs().clone());//填充BrokerData信息到brokerDatasbrokerDataList.add(brokerDataClone);foundBrokerData = true;for (final String brokerAddr : brokerDataClone.getBrokerAddrs().values()) {List<String> filterServerList = this.filterServerTable.get(brokerAddr);filterServerMap.put(brokerAddr, filterServerList);}}}}} finally {//釋放讀鎖this.lock.readLock().unlock();} } catch (Exception e) {log.error("pickupTopicRouteData Exception", e); }log.debug("pickupTopicRouteData {} {}", topic, topicRouteData);if (foundBrokerData && foundQueueData) {return topicRouteData; } return null;

3.5總結

• 路由注冊：Broker啟動之后會將自己的信息注冊到NameServer，NameServer收到Broker心跳包之后會更新BrokerLiveTable中對應的Broker的相關信息，特別記錄lastUpdateTimestamp。
• 心跳機制：Broker會每隔30s向NameServer發送心跳包，告訴NameServer自己的運行狀態良好。
• 路由發現：NameServer中的主題Topic信息更新之后，NameServer并不會實時將數據推送給訂閱方Producer，而需要Producer自行通過Topic來查詢路由信息。
• 路由刪除：
  • Broker異常宕機：NameServer每隔10s掃描BrokerLiveTable，檢查表中各個Broker最新的發送心跳包時間，如果這個時間與當前系統時間差超過120s，則該Broker會被視作已掉線，NameServer會將這個Broker從表中刪除，并且更新各個表中與該Broker有關的所有信息。
  • Broker正常下線：NameServer接受到BrokerController發送來的UNREGISTER_BROKER請求之后，會執行上述步驟，對路由信息進行更新。

以上。

本文僅作為個人學習使用，如有不足或者錯誤請指正！

總結

以上是生活随笔為你收集整理的RocketMQ：NameServer路由管理源码分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。