Rocket MQ 背景

至 2010 年，隨著阿里業(yè)務(wù)的快速發(fā)展，急需一款支持順序消息，擁有海量消息堆積能力的消息 Meta Q 1.0 在 2011 年誕生。

到 2012 年， Meta Q 已經(jīng)發(fā)展到了 Meta Q 3.0 ，并抽象出了通用的消息引擎 Rocket MQ。隨后，將 Rocket MQ 進(jìn)行了開源。

到 2016 年， Meta Q 在當(dāng)年雙十一承載了萬億級(jí)消息的流轉(zhuǎn)，跨越了一個(gè)新的里程碑，同時(shí) Rocket MQ 進(jìn)入 Apache 孵化。

圖片

Rocket MQ 是什么

• 是一個(gè)隊(duì)列模型的消息中間件，具有高性能、高可靠、高實(shí)時(shí)、分布式特點(diǎn)
• Producer、Consumer、隊(duì)列都可以分布式
• Producer 向一些隊(duì)列輪流發(fā)送消息，隊(duì)列集合稱為 Topic，Consumer 如果做廣播消費(fèi)，則一個(gè) Consumer 實(shí)例消費(fèi)這個(gè) Topic 對(duì)應(yīng)的所有隊(duì)列，如果做集群消費(fèi)，則多個(gè) Consumer 實(shí)例平均消費(fèi)這個(gè) Topic 對(duì)應(yīng)的隊(duì)列集合
• 能夠保證嚴(yán)格的消息順序
• 提供豐富的消息拉取模式
• 高效的訂閱者水平擴(kuò)展能力
• 實(shí)時(shí)的消息訂閱機(jī)制
• 億級(jí)消息堆積能力
• 較少的依賴

Rocket MQ 術(shù)語

Topic：標(biāo)識(shí)一類消息的邏輯名字，消息的邏輯管理單位。無論消息生產(chǎn)還是消費(fèi)，都需要指定 Topic。

Tag：Rocket MQ 支持給在發(fā)送的時(shí)候給 Topic 打 Tag，同一個(gè) Topic 的消息雖然邏輯管理是一樣的。但是消費(fèi) Topic1 的時(shí)候，如果你訂閱的時(shí)候指定的是 Tag A，那么 Tag B 的消息將不會(huì)投遞。

Message Queue：簡稱 Q。一個(gè) Topic 將有若干個(gè) Q。若 Topic 同時(shí)創(chuàng)建在不通的 Broker，則不同的 Broker 上都有若干 Q，消息將物理地存儲(chǔ)落在不同 Broker 結(jié)點(diǎn)上，具有水平擴(kuò)展的能力。

Offset：可以認(rèn)為一條邏輯的 Message Queue 是無限長的數(shù)組。一條消息進(jìn)來下標(biāo)就會(huì)漲 1，而這個(gè)數(shù)組的下標(biāo)就是 Offset。

Max Offset：這個(gè) Offset 實(shí)際上是最新消息的 Offset + 1，即：下一條消息的 Offset。

Min Offset：標(biāo)識(shí)現(xiàn)存在的最小 Offset。而由于消息存儲(chǔ)一段時(shí)間后，消費(fèi)會(huì)被物理地從磁盤刪除，Message Queue 的 Min Offset 也就對(duì)應(yīng)增長。這意味著比 Min Offset 要小的那些消息已經(jīng)不在 Broker 上了，無法被消費(fèi)。

Consumer Offset：表示的是下次拉取的 Offset 位置。

Rocket MQ 模塊

Name Server

存儲(chǔ)當(dāng)前集群所有 Brokers 信息、Topic 跟 Broker 的對(duì)應(yīng)關(guān)系。

Broker

集群最核心模塊，主要負(fù)責(zé) Topic 消息存儲(chǔ)、消費(fèi)者的消費(fèi)位點(diǎn)管理（消費(fèi)進(jìn)度）。只有 Master 才能進(jìn)行寫入操作，Slave 從 Master 中同步數(shù)據(jù)。

Producer

生產(chǎn)者。發(fā)送消息的客戶端角色。發(fā)送消息的時(shí)候需要指定 Topic。Producer 完全無狀態(tài)，可集群部署。

Consumer

消費(fèi)者，通常有兩種實(shí)現(xiàn)，分別為 Push Consumer 和 Pull Consumer，通常我們采用 Push 的方式為主， Pull 的為輔的來進(jìn)行消息的消費(fèi)。

集群部署架構(gòu)

圖片

集群工作流程

1. 啟動(dòng) Name Sever， Name Sever 起來后監(jiān)聽端口，等待 Broker、 Producer 、Consumer 連上來，相當(dāng)于一個(gè)路由控制中心。
2. Broker 啟動(dòng)，跟所有的 Name Sever 保持長連接，定時(shí)發(fā)送心跳包。心跳包中包含當(dāng)前 Broker 信息(IP + 端口等)以及存儲(chǔ)所有 Topic 信息。注冊(cè)成功后， Name Sever 集群中就有 Topic 跟 Broker 的映射關(guān)系。
3. 收發(fā)消息前，先創(chuàng)建 Topic，創(chuàng)建 Topic 時(shí)需要指定該 Topic 要存儲(chǔ)在哪些Broker 上。也可以在發(fā)送消息時(shí)自動(dòng)創(chuàng)建 Topic。
4. Producer 發(fā)送消息，啟動(dòng)時(shí)先跟 Name Sever 集群中的其中一臺(tái)建立長連接，并從 Name Sever 中獲取當(dāng)前發(fā)送的 Topic 存在哪些 Broker 上，然后跟對(duì)應(yīng)的Broker 建長連接，直接向 Broker 發(fā)消息。
5. Consumer 跟 Producer 類似。跟其中一臺(tái) Name Sever 建立長連接，獲取當(dāng)前訂閱 Topic 存在哪些 Broker，然后直接跟 Broker 建立連接通道，開始消費(fèi)消息。

模塊功能特性

Name Sever

1. Name Sever 用于存儲(chǔ) Topic、Broker 關(guān)系信息，功能簡單，穩(wěn)定性高。多個(gè) Name Sever 之間相互沒有通信，單臺(tái) Name Sever 宕機(jī)不影響其他 Name Sever 與集群；即使整個(gè) Name Sever 集群宕機(jī)，已經(jīng)正常工作的 Producer，Consumer，Broker 仍然能正常工作，但新起的 Producer， Consumer，Broker 就無法工作。
2. Name Sever 壓力不會(huì)太大，平時(shí)主要開銷是在維持心跳和提供 Topic-Broker 的關(guān)系數(shù)據(jù)。但有一點(diǎn)需要注意，Broker 向 Name Sever 發(fā)心跳時(shí)，會(huì)帶上當(dāng)前自己所負(fù)責(zé)的所有 Broker 信息，如果 Topic 個(gè)數(shù)太多（萬級(jí)別），會(huì)導(dǎo)致一次心跳中，就 Topic 的數(shù)據(jù)就 幾十M，網(wǎng)絡(luò)情況差的話，網(wǎng)絡(luò)傳輸失敗，心跳失敗，導(dǎo)致 Name Sever 誤認(rèn)為 Broker 心跳失敗。

Broker

高并發(fā)讀寫服務(wù)

1. 消息順序?qū)懀核?Topic 數(shù)據(jù)同時(shí)只會(huì)寫一個(gè)文件，一個(gè)文件滿 1G，再寫新文件，真正的順序?qū)懕P，使得發(fā)消息 TPS 大幅提高。
2. 消息隨機(jī)讀：Rocket MQ 盡可能讓讀命中系統(tǒng) Page Cache，因?yàn)椴僮飨到y(tǒng)訪問 Page Cache 時(shí)，即使只訪問 1K 的消息，系統(tǒng)也會(huì)提前預(yù)讀出更多的數(shù)據(jù)，在下次讀時(shí)就可能命中 Page Cache ，減少 IO 操作。

負(fù)載均衡與動(dòng)態(tài)伸縮

• Topic 維度：假如一個(gè) Topic 的消息量特別大，但集群水位壓力還是很低，就可以擴(kuò)大該 Topic 的隊(duì)列數(shù)，Topic 的隊(duì)列數(shù)跟發(fā)送、消費(fèi)速度成正比。
• Broker 維度：如果集群水位很高了，需要擴(kuò)容，直接加機(jī)器部署 Broker 就可以。Broker 起來后向 Name Sever 注冊(cè)，Producer、Consumer 通過 Name Sever 發(fā)現(xiàn)新 Broker，立即跟該 Broker 直連，收發(fā)消息。

1. 負(fù)載均衡：Broker 上存 Topic 信息，Topic 由多個(gè)隊(duì)列組成，隊(duì)列會(huì)平均分散在多個(gè) Broker 上，而 Producer 的發(fā)送機(jī)制保證消息盡量平均分布到所有隊(duì)列中，最終效果就是所有消息都平均落在每個(gè) Broker 上。
2. 動(dòng)態(tài)伸縮能力（非順序消息）：Broker 的伸縮性體現(xiàn)在兩個(gè)維度：Topic， Broker。

高可用&高可靠

3. 高可用：集群部署時(shí)一般都為主備，備機(jī)實(shí)時(shí)從主機(jī)同步消息，如果其中一個(gè)主機(jī)宕機(jī)，備機(jī)提供消費(fèi)服務(wù)，但不提供寫服務(wù)。
4. 高可靠：所有發(fā)往 Broker 的消息，有同步刷盤和異步刷盤機(jī)制；同步刷盤時(shí)，消息寫入物理文件才會(huì)返回成功，異步刷盤時(shí)，只有機(jī)器宕機(jī)，才會(huì)產(chǎn)生消息丟失，Broker 掛掉可能會(huì)發(fā)生，但是機(jī)器宕機(jī)崩潰是很少發(fā)生的，除非突然斷電。

Broker 與 Name Sever 的心跳機(jī)制

單個(gè) Broker 跟所有 Name Sever 保持心跳請(qǐng)求，心跳間隔為30秒，心跳請(qǐng)求中包括當(dāng)前 Broker 所有的 Topic 信息。Name Sever 會(huì)反查 Broker 的心跳信息，如果某個(gè) Broker 在2分鐘之內(nèi)都沒有心跳，則認(rèn)為該 Broker 下線，調(diào)整 Topic 跟 Broker 的對(duì)應(yīng)關(guān)系。但此時(shí) Name Sever 不會(huì)主動(dòng)通知 Producer、Consumer 有 Broker 宕機(jī)。

Broker 刷盤策略

刷盤策略：Rocket MQ 的所有消息都是持久化的，先寫入系統(tǒng) Page Cache，然后刷盤，可以保證內(nèi)存不磁盤都有一份數(shù)據(jù)，訪問時(shí)，直接從內(nèi)存讀取。

異步刷盤： 舉例：現(xiàn)有 SAS 15000 轉(zhuǎn)磁盤測試順序?qū)懳募?，速度可以達(dá)到 300M 每秒左右，而線上的網(wǎng)卡一般都為千兆網(wǎng)卡，速度最快可達(dá) 128M 每秒， 寫磁盤速度明顯快于數(shù)據(jù)網(wǎng)絡(luò)入口速度，那么刷盤的迕度肯定可以跟上消息的寫入速度。

同步刷盤：和異步的唯一區(qū)別是異步寫完 Page Cache 直接返回，而同步是等待刷盤完成之后再返回。

圖片

Producer

1. Producer 啟動(dòng)時(shí)，也需要指定 Name Sever 的地址，從 Name Sever 集群中選一臺(tái)建立長連接。如果該 Name Sever 宕機(jī)，會(huì)自動(dòng)連其他 Name Sever 。直到有可用的 Name Sever 為止。
2. 心跳檢測和 Broker 類似。
3. Producer 端，每個(gè)實(shí)例在發(fā)消息的時(shí)候，默認(rèn)會(huì)輪詢所有的 Message Queue 發(fā)送，以達(dá)到讓消息平均落在不同的 Queue 上。而由于 Queue 可以散落在不同的 Broker，所以消息就發(fā)送到不同的 Broker 下。當(dāng)然了，可以發(fā)送消息到指定的隊(duì)列中。比如根據(jù)店家的編號(hào) Hash 到不同的隊(duì)列中，形成有序的消息。如果想自己實(shí)現(xiàn)發(fā)送的策略，可以實(shí)現(xiàn) MessageQueueSelector 這個(gè)接口。

圖片

Consumer

1. 消費(fèi)者啟動(dòng)時(shí)需要指定 Name Sever 地址，與其中一個(gè) Name Sever 建立長連接。消費(fèi)者每隔 30 秒從 Name Sever 獲取所有 Topic 的最新隊(duì)列情況，這意味著某個(gè) Broker 如果宕機(jī)，客戶端最多要 30 秒才能感知。連接建立后，從 Name Sever 中獲取當(dāng)前消費(fèi) Topic 所涉及的 Broker，直連 Broker。
2. 和 Name Sever 的心跳檢測和 Broker 類似。
3. 消費(fèi)者消費(fèi)模式

• 集群消費(fèi)：一個(gè) Consumer Group 中的各個(gè) Consumer 實(shí)例分?jǐn)側(cè)ハM(fèi)消息，即一條消息只會(huì)投遞到一個(gè) Consumer Group 下面的一個(gè)實(shí)例。
• 廣播消費(fèi)：消息將對(duì)一個(gè) Consumer Group 下的各個(gè) Consumer 實(shí)例都投遞一遍。即即使這些 Consumer 屬于同一個(gè) Consumer Group，消息也會(huì)被 Consumer Group 中的每個(gè) Consumer 都消費(fèi)一次。

4. 負(fù)載均衡：消費(fèi)者端的負(fù)載均衡，就是集群消費(fèi)模式下，同一個(gè) ID 的所有消費(fèi)者實(shí)例平均消費(fèi)該 Topic 的所有隊(duì)列。
5. 拉取流程：Consumer 端每隔一段時(shí)間主動(dòng)向 Broker 發(fā)送拉消息請(qǐng)求，Broker 在收到 Pull 請(qǐng)求后，如果有消息就立即返回?cái)?shù)據(jù)，Consumer 端收到返回的消息后，再回調(diào)消費(fèi)者設(shè)置的 Listener 方法。如果 Broker 在收到 Pull 請(qǐng)求時(shí)，消息隊(duì)列里沒有數(shù)據(jù)，Broker 端會(huì)阻塞請(qǐng)求直到有數(shù)據(jù)傳遞或超時(shí)才返回。

圖片

持久化 Commit Log

雖然每個(gè) Topic 下面有很多 Message Queue，但是 Message Queue 本身并不存儲(chǔ)消息。真正的消息存儲(chǔ)會(huì)寫在 Commit Log 的文件，Message Queue 只是存儲(chǔ) Commit Log 中對(duì)應(yīng)的位置信息，方便通過 Message Queue 找到對(duì)應(yīng)存儲(chǔ)在 Commit Log 的消息。不同的 Topic，Message Queue 都是寫到相同的 Commit Log 文件，也就是說 Commit Log 完全的順序?qū)憽?/p>

圖片

什么時(shí)候清理物理消息文件

消息存儲(chǔ)在 Commit Log 之后，的確是會(huì)被清理的，但是這個(gè)清理只會(huì)在以下任一條件成立才會(huì)批量刪除消息文件（Commit Log）：

1. 消息文件過期（默認(rèn)72小時(shí)），且到達(dá)清理時(shí)點(diǎn)（默認(rèn)是凌晨4點(diǎn)），刪除過期文件。
2. 消息文件過期（默認(rèn)72小時(shí)），且磁盤空間達(dá)到了水位線（默認(rèn)75%），刪除過期文件。
3. 磁盤已經(jīng)達(dá)到必須釋放的上限（85%水位線）的時(shí)候，則開始批量清理文件（無論是否過期），直到空間充足。

注意：若磁盤空間達(dá)到危險(xiǎn)水位線（默認(rèn)90%），出于保護(hù)自身的目的，Broker 會(huì)拒絕寫入服務(wù)。

參考文獻(xiàn)

• Rocket MQ 用戶指南 v3.2.4
• 阿里云社區(qū)

https://yq.aliyun.com/articles/69647?spm=5176.100240.searchblog.7.ZgGuDF

https://yq.aliyun.com/articles/66101?spm=5176.100240.searchblog.101.s7dvlU

https://yq.aliyun.com/articles/66110?spm=5176.100239.blogcont66101.23.kpzm2R

• 其他資料
• http://blog.csdn.net/damacheng/article/details/42846549
• http://jameswxx.iteye.com/blog/2091966