[[389254]]

 Topic

Topic是一類消息的集合，是一種邏輯上的分區。為什么說是邏輯分區呢?因為最終數據是存儲到Broker上的，而且為了滿足高可用，采用了分布式的存儲。

這和Kafka中的實現如出一轍，Kafka的Topic也是一種邏輯概念，每個Topic的數據會分成很多份，然后存儲在不同的Broker上，這個「份」叫Partition。而在RocketMQ中，Topic的數據也會分布式的存儲，這個「份」叫MessageQueue。

其分布可以用下圖來表示。

這樣一來，如果某個Broker所在的機器意外宕機，而且剛好MessageQueue中的數據還沒有持久化到磁盤，那么該Topic下的這部分消息就會完全丟失。此時如果有備份的話，MQ就可以繼續對外提供服務。

為什么還會出現沒有持久化到磁盤的情況呢?現在的OS當中，程序寫入數據到文件之后，并不會立馬寫入到磁盤，因為磁盤I/O是非常耗時的操作，在計算機來看是非常慢的一種操作。所以寫入文件的數據會先寫入到OS自己的緩存中去，然后擇機異步的將Buffer中的數據刷入磁盤。

通過多副本冗余的機制，使得RocketMQ具有了高可用的特性。除此之外，分布式存儲能夠應對后期業務大量的數據存儲。如果不使用分布式進行存儲，那么隨著后期業務發展，消息量越來越大，單機是無論如何也滿足不了RocketMQ消息的存儲需求的。如果不做處理，那么一臺機器的磁盤總有被塞滿的時候，此時的系統就不具備可伸縮的特性，也無法滿足業務的使用要求了。

但是這里的可伸縮，和微服務中的服務可伸縮還不太一樣。因為在微服務中，各個服務是無狀態的。而Broker是有狀態的，每個Broker上存儲的數據都不太一樣，因為Producer在發送消息的時候會通過指定的算法，從Message Queue列表中選出一個MessageQueue發送消息。

如果不是很理解這個橫向擴展，那么可以把它當成Redis的Cluster，通過一致性哈希，選擇到Redis Cluster中的具體某個節點，然后將數據寫入Redis Master中去。如果此時想要擴容很方便，只需要往Redis Cluster中新增Master節點就好了。

所以，數據分布式的存儲本質上是一種數據分片的機制。在此基礎上，通過冗余多副本，達成了高可用。

Broker

Broker可以理解為我們微服務中的一個服務的某個實例，因為微服務中我們的服務一般來說都會多實例部署，而RocketMQ也同理，多實例部署可以幫助系統扛住更多的流量，也從某種方面提高了系統的健壯性。

在RocketMQ4.5之前，它使用主從架構，每一個Master Broker都有一個自己的Slave Broker。

那RocketMQ的主從Broker是如何進行數據同步的呢?

Broker啟動的時候，會啟動一個定時任務，定期的從Master Broker同步全量的數據。

這塊可以先不用糾結，后面我們會通過源碼來驗證這個主從同步邏輯。

上面提到了Broker會部署很多個實例，那么既然多實例部署，那必然會存在一個問題，客戶端是如何得知自己是連接的哪個服務器?如何得知對應的Broker的IP地址和端口?如果某個Broker突然掛了怎么辦?

NameServer

這就需要NameServer了，NameServer是什么?

這里先拿Spring Cloud舉例子——Spring Cloud中服務啟動的時候會將自己注冊到Eureka注冊中心上。當服務實例啟動的時候，會從Eureka拉取全量的注冊表，并且之后定期的從Eureka增量同步，并且每隔30秒發送心跳到Eureka去續約。如果Eureka檢測到某個服務超過了90秒沒有發送心跳，那么就會該服務宕機，就會將其從注冊表中移除。

RocketMQ中，NameServer充當的也是類似的角色。兩者從功能上也有一定的區別。

Broker在啟動的時候會向NameServer注冊自己，并且每隔30秒向NameServerv發送心跳。如果某個Broker超過了120秒沒有發送心跳，那么就會認為該Broker宕機，就會將其從維護的信息中移除。這塊后面也會從源碼層面驗證。

當然NameServer不僅僅是存儲了各個Broker的IP地址和端口，還存儲了對應的Topic的路由數據。什么是路由數據呢?那就是某個Topic下的哪個Message Queue在哪臺Broker上。

Producer

總體流程

接下來，我們來看看Producer發送一條消息到Broker的時候會做什么事情，整體的流程如下。

檢查消息合法性

整體來看，其實是個很簡單的操作，跟我們平時寫代碼是一樣的，來請求了先校驗請求是否合法。Producer啟動這里會去校驗當前Topic數據的合法性。

• Topic名稱中是否包含了非法字符
• Topic名稱長度是否超過了最大的長度限制，由常量TOPIC_MAX_LENGTH來決定，其默認值為127
• 當前消息體是否是NULL或者是空消息
• 當前消息體是否超過了最大限制，由常量maxMessageSize決定，值為1024 * 1024 * 4，也就是4M。

都是些很常規的操作，和我們平時寫的checker都差不多。

獲取Topic的詳情

當通過了消息的合法性校驗之后，就需要繼續往下走。此時的關注點就應該從消息是否合法轉移到我要發消息給誰。

此時就需要通過當前消息所屬的Topic拿到Topic的詳細數據。

獲取Topic的方法源碼在上面已經給出來了，首先會從內存中維護的一份Map中獲取數據。順帶一提，這里的Map是ConcurrentHashMap，是線程安全的，和Golang中的Sync.Map類似。

當然，首次發送的話，這個Map肯定是空的，此時會調用NameServer的接口，通過Topic去獲取詳情的Topic數據，此時會在上面的方法中將其加入到Map中去，這樣一來下次再往該Topic發送消息就能夠直接從內存中獲取。這里就是簡單的實現的緩存機制 。

從方法名稱來看，是通過Topic獲取路由數據。實際上該方法，通過調用NameServer提供的API，更新了兩部分數據，分別是：

• Topic路由信息
• Topic下的Broker相關信息

而這兩部分數據都來源于同一個結構體TopicRouteData。其結構如下。

通過源碼可以看到，就包含了該Topic下所有Broker下的Message Queue相關的數據、所有Broker的地址信息。

發送的具體Queue

此時我們獲取到了需要發送到的Broker詳情，包括地址和MessageQueue，那么此時問題的關注點又該從「消息發送給誰」轉移到「消息具體發送到哪兒」。

什么叫發送到哪兒?

開篇提到過一個Topic下會被分為很多個MessageQueue，「發送到哪兒」指的就是具體發送到哪一個Message Queue中去。

Message Queue選擇機制

核心的選擇邏輯

還是先給出流程圖

核心邏輯，用大白話講就是將一個隨機數和Message Queue的容量取模。這個隨機數存儲在Thread Local中，首次計算的時候，會直接隨機一個數。

此后，都直接從ThreadLocal中取出該值，并且+1返回，拿到了MessageQueue的數量和隨機數兩個關鍵的參數之后，就會執行最終的計算邏輯。

接下來，我們來看看選擇Message Queue的方法SelectOneMessageQueue都做了什么操作吧。

可以看到，主邏輯被變量sendLatencyFaultEnable分為了兩部分。

容錯機制下的選擇邏輯

該變量表意為發送延遲故障。本質上是一種容錯的策略，在原有的MessageQueue選擇基礎上，再過濾掉不可用的Broker，對之前失敗的Broker，按一定的時間做退避。

可以看到，如果調用Broker信息發生了異常，那么就會調用updateFault這個方法，來更新Broker的Aviable情況。注意這個參數isolation的值為true。接下來我們從源碼級別來驗證上面說的退避3000ms的事實。

可以看到，isolation值是true，則duration通過三元運算符計算出來結果為30000，也就是30秒。所以我們可以得出結論，如果發送消息拋出了異常，那么直接會將該Broker設置為30秒內不可用。

而如果只是發送延遲較高，則會根據如下的map，根據延遲的具體時間，來判斷該設置多少時間的不可用。

例如，如果上次請求的latency超過550ms，就退避3000ms;超過1000，就退避60000;

正常情況下的選擇邏輯

而正常情況下，如果當前發送故障延遲沒有啟用，則會走常規邏輯，同樣的會去for循環計算，循環中取到了MessageQueue之后會去判斷是否和上次選擇的MessageQueue屬于同一個Broker，如果是同一個Broker，則會重新選擇，直到選擇到不屬于同一個Broker的MessageQueue，或者直到循環結束。這也是為了將消息均勻的分發存儲，防止數據傾斜。

發送消息

選到了具體的Message Queue之后就會開始執行發送消息的邏輯，就會調用底層Netty的接口給發送出去，這塊暫時沒啥可看的。

Broker的啟動流程

主從同步

在上面提到過，RocketMQ有自己的主從同步，但是有兩個不同的版本，版本的分水嶺是在4.5版本。這兩個版本區別是什么呢?

• 4.5之前：有點類似于Redis中，我們手動的將某臺機器通過命令slave of 變成另一臺Redis的Slave節點，這樣一來就變成了一個較為原始的一主一從的架構。為什么說原始呢?因為如果此時Master節點宕機，我們需要人肉的去做故障轉移。RocketMQ的主從架構也是這種情況。
• 4.5之后：引入了Dleger，可以實現一主多從，并且實現自動的故障轉移。這就跟Redis后續推出了Sentinel是一樣的。Dleger也是類似的作用。

下圖是Broker啟動代碼中的源碼。

可以看到判斷了是否開啟了Dleger，默認是不開啟的。所以就會執行其中的邏輯。

剛好我們就看到了，里面有Rocket主從同步數據的相關代碼。

如果當前Broker節點的角色是Slave，則會啟動一個周期性的定時任務，定期(也就是10秒)去Master Broker同步全量的數據。同步的數據包括：

• Topic的相關配置
• Cosumer的消費偏移量
• 延遲消息的Offset
• 訂閱組的相關數據和配置

注冊Broker

完成了主動同步定時任務的啟動之后，就會去調用registerBrokerAll去注冊Broker?？赡苓@里會有點疑問，我這里是Broker啟動，只有當前一個Broker實例，那這個All是什么意思呢?

All是指所有的NameServer，Broker啟動的時候會將自己注冊到每一個NameServer上去。為什么不只注冊到一個NameServer就完事了呢?這樣一來還可以提高效率。歸根結底還是高可用的問題。

如果Broker只注冊到了一臺NameServer上，萬一這臺NameServer掛了呢?這個Broker對所有客戶端就都不可見了。實際上Broker還在正常的運行。

進到registerBrokerAll中去。

可以看到，這里會判斷是否需要進行注冊。通過上面的截圖可以看到，此時forceRegister的值為true，而是否要注冊，決定權就交給了needRegister

為什么需要判斷是否需要注冊呢?因為Broker一旦注冊到了NameServer之后，由于Producer不停的在寫入數據，Consumer也在不停的消費數據，Broker也可能因為故障導致某些Topic下的Message Queue等關鍵的路由信息發生變動。

這樣一來，NameServer中的數據和Broker中的數據就會不一致。

如何判斷是否需要注冊

大致的思路是，Broker會從每一個NameServer中獲取到當前Broker的數據，并和當前Broker節點中的數據做對比。但凡有一臺NameServer數據和當前Broker不一致，都會進行注冊操作。

接下來，我們從源碼層面驗證這個邏輯。關鍵的邏輯我在圖中也標注了出來。

可以看到， 就是通過對比Broker中的數據版本和NameServer中的數據版本來實現的。這個版本，注冊的時候會寫到注冊的數據中存入NameServer中。

這里由于是有多個，所以RocketMQ用線程池來實現了多線程操作，并且用CountDownLatch來等待所有的返回結果。經典的用空間換時間，Golang里面也有類似的操作，那就是sync.waitGroup。

關于任何一個數據不匹配，都會進行重新注冊的事實，我們也從源碼層面來驗證一下。

可以看到，如果任何一臺NameServer的數據發生了Change，都會break，返回true。

這里的結果列表使用的是CopyOnWriteList來實現的。

因為這里是多線程去執行的判斷邏輯，而正常的列表不是線程安全的。CopyOnWriteArrayList之所以是線程安全的，這歸功于COW(Copy On Write)，讀請求時共用同一個List，涉及到寫請求時，會復制出一個List，并在寫入數據的時候加入獨占鎖。比起直接對所有操作加鎖，讀寫鎖的形式分離了讀、寫請求，使其互不影響，只對寫請求加鎖，降低了加鎖的次數、減少了加鎖的消耗，提升了整體操作的并發。

執行注冊邏輯

這塊就是構建數據，然后多線程并發的去發送請求，用CopyOnWriteArrayList來保存結果。不過，上面我們提到過，Broker注冊的時候，會把數據版本發送到NameServer并且存儲起來，這塊我們可以看看發送到NameServer的數據結構。

可以看到，Topic的數據分為了兩部分，一部分是核心的邏輯，另一部分是DataVersion，也就是我們剛剛一直提到的數據版本。

Broker如何存儲數據

剛剛在聊Producer最后提到的是，發送消息到Broker就完了。不知道大家有沒有想過Broker是如何存儲消息的?

Commit log

先給出流程圖

然后給出結論，Producer發送的消息是存儲在一種叫commit log的文件中的，Producer端每次寫入的消息是不等長的，當該CommitLog文件寫入滿1G，就會新建另一個新的CommitLog，繼續寫入。此次采取的是順序寫入。

那么問題來了，Consumer來消費的時候，Broker是如何快速找到對應的消息的呢?我們首先排除遍歷文件查找的方法， 因為RocketMQ是以高吞吐、高性能著稱的，肯定不可能采取這種對于很慢的操作。那RocketMQ是如何做的呢?

答案是ConsumerQueue

ConsumerQueue

ConsumerQueue是什么?是文件。引入的目的是什么呢?提高消費的性能。

Broker在收到一條消息的時候，寫入Commit Log的同時，還會將當前這條消息在commit log中的offset、消息的size和對應的Tag的Hash寫入到consumer queue文件中去。

每個MessageQueue都會有對應的ConsumerQueue文件存儲在磁盤上，每個ConsumerQueue文件包含了30W條消息，每條消息的size大小為20字節，包含了8字節CommitLog的Offset、4字節的消息長度、8字節的Tag的哈希值。這樣一來，每個ConsumerQueue的文件大小就約為5.72M。

當該ConsumerQueue文件寫滿了之后，就會再新建一個ConsumerQueue文件，繼續寫入。

所以，ConsumerQueue文件可以看成是CommitLog文件的索引。

負載均衡

什么意思呢?假設我們總共有6個MessageQueue，然后此時分布在了3臺Broker上，每個Broker上包含了兩個queue。此時Consumer有3臺，我們可以大致的認為每個Consumer負責2個MessageQueue的消費。但是這里有一個原則，那就是一個MessageQueue只能被一臺Consumer消費，而一臺Consumer可以消費多個MessageQueue。

為什么?道理很簡單，RocketMQ支持的順序消費，是指的分區順序性，也就是在單個MessageQueue中，消息是具有順序性的，而如果多臺Consumer去消費同一個MessageQueue，就很難去保證順序消費了。

由于有很多個Consumer在消費多個MessageQueue，所以為了不出現數據傾斜，也為了資源的合理分配利用，在Producer發送消息的時候，需要盡可能的將消息均勻的分發給多個MessageQueue。

同時，上面那種一個Consumer消費了2個MessageQueue的情況，萬一這臺Consumer掛了呢?這兩個MessageQueue不就沒人消費了?

以上兩種情況分別是Producer端的負載均衡、Consumer端的負載均衡。

Producer端負載均衡

關于Producer端上面的負載均衡，上面的流程圖已經給了出來，并且給出了源碼的驗證。首先是容錯策略，會去避開一段時間有問題的Broker，并且加上如果選擇了上次的Broker，就會重新進行選擇。

Consumer端負載均衡

首先Consumer端的負責均衡可以由兩個對象觸發：

• Broker
• Consumer自身

Consumer也會向所有的Broker發送心跳，將消息的消費組名稱、訂閱關系集合、消息的通信模式和客戶端的ID等等。Broker收到了Consumer的心跳之后，會將其存在Broker維護的一個Manager中，名字叫ConsumerManager。當Broker監聽到了Consumer數量發生了變動，就會通知Consumer進行Rebalance。

但是如果Broker通知Consumer進行Rebalance的消息丟了呢?這也就是為什么需要對Consumer自身進行觸發的原因。Consumer會在啟動的時候啟動定時任務，周期性的執行rebalance操作。

默認是20秒執行一次。具體的代碼如下。

具體流程

首先，Consumer的Rebalance會獲取到本地緩存的Topic的全部數據，然后向Broker發起請求，拉取該Topic和ConsumerGroup下的所有的消費者信息。此處的Broker數據來源就是Consumer之前的心跳發送過去的數據。然后會對Topic中MessageQueue和消費者ID進行排序，然后用消息隊列默認分配算法來進行分配，這里的默認分配策略是平均分配。

首先會均勻的按照類似分頁的思想，將MessageQueue分配給Consumer，如果分配的不均勻，則會依次的將剩下的MessageQueue按照排序的順序，從上往下的分配。所以在這里Consumer 1被分配到了4個MessageQueue，而Consumer 2被分配到了3個MessageQueue。

Rebalance完了之后，會將結果和Consumer緩存的數據做對比，移除不在ReBalance結果中的MessageQueue，將原本沒有的MessageQueue給新增到緩存中。

觸發時機

• Consumer啟動時 啟動之后會立馬進行Rebalance
• Consumer運行中 運行中會監聽Broker發送過來的Rebalance消息，以及Consumer自身的定時任務觸發的Rebalance
• Consumer停止運行 停止時沒有直接地調用Rebalance，而是會通知Broker自己下線了，然后Broker會通知其余的Consumer進行Rebalance。

換一個角度來分析，其實就是兩個方面，一個是隊列信息發生了變化，另一種是消費者發生了變化。

源碼驗證

然后給出核心的代碼驗證，獲取數據的邏輯如下

驗證了我們剛剛說的獲取了本地的Topic數據緩存，和從Broker端拉取所有的ConsumerID。

接下來是驗證剛說的排序邏輯。

接下來是看判斷結果是否發生了變化的源碼。

可以看到，Consumer通知Broker策略，其本質上就是發送心跳，將更新后的數據通過心跳發送給所有的Broker。

Consumer更多的細節

可能關于Consumer，我們使用的更多一點。例如我們知道我們可以設置集群消費和廣播消息，分別對應RocketMQ中的CLUSTERING和BROADCASTING**。

再比如我們知道，我們可以設置順序消費和并發消費等等，接下來就讓我們用源碼來看看這些功能在RocketMQ中是怎么實現的。

消費模型

在Consumer中，默認都是采用集群消費，這塊在Consumer的代碼中也有體現。

而消費模式的不同，會影響到管理offset的具體實現。

可以看到，當消費模型是廣播模式時，Offset的持久化管理會使用實現LocalFileOffsetStorage

當消費模式是集群消費時，則會使用RemoteBrokerOffsetStore。

具體原因是什么呢?首先我們得知道廣播模式和集群模式的區別在哪兒：

• 廣播模式下，一條消息會被ConsumerGroup中的每一臺機器所消費
• 集群模式下，一條消息只會被ConsumerGroup中的一臺機器消費

所以在廣播模式下，每個ConsumerGroup的消費進度都不一樣，所以需要由Consumer自身來管理Offset。而集群模式下，同個ConsumerGroup下的消費進度其實是一樣的，所以可以交由Broker統一管理。

消費模式

消費模式則分為順序消費和并發消費，分別對應實現MessageListenerOrderly和MessageListenerConcurrently兩種方式。

不同的消費方式會采取不同的底層實現，配置完成之后就會調用start。

拉取消息

接下來我們來看一個跟我們最最相關的問題，那就是我們平時消費的消息到底是怎么樣從Broker發到的Consumer。在靠近啟動Rebalance的地方，Consumer也開啟了一個定時拉取消息的線程。

這個線程做了什么事呢?它會不停的從一個維護在內存中的Queue中獲取一個在寫入的時候就構建好的PullRequest對象，調用具體實現去不停的拉取消息了。

處理消費結果

在這里是否開啟AutoCommit，所做的處理差不了很多，大家也都知道，唯一區別就在于是否自動的提交Offset。對于處理成功的邏輯也差不多，我們平時業務邏輯中可能也并不關心消費成功的消息。我們更多關注的是如果消費失敗了，RocketMQ是怎么處理的?

這是在AutoCommit下，如果消費失敗了的處理邏輯。會記錄一個失敗的TPS，然后這里有一個非常關鍵的邏輯，那就是checkReconsumeTimes。

如果當前消息的重試次數，如果大于了最大的重試消費次數，就會把消費發回給Broker。那最大重試次數是如何定義的。

如果值為-1，那么最大次數就是MAX_VALUE，也就是2147483647。這里有點奇怪啊，按照我們平常的認知，難道不是重試16次嗎?然后就看到了很騷的一句注釋。

-1 means 16 times，這代碼確實有點，一言難盡。

然后，如果超過了最大的次數限制，就會將該消息調用Prodcuer的默認實現，將其發送到死信隊列中。當然，死信隊列也不是什么特殊的存在，就是一個單獨的Topic而已。

通過getRetryTopic來獲取的，默認是給當前的ConsumerGroup名稱加上一個前綴。