前言

順序消息是指對于一個(gè)指定的 Topic ，消息嚴(yán)格按照先進(jìn)先出（FIFO）的原則進(jìn)行消息發(fā)布和消費(fèi)，即先發(fā)布的消息先消費(fèi)，后發(fā)布的消息后消費(fèi)。

順序消息分為分區(qū)順序消息和全局順序消息。

1、分區(qū)順序消息

對于指定的一個(gè) Topic ，所有消息根據(jù) Sharding Key 進(jìn)行區(qū)塊分區(qū)，同一個(gè)分區(qū)內(nèi)的消息按照嚴(yán)格的先進(jìn)先出（FIFO）原則進(jìn)行發(fā)布和消費(fèi)。同一分區(qū)內(nèi)的消息保證順序，不同分區(qū)之間的消息順序不做要求。

• 適用場景：適用于性能要求高，以 Sharding Key 作為分區(qū)字段，在同一個(gè)區(qū)塊中嚴(yán)格地按照先進(jìn)先出（FIFO）原則進(jìn)行消息發(fā)布和消費(fèi)的場景。
• 示例：電商的訂單創(chuàng)建，以訂單 ID 作為 Sharding Key ，那么同一個(gè)訂單相關(guān)的創(chuàng)建訂單消息、訂單支付消息、訂單退款消息、訂單物流消息都會按照發(fā)布的先后順序來消費(fèi)。

2、全局順序消息

對于指定的一個(gè) Topic ，所有消息按照嚴(yán)格的先入先出（FIFO）的順序來發(fā)布和消費(fèi)。

• 適用場景：適用于性能要求不高，所有的消息嚴(yán)格按照 FIFO 原則來發(fā)布和消費(fèi)的場景。
• 示例：在證券處理中，以人民幣兌換美元為 Topic，在價(jià)格相同的情況下，先出價(jià)者優(yōu)先處理，則可以按照 FIFO 的方式發(fā)布和消費(fèi)全局順序消息。

全局順序消息實(shí)際上是一種特殊的分區(qū)順序消息，即 Topic 中只有一個(gè)分區(qū)，因此全局順序和分區(qū)順序的實(shí)現(xiàn)原理相同。

因?yàn)榉謪^(qū)順序消息有多個(gè)分區(qū)，所以分區(qū)順序消息比全局順序消息的并發(fā)度和性能更高。

消息的順序需要由兩個(gè)階段保證：

消息發(fā)送

如上圖所示，A1、B1、A2、A3、B2、B3 是訂單 A 和訂單 B 的消息產(chǎn)生的順序，業(yè)務(wù)上要求同一訂單的消息保持順序，例如訂單 A 的消息發(fā)送和消費(fèi)都按照 A1、A2、A3 的順序。

如果是普通消息，訂單A 的消息可能會被輪詢發(fā)送到不同的隊(duì)列中，不同隊(duì)列的消息將無法保持順序，而順序消息發(fā)送時(shí) RocketMQ 支持將 Sharding Key 相同（例如同一訂單號）的消息序路由到一個(gè)隊(duì)列中。

消息消費(fèi)

消費(fèi)者消費(fèi)消息時(shí)，需要保證消息消費(fèi)順序和存儲順序一致，最終實(shí)現(xiàn)消費(fèi)順序和發(fā)布順序的一致。

我們知道負(fù)載均衡服務(wù)是客戶端開始消費(fèi)的起點(diǎn)。在負(fù)載均衡階段，并發(fā)消費(fèi)和順序消費(fèi)并沒有什么大的差別，最大的差別在于：向 Borker 申請鎖 。

消費(fèi)者根據(jù)分配的隊(duì)列 messageQueue ，向 Borker 申請鎖 ，如果申請成功，則會拉取消息，如果失敗，則定時(shí)任務(wù)每隔20秒會重新嘗試。

見上圖，順序消費(fèi)核心流程如下：

1、 組裝成消費(fèi)對象

消費(fèi)快照 processQueue + 消息隊(duì)列對象 MessageQueue

2、 將請求對象提交到消費(fèi)線程池

和并發(fā)消費(fèi)不同的是，這里的消費(fèi)請求包含消費(fèi)快照 processQueue ，消息隊(duì)列 messageQueue 兩個(gè)對象，并不對消息列表做任何處理。

3、 消費(fèi)線程內(nèi)，對消費(fèi)隊(duì)列加鎖

4、 從消費(fèi)快照中取得待消費(fèi)的消息列表

消費(fèi)快照 processQueue 對象里，創(chuàng)建了一個(gè)紅黑樹對象 consumingMsgOrderlyTreeMap 用于臨時(shí)存儲的待消費(fèi)的消息。

5、 執(zhí)行消息監(jiān)聽器

執(zhí)行監(jiān)聽器邏輯容易理解，消費(fèi)快照的消費(fèi)鎖 consumeLock的作用是：防止 Rebalance 線程把當(dāng)前消費(fèi)的 MessageQueue 對象移除掉。

6、 處理消費(fèi)結(jié)果

消費(fèi)成功時(shí)，首先計(jì)算需要提交的偏移量，然后更新本地消費(fèi)進(jìn)度。

消費(fèi)失敗時(shí)，分兩種場景：

• 假如已消費(fèi)次數(shù)小于最大重試次數(shù)，則將放入對象 consumingMsgOrderlyTreeMap 用例臨時(shí)存儲的待消費(fèi)的消息，重新加入到消費(fèi)快照紅黑樹 msgTreeMap中，然后使用定時(shí)任務(wù)嘗試重新消費(fèi)。
• 假如已消費(fèi)次數(shù)大于等于最大重試次數(shù)，則將失敗消息發(fā)送到 Broker ，Broker 接收到消息后，會加入到死信隊(duì)列里 , 最后計(jì)算需要提交的偏移量，然后更新本地消費(fèi)進(jìn)度。

我們做一個(gè)關(guān)于順序消費(fèi)的總結(jié)：

1. 順序消費(fèi)需要由兩個(gè)階段消息發(fā)送和消息消費(fèi)協(xié)同配合，底層支撐依靠的是 RocketMQ 的存儲模型；
2. 順序消費(fèi)服務(wù)啟動后，通過三把鎖的機(jī)制，消息隊(duì)列 messageQueue 的數(shù)據(jù)都會被消費(fèi)者實(shí)例單線程的執(zhí)行消費(fèi)；
3. 假如消費(fèi)者擴(kuò)容，消費(fèi)者重啟，或者 Broker 宕機(jī) ，順序消費(fèi)也會有一定幾率較短時(shí)間內(nèi)亂序，所以消費(fèi)者的業(yè)務(wù)邏輯還是要保障冪等。

到此這篇關(guān)于RocketMQ的順序消費(fèi)機(jī)制詳解的文章就介紹到這了,更多相關(guān)RocketMQ的順序消費(fèi)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論