[[437202]]

本文轉載自微信公眾號「程序員jinjunzhu」，作者jinjunzhu 。轉載本文請聯系程序員jinjunzhu公眾號。

RocketMQ 是阿里巴巴的分布式消息中間件，在 2012 年開源，在 2017 年成為 Apache 頂級項目。

1、集群架構

RocketMQ 的集群架構如下圖：

從上圖可以看到，整個集群中有四個角色：Name Server集群、Broker主從集群、Producer、Consumer。

1.1 Name Server 集群

Name Server 集群部署，但是節點之間并不會同步數據，因為每個節點都會保存完整的數據。因此單個節點掛掉，并不會對集群產生影響。

1.2 Broker

Broker 采用主從集群，實現多副本存儲和高可用。每個 Broker 節點都要跟所有的 Name Server 節點建立長連接，定義注冊 Topic 路由信息和發送心跳。

跟所有 Name Server 建立連接，就不會因為單個 Name Server 掛了影響 Broker 使用。Broker 主從模式中， Slave 節點主動從 Master 節點拉取消息。

1.3 Producer

Producer 跟 Name Server 的任意一個節點建立長連接，定期從 Name Server 拉取 Topic 路由信息。Producer 是否采用集群，取決于它所在的業務系統。

1.4 Consumer

Consumer 跟 Name Server 的任意一個節點建立長連接，定期從 Name Server 拉取 Topic 路由信息。Consumer 是否采用集群，取決于它所在的業務系統。

Producer 和 Consumer 只跟任意一個 Name Server 節點建立連接，因為 Broker 會向所有 Name Server 注冊 Topic 信息，所以每個 Name Server 保存的數據其實是一致的。

2、MessageQueue

Producer 發送的消息會在 Broker 的 MessageQueue 中保存，如下圖：

有了 MessageQueue ，Topic 就可以在 Broker 中實現分布式存儲，如上圖，Broker 集群中保存了 4 個MessageQueue，這些 MessageQueue 保存了 Topic1-Topic3 這三個 Topic 的消息。

MessageQueue 類似于 Kafka 中的 Partition，有了 MessageQueue，Producer 可以并發地向 Broker 中發送消息，Consumer 也可以并發地消費消息。

默認Topic 可以創建的 MessageQueue 數量是 4，Broker 可以創建的MessageQueue 數量是 8, RocketMQ 選擇二者中數量小的，也就是 4。不過這兩個值都可以配置。

3、Consumer

RocketMQ的消費模式如下圖：

圖中，Topic1 的消息寫入了兩個 MessageQueue，兩個隊列保存在 Broker1 和 Broker2 上。

RocketMQ 通過 Consumer Group 實現消息廣播。比如上圖中有兩個消費者組，每個消費者組有兩個消費者。

一個消費者可以消費多個 MessageQueue，但是同一個 MessageQueue 只能被同一個消費者組的一個消費者消費。比如 MessageQueue0 只能被 Consumer Group1 中的 Consumer1 消費， 不能被 Consumer2 消費。

4、Broker 高可用集群

Broker 集群如下圖：

Broker 通過主從集群來實現消息高可用。跟 Kafka 不同的是，RocketMQ 并沒有 Master 節點選舉功能，而是采用多 Master 多Slave 的集群架構。Producer 寫入消息時寫入 Master 節點，Slave 節點主動從 Master 節點拉取數據來保持跟 Master 節點的數據一致。

Consumer 消費消息時，既可以從 Master 節點拉取數據，也可以從 Slave 節點拉取數據。到底是從 Master 拉取還是從 Slave 拉取取決于 Master 節點的負載和 Slave 的同步情況。如果 Master 負載很高，Master 會通知 Consumer 從 Slave 拉取消息，而如果 Slave 同步消息進度延后，則 Master 會通知 Consumer 從 Master 拉取數據。總之，從 Master 拉取還是從 Slave 拉取由 Master 來決定。

如果 Master 節點發生故障，RocketMQ 會使用基于 raft 協議的 DLedger 算法來進行主從切換。

Broker 每隔 30s 向 Name Server 發送心跳，Name Server 如果 120s 沒有收到心跳，就會判斷 Broker 宕機了。

5、消息存儲

RocketMQ 的存儲設計是非常有創造性的。存儲文件主要有三個：CommitLog、ConsumeQueue、Index。如下圖：

5.1 CommitLog

RocketMQ 的消息保存在 CommitLog 中，CommitLog 每個文件 1G 大小。有趣的是，文件名并不叫 CommitLog，而是用消息的偏移量來命名。比如第一個文件文件名是 0000000000000000000，第二個文件文件名是 00000000001073741824，依次類推就可以得到所有文件的文件名。

有了上面的命名規則，給定一個消息的偏移量，就可以根據二分查找快速找到消息所在的文件，并且用消息偏移量減去文件名就可以得到消息在文件中的偏移量。

RocketMQ 寫 CommitLog 時采用順序寫，大大提高了寫入性能。

5.2 ConsumeQueue

如果直接從 CommitLog 中檢索 Topic 中的一條消息，效率會很低，因為需要從文件的第一條消息開始依次查找。引入了 ConsumeQueue 作為 CommitLog 的索引文件，會讓檢索效率大增。

剛開始不理解 ConsumeQueue 和 MessageQueue 的區別，網上查了一些資料發現，每個ConsumeQueue對應一個上面介紹的 MessageQueue，MessageQueue 只是一個概念模型。

ConsumeQueue 中的元素內容如下：

• 前 8 個字節記錄消息在 CommitLog 中的偏移量。
• 中間 4 個字節記錄消息消息大小。
• 最后 8 個字節記錄消息中 tag 的 hashcode。

這個 tag 的作用非常重要，假如一個 Consumer 訂閱了 TopicA，Tag1 和 Tag2，那這個 Consumer 的訂閱關系如下圖：

可以看到，這個訂閱關系是一個 hash 類型的結構，key 是 Topic 名稱，value 是一個 SubscriptionData 類型的對象，這個對象封裝了 tag。

拉取消息時，首先從 Name Server 獲取訂閱關系，得到當前 Consumer 所有訂閱 tag 的 hashcode 集合 codeSet，然后從 ConsumerQueue 獲取一條記錄，判斷最后 8 個字節 tag hashcode 是否在 codeSet 中，以決定是否將該消息發送給Consumer。

5.3 Index 文件

RocketMQ 支持按照消息的屬性查找消息，為了支持這個功能，RocketMQ 引入了 Index 索引文件。Index 文件有三部分組成，文件頭 IndexHead、500萬個 hash 槽和 2000 萬個 Index 條目組成。

5.3.1 IndexHead

總共有 6 個元素組成，前兩個元素表示當前這個 Index 文件中第一條消息和最后一條消息的落盤時間，第三、第四兩個元素表示當前這個 Index 文件中第一條消息和最后一條消息在 CommitLog 文件中的物理偏移量，第五個元素表示當前這個 Index 文件中 hash 槽的數量，第六個元素表示當前這個 Index 文件中索引條目的個數。

查找的時候除了傳入 key 還需要傳入第一條消息和最后一條消息的落盤時間，這是因為 Index 文件名是時間戳命名的，傳入落盤時間可以更加精確地定位 Index 文件。

5.3.2 Hash 槽

熟悉 Java 中 HashMap 的同學應該都比較熟悉 Hash 槽這個概念了，其實就是 Hash 結構的底層數組。Index 文件中的 Hash 槽有 500 萬個數組元素，每個元素是 4 個字節 int 類型元素，保存當前槽下最新的那個 index 條目的序號。

這里 Hash 槽解決 Hash 沖突的方式是鏈表法，如下圖：

5.3.3 Index 條目

每個 Index 條目中，key 的 hashcode 占 4 個字節，phyoffset 表示消息在 CommitLog 中的物理偏移量占 8 個字節，timediff 表示消息的落盤時間與 header 里的 beginTimestamp 的差值占 4 個字節，pre index no 占 4 個字節。

pre index no 保存的是當前的 Hash 槽中前一個 index 條目的序號，一般在 key 發生 Hash 沖突時才會有值，否則這個值就是 0，表示當前元素是 Hash 槽中第一個元素。

Index 條目中保存 timediff，是為了防止 key 重復。查找 key 時，在 key 相同的情況下， 如果傳入的時間范圍跟 timediff 不滿足，則會查找 pre index no 這個條目。

5.3.4 本節總結

通過上面的分析，我們可以總結一個通過 key 在 Index 文件中查找消息的流程，如下圖：

計算 key 的 hashcode;

根據 hashcode 在 Hash 槽中查找位置 s;

計算 Hash 槽在 Index 文件中位置 40+(s-1)*4;

讀取這個槽的值，也就是Index條目序號 n;

計算該 index 條目在 Index 文件中的位置，公式：40 + 500萬 * 4 + (n-1) * 20;

讀取這個條目，比較 key 的 hashcode 和 index 條目中 hashcode是否相同，以及 key 傳入的時間范圍跟 Index 條目中的 timediff 是否匹配。如果條件不符合，則查找 pre index no 這個條目，找到后，從 CommitLog 中取出消息。

6、刷盤策略

Rocket MQ 采用靈活的刷盤策略。

6.1 異步刷盤

消息寫入 CommitLog 時，并不會直接寫入磁盤，而是先寫入PageCache 緩存中，然后用后臺線程異步把消息刷入磁盤。異步刷盤策略就是消息寫入 PageCache 后立即返回成功，這樣寫入效率非常高。如果能容忍消息丟失，異步刷盤是最好的選擇。

6.2 同步刷盤

即使同步刷盤，RocketMQ 也不是每條消息都要刷盤，線程將消息寫入內存后，會請求刷盤線程進行刷盤，但是刷盤線程并不會只把當前請求的消息刷盤，而是會把待刷盤的消息一同刷盤。同步刷盤策略保證了消息的可靠性，但是也降低了吞吐量，增加了延遲。

7、總結

本文用 7 張圖總結了 RocketMQ 的核心知識，希望能帶你快速入門。