目錄

1. RocketMQ介紹
2. RocketMQ概念
3. 為什么要用RocketMQ？
1. 異步解耦
2. 削峰填谷
3. 分布式事務(wù)最終一致性
4. 數(shù)據(jù)分發(fā)
4. RocketMQ架構(gòu)
5. RocketMQ消息類型
1. 普通消息
2. 順序消息
3. 定時消息
4. 事務(wù)消息
6. 最佳實(shí)踐
1. 消息重試
2. 消息過濾
3. 消費(fèi)模式
4. 消費(fèi)冪等
7. 本地事務(wù)消息封裝
8. 參考代碼

RocketMQ 介紹

Apache RocketMQ 是一款 低延遲、高并發(fā)、高可用、高可靠的分布式消息中間件。消息隊(duì)列 RocketMQ 可為分布式應(yīng)用系統(tǒng)提供異步解耦和削峰填谷的能力，同時也具備互聯(lián)網(wǎng)應(yīng)用所需的海量消息堆積、高吞吐、可靠重試等特性。

RocketMQ 概念

• Topic：消息主題，用于將一類的消息進(jìn)行歸類，比如訂單主題，就是所有訂單相關(guān)的消息都可以由這個主題去承載，生產(chǎn)者向這個主題發(fā)送消息。
• 生產(chǎn)者：負(fù)責(zé)生產(chǎn)消息并發(fā)送消息到 Topic 的角色。
• 消費(fèi)者：負(fù)責(zé)從 Topic 接收并消費(fèi)消息 的角色。
• 消息：生產(chǎn)者向 Topic 發(fā)送的內(nèi)容，會被消費(fèi)者消費(fèi)。
• 消息屬性：生產(chǎn)者發(fā)送的時候可以為消息自定義一些業(yè)務(wù)相關(guān)的屬性，比如 Message Key 和 Tag 等。
• Group：一類生產(chǎn)者或消費(fèi)者，這類生產(chǎn)者或消費(fèi)者通常生產(chǎn)或消費(fèi)同一類消息，且消息發(fā)布或訂閱的邏輯一致。

為什么要使用 RocketMQ？

異步解耦

隨著微服務(wù)架構(gòu)的流行，服務(wù)之間的關(guān)系梳理非常重要。異步解耦可以降低服務(wù)之間的耦合程度，同時也能提高服務(wù)的吞吐量。

使用異步解耦的業(yè)務(wù)場景非常多，因?yàn)槊總€行業(yè)的業(yè)務(wù)都會不太一樣，以一些比較通用的業(yè)務(wù)來說明相信大家都能理解。

比如電商行業(yè)的下單業(yè)務(wù)場景，以最簡單的下單流程來說，下單流程如下：

1. 鎖庫存
2. 創(chuàng)建訂單
3. 用戶支付
4. 扣減庫存
5. 給用戶發(fā)送購買短信通知
6. 給用戶增加積分
7. 通知商家發(fā)貨

我們以下單成功后，用戶進(jìn)行支付，支付完成會有個邏輯叫支付回調(diào)，在回調(diào)里面需要去做一些業(yè)務(wù)邏輯。首先來看下同步處理需要花費(fèi)的時間，如下圖：

上面的下單流程從 3 到 5 都是可以采用異步流程進(jìn)行處理，對于用戶來說，支付完成后他就不需要關(guān)注后面的流程了。后臺慢慢處理就行了，這樣就能簡化三個步驟，提高回調(diào)的處理時間。

削峰填谷

削峰填谷指的是在大流量的沖擊下，利用 RocketMQ 可以抗住瞬時的大流量，保護(hù)系統(tǒng)的穩(wěn)定性，提升用戶體驗(yàn)。

在電商行業(yè)，最常見的流量沖擊就是秒殺活動了，利用 RocketMQ 來實(shí)現(xiàn)一個完整的秒殺業(yè)務(wù)還是與很多需要做的工作，不在本文的范圍內(nèi)，后面有機(jī)會可以單獨(dú)跟大家聊聊。想告訴大家的是像諸如此類的場景可以利用 RocketMQ 來扛住高并發(fā)，前提是業(yè)務(wù)場景支持異步處理。

分布式事務(wù)最終一致性

眾所周知，分布式事務(wù)有 2PC，TCC，最終一致性等方案。其中使用消息隊(duì)列來做最終一致性方案是比較常用的。

在電商的業(yè)務(wù)場景中，交易相關(guān)的核心業(yè)務(wù)一定要確保數(shù)據(jù)的一致性。通過引入消息隊(duì)列 RocketMQ 版的分布式事務(wù)，既可以實(shí)現(xiàn)系統(tǒng)之間的解耦，又可以保證最終的數(shù)據(jù)一致性。

數(shù)據(jù)分發(fā)

數(shù)據(jù)分發(fā)指的是可以將原始數(shù)據(jù)分發(fā)到多個需要使用這份數(shù)據(jù)的系統(tǒng)中，實(shí)現(xiàn)數(shù)據(jù)異構(gòu)的需求。最常見的有將數(shù)據(jù)分發(fā)到 ES, Redis 中為業(yè)務(wù)提供搜索，緩存等服務(wù)。

除了手動通過消息機(jī)制進(jìn)行數(shù)據(jù)分發(fā)，還可以訂閱 Mysql 的 binlog 來分發(fā)，在分發(fā)這個場景，需要使用 RocketMQ 的順序消息來保證數(shù)據(jù)的一致性。

RocketMQ 架構(gòu)

• Name Server：是一個幾乎無狀態(tài)節(jié)點(diǎn)，可集群部署，在消息隊(duì)列 RocketMQ 版中提供命名服務(wù)，更新和發(fā)現(xiàn) Broker 服務(wù)。就是一個注冊中心。
• Broker：消息中轉(zhuǎn)角色，負(fù)責(zé)存儲消息，轉(zhuǎn)發(fā)消息。分為 Master Broker 和 Slave Broker，一個 Master Broker 可以對應(yīng)多個 Slave Broker，但是一個 Slave Broker 只能對應(yīng)一個 Master Broker。Broker 啟動后需要完成一次將自己注冊至 Name Server 的操作；隨后每隔 30s 定期向 Name Server 上報(bào) Topic 路由信息。
• 生產(chǎn)者：與 Name Server 集群中的其中一個節(jié)點(diǎn)（隨機(jī)）建立長鏈接（Keep-alive），定期從 Name Server 讀取 Topic 路由信息，并向提供 Topic 服務(wù)的 Master Broker 建立長鏈接，且定時向 Master Broker 發(fā)送心跳。
• 消費(fèi)者：與 Name Server 集群中的其中一個節(jié)點(diǎn)（隨機(jī)）建立長連接，定期從 Name Server 拉取 Topic 路由信息，并向提供 Topic 服務(wù)的 Master Broker、Slave Broker 建立長連接，且定時向 Master Broker、Slave Broker 發(fā)送心跳。Consumer 既可以從 Master Broker 訂閱消息，也可以從 Slave Broker 訂閱消息，訂閱規(guī)則由 Broker 配置決定。

RocketMQ 消息類型

RocketMQ 支持豐富的消息類型，可以滿足多場景的業(yè)務(wù)需求。不同的消息有不同的應(yīng)用場景，下面為大家介紹常用的四種消息類型。

普通消息

普通消息是指 RocketMQ 中無特性的消息。當(dāng)沒有特殊的業(yè)務(wù)場景，使用普通消息就夠了。如果有特殊的場景，就可以使用特殊的消息類型，比如順序，事務(wù)等。

同步發(fā)送

同步發(fā)送：消息發(fā)送方發(fā)送出去一條消息，會同步得到服務(wù)端返回的結(jié)果。

異步發(fā)送

異步發(fā)送：消息發(fā)送方發(fā)出去一條消息，不用等待服務(wù)端返回結(jié)果，可以接著發(fā)送下一條消息。發(fā)送方可以通過回調(diào)接口接收服務(wù)端響應(yīng)，并處理響應(yīng)結(jié)果。

單向發(fā)送

單向發(fā)送：消息發(fā)送方只負(fù)責(zé)發(fā)送消息，發(fā)送出去后就不管了，這種方式發(fā)送速度非?？?，存在丟失消息的風(fēng)險(xiǎn)。

順序消息

順序消息是指生產(chǎn)者按照一定的先后順序發(fā)布消息；消費(fèi)者按照既定的先后順序訂閱消息，即先發(fā)布的消息一定會先被消費(fèi)者接收到。

比如數(shù)據(jù)分發(fā)的場景，如果我們訂閱了 Mysql 的 binlog 來進(jìn)行數(shù)據(jù)異構(gòu)。消息要是沒有順序，就會出現(xiàn)數(shù)據(jù)錯亂問題。

比如新增一條 id=1 的數(shù)據(jù)，然后馬上刪除。這樣就產(chǎn)生了兩條消息。正常的消費(fèi)順序是先新增，然后刪除，此時數(shù)據(jù)是沒有的。如果消息沒有順序，刪除的先被消費(fèi)了，然后消費(fèi)新增的，此時數(shù)據(jù)還在，沒被刪除掉，就會導(dǎo)致不一致。

定時消息

定時消息是指消息具備定時發(fā)送的功能，當(dāng)消息發(fā)送到服務(wù)端后，不會立即投遞給消費(fèi)者。而是要等到消息指定的時間后才會投遞給消費(fèi)者進(jìn)行消費(fèi)。

延遲消息也就是定時消息，定時消息是定在某個時間點(diǎn)進(jìn)行發(fā)送，比如 2020-11-11 12:00:00 發(fā)送。

延遲消息一般是在當(dāng)前發(fā)送時間的基礎(chǔ)上延遲多久進(jìn)行發(fā)送，比如當(dāng)前時間是 2020-09-10 12:00:00，延遲 10 分鐘，那么消息發(fā)送成功后將在 2020-09-10 12:10:00 進(jìn)行投遞給消費(fèi)者。

定時消息可以在訂單超時未支付自動取消等場景使用。

事務(wù)消息

RocketMQ 提供類似 X/Open XA 的分布式事務(wù)功能，通過 RocketMQ 事務(wù)消息能達(dá)到分布式事務(wù)的最終一致。

交互流程：

1. 發(fā)送方首先發(fā)送半事務(wù)消息到 RocketMQ 服務(wù)端。

2. RocketMQ 服務(wù)端接收到消息，然后將消息持久化成功之后，向發(fā)送方返回 Ack 確認(rèn)消息已經(jīng)發(fā)送成功，此時消息為半事務(wù)消息，不會投遞給消費(fèi)方。

3. 收到半事務(wù)消息的 Ack 后，發(fā)送方開始執(zhí)行本地事務(wù)邏輯。

4. 發(fā)送方根據(jù)本地事務(wù)執(zhí)行結(jié)果向服務(wù)端提交二次確認(rèn)，如果本地事務(wù)執(zhí)行成則進(jìn)行消息的 Commit，如果執(zhí)行失敗則進(jìn)行消息的 Rollback，服務(wù)端收到 Commit 狀態(tài)則將半事務(wù)消息標(biāo)記為可投遞，消費(fèi)方最終將收到該消息；服務(wù)端收到 Rollback 狀態(tài)則刪除半事務(wù)消息，消費(fèi)方將不會收到該消息。

5. 如果出現(xiàn)意外情況，步驟 4 沒有進(jìn)行消息的二次確認(rèn)，等待固定時間后服務(wù)端將對該消息發(fā)起消息回查。

6. 發(fā)送方收到消息回查后，需要檢查對應(yīng)消息的本地事務(wù)執(zhí)行的最終結(jié)果。發(fā)送方根據(jù)檢查得到的本地事務(wù)的最終狀態(tài)再次提交二次確認(rèn)，服務(wù)端仍按照步驟 4 對半事務(wù)消息進(jìn)行操作。

最佳實(shí)踐

消息重試

消息在消費(fèi)方消費(fèi)失敗后，RocketMQ 服務(wù)端會重新進(jìn)行消息的投遞，知道消費(fèi)者成功消費(fèi)消息，當(dāng)然重試有次數(shù)限制，默認(rèn) 16 次。

消息重試在一定程度上保證了消息不丟失，通過重試來達(dá)到最終被消費(fèi)的目的。需要注意的是消費(fèi)者在消費(fèi)的時候一定要等本地業(yè)務(wù)成功后才能進(jìn)行 ACK(消費(fèi)確認(rèn))，不然就會出現(xiàn)消費(fèi)失敗，但是已經(jīng) ACK，消息將不會重復(fù)投遞。

如果采取異步消費(fèi)的方式，需要進(jìn)行異步轉(zhuǎn)同步，等異步操作完才進(jìn)行 ACK，具體可以參考我之前寫的一篇文章https://mp.weixin.qq.com/s/Bbh1GDpmkLhZhw5f0POJ2A。

最后需要做好對應(yīng)的監(jiān)控，如果重試了 4，5 次還是失敗的，基本上后面重試也是失敗的。這個時候需要讓開發(fā)人員知道，該人工處理的就人工介入?；蛘咧苯颖O(jiān)控死信隊(duì)列。

消息過濾

消息主題，一般用于一類消息的統(tǒng)一分類。比如訂單主題，但是訂單下的消息會分為很多種。比如創(chuàng)建訂單，取消訂單等。

不同類型的消息有不同的業(yè)務(wù)處理，我們可以統(tǒng)一定義消息格式，然后通過一個字段去區(qū)分消息類型來做不同的業(yè)務(wù)邏輯。不好的點(diǎn)在于所有消息都會推送到消費(fèi)方，不能按需消費(fèi)。

在 RocketMQ 中可以給消息指定 tag，通過 tag 來區(qū)分消息類型。消費(fèi)者可以根據(jù) Tag 在 RocketMQ 服務(wù)端完成消息過濾，以確保消費(fèi)者最終只消費(fèi)到其關(guān)注的消息類型。

我曾經(jīng)遇到過一個 tag 沒有正確使用的方式，只有一個 MQ 實(shí)例，用 tag 來區(qū)分環(huán)境。所有消息都在一個主題中，測試環(huán)境消費(fèi)測試環(huán)境的 tag，線上消費(fèi)線上的 tag。

這種方式的問題在于消息沒做隔離，線上線下的消息都在一起。另一個就是 tag 被固定成了環(huán)境的區(qū)分，無法用于消息類型場景，導(dǎo)致只能建多個 topic 來承載多個業(yè)務(wù)消息類型。

消費(fèi)模式

RocketMQ 消費(fèi)模式有兩種，集群消費(fèi)和廣播消費(fèi)。

集群消費(fèi)：

消費(fèi)者部署了多個實(shí)例我們稱之為一個集群，集群消費(fèi)只會被其中的某一個實(shí)例進(jìn)行消費(fèi)。

適合大部分的業(yè)務(wù)場景，大部分的場景我們的消息只允許被消費(fèi)一次，而且只能有一個消費(fèi)者去消費(fèi)，比如支付回調(diào)場景，如果一個消息被多個實(shí)例同時消費(fèi)，那么就會出現(xiàn)同時去修改訂單狀態(tài)，同時去扣減庫存的情況。

廣播消費(fèi)：

廣播消費(fèi)會讓集群中每個實(shí)例都消費(fèi)一次。

比如我們使用了本地緩存，當(dāng)數(shù)據(jù)變更的時候，我們需要刷新每個節(jié)點(diǎn)本地的緩存，所以每個節(jié)點(diǎn)都需要收到消息。

消費(fèi)冪等

冪等問題，無論是在 API 請求場景還是在消息消費(fèi)場景，都會遇到。一條消息不能重復(fù)消費(fèi)多次這個肯定是要保證的，因?yàn)槲覀儾荒鼙ＷC消息發(fā)送方不發(fā)送多次，也不能保證消息不重復(fù)投遞。

RocketMQ 的 Exactly-Once 投遞語義，就是用于解決冪等問題。Exactly-Once 是指發(fā)送到消息系統(tǒng)的消息只能被消費(fèi)端處理且僅處理一次，即使生產(chǎn)端重試消息發(fā)送導(dǎo)致某消息重復(fù)投遞，該消息在消費(fèi)端也只被消費(fèi)一次。

最佳的冪等處理方式還是需要有一個唯一的業(yè)務(wù)標(biāo)識，雖然每條消息都有 MessageId，但是不建議用 MessageId 來做冪等判斷，在發(fā)送消息的時候，可以為每條消息設(shè)置一個 MessageKey，這個 MessageKey 就可以用來做業(yè)務(wù)的唯一標(biāo)識。

關(guān)于冪等怎么處理，就不細(xì)講了?？梢詤⒖嘉抑皩懙囊黄恼耯ttps://mp.weixin.qq.com/s/9fhqnbeXPz7-7x0Eadd8DA，通用的冪等實(shí)現(xiàn)方案。

本地事務(wù)消息封裝

上面介紹了事務(wù)消息，RocketMQ 的事務(wù)消息采用了二階段提交的方式。并且結(jié)合了消息反查的機(jī)制來確保最終一致性。

從使用層面來說，每個業(yè)務(wù)場景都要去實(shí)現(xiàn)一個反查的邏輯，有點(diǎn)煩。

下面介紹另一種經(jīng)常被使用的方式，就是本地事務(wù)消息。本地消息表這個方案最初是 ebay 提出的，本地事務(wù)消息需要在服務(wù)對應(yīng)的數(shù)據(jù)庫中創(chuàng)建一個消息表，發(fā)送消息的時候不是真正的將消息發(fā)送給 MQ，而是往消息表中插入一條消息數(shù)據(jù)。

插入的動作跟本地的業(yè)務(wù)邏輯是同一個事務(wù)，如果本地事務(wù)執(zhí)行成功，消息才會落表成功，才會發(fā)送給 MQ, 本地事務(wù)失敗，消息數(shù)據(jù)回滾。

然后需要有一個專門的程序去拉取消息表中未發(fā)送的消息投遞給 MQ，如果投遞失敗，可以一直重試，直到成功或者人工介入。

消息寫到消息表，然后會一直給 MQ 發(fā)送，這個步驟沒問題。如果 MQ 收到消息后，消息還在 PageCache 中的時候，Broker 宕機(jī)了，這個時候是會出現(xiàn)消息丟失。當(dāng)然你也可以使用同步刷盤等方式來避免丟失。假如我們就是異步刷盤，有辦法保證消息不丟失嗎？

前面我們提到，RocketMQ 的事務(wù)消息會有回查的機(jī)制，消息表的方式，也需要有一個機(jī)制來保證消息被消費(fèi)了，否則就需要不斷的重試去發(fā)送消息，直到消息被消費(fèi)。

在消息表中需要有一個字段來標(biāo)識當(dāng)前這條消息的狀態(tài)，比如 未發(fā)送，已發(fā)送，已消費(fèi)。當(dāng)消息還是未發(fā)送的時候就會被發(fā)送到 MQ, 如果發(fā)送成功了，狀態(tài)就是已發(fā)送。但是過了幾分鐘，狀態(tài)還是已發(fā)送，這個時候就要去做一些動作了。

這個場景下，有可能是消費(fèi)者跟不上生產(chǎn)的速度，消息堆積了，導(dǎo)致消息一直沒被消費(fèi)。另一種可能就是消息是不是丟失了？

可以獲取對應(yīng)的消息堆積數(shù)據(jù)來判斷是否消息堆積了，如果不是就重新發(fā)送消息給 MQ，知道消息被消費(fèi)。

問題是消息被消費(fèi)了，我怎么知道？

像我是用的云服務(wù)，是有對應(yīng)的 Open API 可以直接查詢消息軌跡。開源的應(yīng)該也有，沒有仔細(xì)去研究，跟商業(yè)版應(yīng)該差不多。

根據(jù)消息軌跡就可以知道消息有沒有被消費(fèi)，到此為止流程結(jié)束。消息發(fā)送給 MQ 如果失敗會重試，消息如果長時間沒消費(fèi)，也會重新發(fā)送，即使最后進(jìn)入了死信隊(duì)列，也可以通過死信隊(duì)列的監(jiān)控來人工干預(yù)，一定會是最終一致性。

跟自帶的事務(wù)消息比，本地消息表的方式不需要實(shí)現(xiàn)回查邏輯，但是要增加消息表，同時也要配套各種發(fā)送，檢查等邏輯，也挺麻煩了。特別是當(dāng)消息量大的時候，如何快速的將消息表中的消息發(fā)送出去，也需要做很多處理，簡單的查表輪詢在量大的情況下不太適用。

兩種方式都可以使用，能實(shí)現(xiàn)我們要的目的即可。