今天給大家分享一篇我對(duì)學(xué)習(xí) RocketMQ 系統(tǒng)架構(gòu)核心知識(shí)點(diǎn)的梳理和總結(jié),在講解時(shí)力求精簡(jiǎn)、通俗易懂，通過圖解來給正在學(xué)習(xí) RocketMQ 的小伙伴帶來幫助。RocketMQ 是阿里巴巴的分布式消息中間件，在 2012 年開源，在 2017 年成為 Apache 頂級(jí)項(xiàng)目。

1 集群架構(gòu)

RocketMQ 的集群架構(gòu)如下圖：

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

1.1 Name Server 集群

Name Server 集群部署，但是節(jié)點(diǎn)之間并不會(huì)同步數(shù)據(jù)，因?yàn)槊總€(gè)節(jié)點(diǎn)都會(huì)保存完整的數(shù)據(jù)。因此單個(gè)節(jié)點(diǎn)掛掉，并不會(huì)對(duì)集群產(chǎn)生影響。

1.2 Broker

Broker 采用主從集群，實(shí)現(xiàn)多副本存儲(chǔ)和高可用。每個(gè) Broker 節(jié)點(diǎn)都要跟所有的 Name Server 節(jié)點(diǎn)建立長連接，定義注冊(cè) Topic 路由信息和發(fā)送心跳。

跟所有 Name Server 建立連接，就不會(huì)因?yàn)閱蝹€(gè) Name Server 掛了影響 Broker 使用。Broker 主從模式中， Slave 節(jié)點(diǎn)主動(dòng)從 Master 節(jié)點(diǎn)拉取消息。

1.3 Producer

Producer 跟 Name Server 的任意一個(gè)節(jié)點(diǎn)建立長連接，定期從 Name Server 拉取 Topic 路由信息。Producer 是否采用集群，取決于它所在的業(yè)務(wù)系統(tǒng)。

1.4 Consumer

Consumer 跟 Name Server 的任意一個(gè)節(jié)點(diǎn)建立長連接，定期從 Name Server 拉取 Topic 路由信息。Consumer 是否采用集群，取決于它所在的業(yè)務(wù)系統(tǒng)。

Producer 和 Consumer 只跟任意一個(gè) Name Server 節(jié)點(diǎn)建立連接，因?yàn)?Broker 會(huì)向所有 Name Server 注冊(cè) Topic 信息，所以每個(gè) Name Server 保存的數(shù)據(jù)其實(shí)是一致的。

2 MessageQueue

Producer 發(fā)送的消息會(huì)在 Broker 的 MessageQueue 中保存，如下圖：

有了 MessageQueue ，Topic 就可以在 Broker 中實(shí)現(xiàn)分布式存儲(chǔ)，如上圖，Broker 集群中保存了 4 個(gè)MessageQueue，這些 MessageQueue 保存了 Topic1-Topic3 這三個(gè) Topic 的消息。

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

默認(rèn) Topic 可以創(chuàng)建的 MessageQueue 數(shù)量是 4，Broker 可以創(chuàng)建的 MessageQueue 數(shù)量是 8, RocketMQ 選擇二者中數(shù)量小的，也就是 4。不過這兩個(gè)值都可以配置。

3 Consumer

RocketMQ的消費(fèi)模式如下圖：

圖中，Topic1 的消息寫入了兩個(gè) MessageQueue，兩個(gè)隊(duì)列保存在 Broker1 和 Broker2 上。

RocketMQ 通過 Consumer Group 實(shí)現(xiàn)消息廣播。比如上圖中有兩個(gè)消費(fèi)者組，每個(gè)消費(fèi)者組有兩個(gè)消費(fèi)者。

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

4 Broker 高可用集群

Broker 集群如下圖：

Broker 通過主從集群來實(shí)現(xiàn)消息高可用。跟 Kafka 不同的是，RocketMQ 并沒有 Master 節(jié)點(diǎn)選舉功能，而是采用多 Master 多 Slave 的集群架構(gòu)。Producer 寫入消息時(shí)寫入 Master 節(jié)點(diǎn)，Slave 節(jié)點(diǎn)主動(dòng)從 Master 節(jié)點(diǎn)拉取數(shù)據(jù)來保持跟 Master 節(jié)點(diǎn)的數(shù)據(jù)一致。

Consumer 消費(fèi)消息時(shí)，既可以從 Master 節(jié)點(diǎn)拉取數(shù)據(jù)，也可以從 Slave 節(jié)點(diǎn)拉取數(shù)據(jù)。 到底是從 Master 拉取還是從 Slave 拉取取決于 Master 節(jié)點(diǎn)的負(fù)載和 Slave 的同步情況 。如果 Master 負(fù)載很高，Master 會(huì)通知 Consumer 從 Slave 拉取消息，而如果 Slave 同步消息進(jìn)度延后，則 Master 會(huì)通知 Consumer 從 Master 拉取數(shù)據(jù)?？傊?，從 Master 拉取還是從 Slave 拉取由 Master 來決定。

如果 Master 節(jié)點(diǎn)發(fā)生故障，RocketMQ 會(huì)使用基于 raft 協(xié)議的 DLedger 算法來進(jìn)行主從切換。

Broker 每隔 30s 向 Name Server 發(fā)送心跳，Name Server 如果 120s 沒有收到心跳，就會(huì)判斷 Broker 宕機(jī)了。

5 消息存儲(chǔ)

RocketMQ 的存儲(chǔ)設(shè)計(jì)是非常有創(chuàng)造性的。存儲(chǔ)文件主要有三個(gè)：CommitLog、ConsumeQueue、Index。如下圖：

5.1 CommitLog

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

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

R oc ketMQ 寫 CommitLog 時(shí)采用順序?qū)?，大大提高了寫入性能?/span>

5.2 ConsumeQueue

如果直接從 CommitLog 中檢索 Topic 中的一條消息，效率會(huì)很低，因?yàn)樾枰獜奈募牡谝粭l消息開始依次查找。引入了 ConsumeQueue 作為 CommitLog 的索引文件，會(huì)讓檢索效率大增。

剛開始不理解 ConsumeQueue 和 MessageQueue 的區(qū)別，網(wǎng)上查了一些資料發(fā)現(xiàn)，每個(gè)ConsumeQueue對(duì)應(yīng)一個(gè)上面介紹的 MessageQueue，MessageQueue 只是一個(gè)概念模型。

ConsumeQueue 中的元素內(nèi)容如下：

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

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

可以看到，這個(gè)訂閱關(guān)系是一個(gè) hash 類型的結(jié)構(gòu)，key 是 Topic 名稱，value 是一個(gè) SubscriptionData 類型的對(duì)象，這個(gè)對(duì)象封裝了 tag。

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

5.3 Index 文件

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

5.3.1 IndexHead

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

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

5.3.2 Hash 槽

熟悉 Java 中 HashMap 的同學(xué)應(yīng)該都比較熟悉 Hash 槽這個(gè)概念了，其實(shí)就是 Hash 結(jié)構(gòu)的底層數(shù)組。Index 文件中的 Hash 槽有 500 萬個(gè)數(shù)組元素，每個(gè)元素是 4 個(gè)字節(jié) int 類型元素，保存當(dāng)前槽下最新的那個(gè) index 條目的序號(hào)。

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

5.3.3 Index 條目

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

pre index no 保存的是當(dāng)前的 Hash 槽中前一個(gè) index 條目的序號(hào)，一般在 key 發(fā)生 Hash 沖突時(shí)才會(huì)有值，否則這個(gè)值就是 0，表示當(dāng)前元素是 Hash 槽中第一個(gè)元素。

In dex 條目中保存 timediff，是為了防止 key 重復(fù)。 查找 key 時(shí)，在 key 相同的情況下， 如果傳入的時(shí)間范圍跟 timediff 不滿足，則會(huì)查找 pre index no 這個(gè)條目。

5.3.4 本節(jié)總結(jié)

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

1. 計(jì)算 key 的 hashcode；
2. 根據(jù) hashcode 在 Hash 槽中查找位置 s；
3. 計(jì)算 Hash 槽在 Index 文件中位置 40+(s-1)*4；
4. 讀取這個(gè)槽的值，也就是Index條目序號(hào) n；
5. 計(jì)算該 index 條目在 Index 文件中的位置，公式：40 + 500萬 * 4 + (n-1) * 20；
6. 讀取這個(gè)條目，比較 key 的 hashcode 和 index 條目中 hashcode是否相同，以及 key 傳入的時(shí)間范圍跟 Index 條目中的 timediff 是否匹配。如果條件不符合，則查找 pre index no 這個(gè)條目，找到后，從 CommitLog 中取出消息。

6 刷盤策略

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

6.1 異步刷盤

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

6.2 同步刷盤

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

7 總結(jié)

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