RocketMQ NameServer深入剖析

本文將深入剖析rocketmq為什么選擇自己開發NameServer，而不是選擇類似于ZK這樣的開源組件。同時對rocketmq的路由注冊、路由發現、路由剔除進行剖析。并通過結合核心源碼，對筆者的觀點進行驗證。同時對不同類型消息的重試機制，以及客戶端選擇nameserver的策略進行深入講解。

文章第一部分是name server在rocketmq整體架構中的作用，熟悉的同學可以直接跳過。

1 NameServer的作用

Name Server 是專為 RocketMQ 設計的輕量級名稱服務，具有簡單、可集群橫吐擴展、無狀態，節點之間互不通信等特點。整個Rocketmq集群的工作原理如下圖所示：

可以看到，Broker集群、Producer集群、Consumer集群都需要與NameServer集群進行通信：

Broker集群

Broker用于接收生產者發送消息，或者消費者消費消息的請求。一個Broker集群由多組Master/Slave組成，Master可寫可讀，Slave只可以讀，Master將寫入的數據同步給Slave。每個Broker節點，在啟動時，都會遍歷NameServer列表，與每個NameServer建立長連接，注冊自己的信息，之后定時上報。

Producer集群

消息的生產者，通過NameServer集群獲得Topic的路由信息，包括Topic下面有哪些Queue，這些Queue分布在哪些Broker上等。Producer只會將消息發送到Master節點上，因此只需要與Master節點建立連接。

Consumer集群

消息的消費者，通過NameServer集群獲得Topic的路由信息，連接到對應的Broker上消費消息。注意，由于Master和Slave都可以讀取消息，因此Consumer會與Master和Slave都建立連接。

2 為什么選擇自己開發NameServer

目前可以作為服務發現組件有很多，如etcd、consul，zookeeper等：

那么為什么rocketmq選擇自己開發一個NameServer，而不是使用這些開源組件呢？特別的，Zookeeper其提供了Master選舉、分布式鎖、數據的發布和訂閱等諸多功能RocketMQ設計之初時參考的另一款消息中間件Kafka就使用了Zookeeper。

事實上，在RocketMQ的早期版本，即MetaQ 1.x和MetaQ 2.x階段，也是依賴Zookeeper的。但MetaQ 3.x（即RocketMQ）卻去掉了ZooKeeper依賴，轉而采用自己的NameServer。

而RocketMQ的架構設計決定了只需要一個輕量級的元數據服務器就足夠了，只需要保持最終一致，而不需要Zookeeper這樣的強一致性解決方案，不需要再依賴另一個中間件，從而減少整體維護成本。

敏銳的同學肯定已經意識到了，根據CAP理論，RocketMQ在名稱服務這個模塊的設計上選擇了AP，而不是CP：

• 一致性(Consistency)：Name Server 集群中的多個實例，彼此之間是不通信的，這意味著某一時刻，不同實例上維護的元數據可能是不同的，客戶端獲取到的數據也可能是不一致的。

• 可用性(Availability)：只要不是所有NameServer節點都掛掉，且某個節點可以在指定之間內響應客戶端即可。

• 分區容錯(Partiton Tolerance)：對于分布式架構，網絡條件不可控，出現網絡分區是不可避免的，只要保證部分NameServer節點網絡可達，就可以獲取到數據。具體看公司如何實施，例如：為了實現跨機房的容災，可以將NameServer部署的不同的機房，某個機房出現網絡故障，其他機房依然可用，當然Broker集群/Producer集群/Consumer集群也要跨機房部署。

事實上，除了RocketMQ開發了自己的NameServer，最近 Kafka 社區也在 Wiki 空間上提交了一項新的改進提案“KIP-500: Replace ZooKeeper with a Self-Managed Metadata Quorum”，其目的是為了消除 Kafka 對 ZooKeeper 的依賴，該提案建議用自管理的元數據仲裁機制替換原來的 ZooKeeper 組件。感興趣的讀者可以自行查閱相關資料。

3?NameServer如何保證數據的最終一致

NameServer作為一個名稱服務，需要提供服務注冊、服務剔除、服務發現這些基本功能，但是NameServer節點之間并不通信，在某個時刻各個節點數據可能不一致的情況下，如何保證客戶端可以最終拿到正確的數據。下面分別從路由注冊、路由剔除，路由發現三個角度進行介紹。

3.1?路由注冊

對于Zookeeper、Etcd這樣強一致性組件，數據只要寫到主節點，內部會通過狀態機將數據復制到其他節點，Zookeeper使用的是Zab協議，etcd使用的是raft協議。

但是NameServer節點之間是互不通信的，無法進行數據復制。RocketMQ采取的策略是，在Broker節點在啟動的時候，輪訓NameServer列表，與每個NameServer節點建立長連接，發起注冊請求。NameServer內部會維護一個Broker表，用來動態存儲Broker的信息。

同時，Broker節點為了證明自己是存活的，會將最新的信息上報給NameServer，然后每隔30秒向NameServer發送心跳包，心跳包中包含 BrokerId、Broker地址、Broker名稱、Broker所屬集群名稱等等，然后NameServer接收到心跳包后，會更新時間戳，記錄這個Broker的最新存活時間。

NameServer在處理心跳包的時候，存在多個Broker同時操作一張Broker表，為了防止并發修改Broker表導致不安全，路由注冊操作引入了ReadWriteLock讀寫鎖，這個設計亮點允許多個消息生產者并發讀，保證了消息發送時的高并發，但是同一時刻NameServer只能處理一個Broker心跳包，多個心跳包串行處理。這也是讀寫鎖的經典使用場景，即讀多寫少。

3.2?路由剔除

正常情況下，如果Broker關閉，則會與NameServer斷開長連接，Netty的通道關閉監聽器會監聽到連接斷開事件，然后會將這個Broker信息剔除掉。

異常情況下，NameServer中有一個定時任務，每隔10秒掃描一下Broker表，如果某個Broker的心跳包最新時間戳距離當前時間超多120秒，也會判定Broker失效并將其移除。

特別的，對于一些日常運維工作，例如：Broker升級，RocketMQ提供了一種優雅剔除路由信息的方式。如在升級一個節Master點之前，可以先通過命令行工具禁止這個Broker的寫權限，生產者發送到這個Broker的請求，都會收到一個NO_PERMISSION響應，之后會自動重試其他的Broker。當觀察到這個broker沒有流量后，再將這個broker移除。

3.3?路由發現

路由發現是客戶端的行為，這里的客戶端主要說的是生產者和消費者。具體來說：

• 對于生產者，可以發送消息到多個Topic，因此一般是在發送第一條消息時，才會根據Topic獲取從NameServer獲取路由信息。

• 對于消費者，訂閱的Topic一般是固定的，所在在啟動時就會拉取。

那么生產者/消費者在工作的過程中，如果路由信息發生了變化怎么處理呢？如：Broker集群新增了節點，節點宕機或者Queue的數量發生了變化。細心的讀者注意到，前面講解NameServer在路由注冊或者路由剔除過程中，并不會主動推送會客戶端的，這意味著，需要由客戶端拉取主題的最新路由信息。

事實上，RocketMQ客戶端提供了定時拉取Topic最新路由信息的機制，這里我們直接結合源碼來講解。?? ??? ?

DefaultMQProducer和DefaultMQConsumer有一個pollNameServerInterval配置項，用于定時從NameServer并獲取最新的路由表，默認是30秒，它們底層都依賴一個MQClientInstance類。

MQClientInstance類中有一個updateTopicRouteInfoFromNameServer方法，用于根據指定的拉取時間間隔，周期性的的從NameServer拉取路由信息。?在拉取時，會把當前啟動的Producer和Consumer需要使用到的Topic列表放到一個集合中，逐個從NameServer進行更新。以下源碼截圖展示了這個過程：

然而定時拉取，還不能解決所有的問題。因為客戶端默認是每隔30秒會定時請求NameServer并獲取最新的路由表，意味著客戶端獲取路由信息總是會有30秒的延時。這就帶來一個嚴重的問題，客戶端無法實時感知Broker服務器的宕機。如果生產者和消費者在這30秒內，依然會向這個宕機的broker發送或消費消息呢？

這個問題，可以通過客戶端重試機制來解決。

4 生產者重試機制

在講解生產者重試機制之前，我們必須先對三種消息類型：普通消息、普通有序消息、嚴格有序消息進行介紹。因為RocketMQ客戶端的生產者重試機制，只會普通消息有作用。對于普通有序消息、嚴格有序消息是沒有作用。目前網上絕大部分文章對此并沒有進行區分，導致參考了這些文章的同學誤以為自己的消息發送失敗會自動進行重試，然而事實上可能根本沒有進行重試。三種消息的類型介紹如下：

• 普通消息：消息是無序的，任意發送發送哪一個隊列都可以。

• 普通有序消息：同一類消息(例如某個用戶的消息)總是發送到同一個隊列，在異常情況下，也可以發送到其他隊列。

• 嚴格有序消息：消息必須被發送到同一個隊列，即使在異常情況下，也不允許發送到其他隊列。

對于這三種類型的消息，RocketMQ對應的提供了對應的方法來分別消息，例如同步發送(異步/批量/oneway也是類似)：

需要注意的是：這些方法重載形式，本意是為了支持以上三種不同的消息類型。但是你不按套路出牌，例如：對于一個用戶的多條消息，在調用第一種send方法形式時，依然在對于同一個用戶每次發送消息時，選擇了不同的隊列(MessageQueue)，那么也沒有人能阻止。我只能說，你忽略了RocketMQ團隊設計這三個方法的意圖。

4.1 普通消息的重試

對于普通消息，消息發送默認采用round-robin機制來選擇發送到哪一個隊列，如果發送失敗，默認重試2次。由于之前發送失敗的Queue必然位于某個Broker上，在重試過程中，這個失敗的Broker上的Queue都不會選擇，這里主要是考慮，既然發送到這個Broker上某個Queue失敗了，那么發送到這個Broker上的Queue失敗的可能性依然很大,所以選擇其他Broker。

但是一定會這樣嗎？例如Broker集群只是由一組Master/Slave組成，發送消息只會選擇Master，如果這個Master失敗了，沒有其他Master可選，此時已然會選擇這個Master上的其他Queue。

在實際生產環境中，通常Broker集群至少由2組Master/Slave組成，甚至更多，例如我司就是3主3從。這樣就可以很好的利用RocketMQ對于普通消息發送的重試機制，每次重試到不同的Broker上。

從源碼層面來看，對于普通消息，RocketMQ選擇隊列默認是通過MQFaultStrategy#selectOneMessageQueue來選擇一個的隊列，在未開啟延遲容錯的情況下，內部會調用TopicPublishInfo#selectOneMessageQueue方法，這個方法源碼體了前面說的重試邏輯：

事情到這里并沒有結束，這段代碼只是單次發送消息失敗重試選擇隊列的邏輯。實際情況可能是，在Broker宕機期間，可能會發送多條消息，那么每次都可能會選擇到失敗的Broker上的Queue，然后再重試，盡管重試可能會成功，但是每次發送消息的耗時會增加。因此，MQFaultStrategy實際上還提供了以下兩個功能(超出本文范疇，將會后續其他文章中講解)：

• 失敗隔離：即發送消息到某個broker失敗之后，將其進行隔離，優先從其他正常的broker中進行選擇

• 延遲隔離：優先發送消息到延遲比較小的broker

?? ?? ?對于無序消息，通過這種異常重試機制，就可以保證消息發送的高可用了。同時由于不需要NameServer通知眾多不固定的生產者，也降低了NameServer實現的復雜性。

既然重試機制有這么明顯的好處，那么對于普通有序消息，和嚴格有序消息，rocketmq為什么默認不進行重試呢？答案很簡單，這些消息只能發送某個特定的Broker上的某個特定的Queue中，如果發送失敗，重試失敗的可能依然很大，所以默認不進行重試。如果需要重試，需要業務方自己來做。

4.2 普通有序消息失敗情況下的短暫無序

首先說明，對于普通有序消息，RocketMQ是不會進行重試的。如果需要重試，那么業務RD同學需要自己編寫重試代碼，例如通過一個for循環，最多重試幾次。這里主要說明：對于普通有序消息，在異常情況下，如何經歷短暫無序之后再恢復有序。從MessageQueueSelector源碼來尋找答案：

可以看到，這個接口到的select方法接收一個List<MessageQueue>類型參數，也就是當前Topic下的隊列集合。這個接口由業務RD實現，生產者客戶端在發送消息之前會回調這個接口。

正常情況下的有序

業務RD在實現這個接口時，為了保證消息的有序?？梢圆扇∫恍┎呗?，例如：發送的是一個用戶的消息，先計算pos=user_id%mqs.size()，之后mqs.get(pos)獲得對應的隊列。因此在正常情況下，一個用戶的消息總是有序的。

異常情況下的短暫無序

在異常情況下，例如一個Broker宕機，路由信息刷新后，這個Broker上隊列就會從List集合中移除。此時按照相同的方式選擇隊列，就會選擇到其他隊列上，造成了無序。但是這個無序是很短暫的，因為之后同一個用戶的信息，都會發送到同一個新的隊列上。如果宕機的broker恢復了，那么再次經歷一下短暫無序，之后又變得有序了。

4.3?嚴格有序消息重試的問題

對于嚴格有序消息，由于直接指定了一個MessageQueue。如果這個MessageQueue所在的Broker宕機了，那么之后的重試必然都失敗，只有無限重試，直到成功。因此，非必要的情況下，是不建議使用嚴格有序消息的。

5?客戶端NameServer選擇策略

前面講解了客戶端在獲取路由信息時，每次都會嘗試先從緩存的路由表中查找Topic路由信息，如果找不到，那么就去NameServer更新嘗試。下面介紹一下客戶端NameServer節點的選擇策略。

RocketMQ會將用戶設置的NameServer列表會設置到NettyRemotingClient類的namesrvAddrList字段中，NettyRemotingClient是RocketMQ對Netty進行了封裝，如下：

private final AtomicReference<List<String>> namesrvAddrList = new AtomicReference<List<String>>();

具體選擇哪個NameServer，也是使用round-robin的策略。需要注意的是，盡管使用round-robin策略，但是在選擇了一個NameServer節點之后，后面總是會優先選擇這個NameServer，除非與這個NameServer節點通信出現異常的情況下，才會選擇其他節點。

為什么客戶端不與所有NameServer節點建立連接呢，而是只選擇其中一個？筆者考慮，通常NameServer節點是固定的幾個，但是客戶端的數量可能是成百上千，為了減少每個NameServer節點的壓力，所以每個客戶端節點只隨機與其中一個NameServer節點建立連接。

為了盡可能保證NameServer集群每個節點的負載均衡，在round-robin策略選擇時，每個客戶端的初始隨機位置都不同，如下：

private?final?AtomicInteger?namesrvIndex?=?new?AtomicInteger(initValueIndex());

其中initValueIndex()就是計算一個隨機值，之后每次選擇NameServer時，namesrvIndex+1之后，對namesrvAddrList取模，計算在數據下標的位置，嘗試創建連接，一旦創建成功，會將當前選擇的NameServer地址記錄到namesrvAddrChoosed字段中：

private final AtomicReference<String> namesrvAddrChoosed = new AtomicReference<String>();

如果某個NameServer節點創建連接失敗是，會自動重試其他節點。具體可參見：getAndCreateNameserverChannel最后留一個筆者沒有想明白的問題，選擇NameServer的初始隨機位置是 ?initValueIndex()方法，這個方法的實現如下：

private static int initValueIndex() {
    Random r = new Random();
    return Math.abs(r.nextInt() % 999) % 999;
}