RocketMQ 介紹

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

RocketMQ 概念Topic：消息主題，用于將一類的消息進行歸類，比如訂單主題，就是所有訂單相關的消息都可以由這個主題去承載，生產者向這個主題發送消息。

生產者：負責生產消息并發送消息到 Topic 的角色。

消費者：負責從 Topic 接收并消費消息 的角色。

消息：生產者向 Topic 發送的內容，會被消費者消費。

消息屬性：生產者發送的時候可以為消息自定義一些業務相關的屬性，比如 Message Key 和 Tag 等。

Group：一類生產者或消費者，這類生產者或消費者通常生產或消費同一類消息，且消息發布或訂閱的邏輯一致。

為什么要使用 RocketMQ？

異步解耦

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

使用異步解耦的業務場景非常多，因為每個行業的業務都會不太一樣，以一些比較通用的業務來說明相信大家都能理解。

比如電商行業的下單業務場景，以最簡單的下單流程來說，下單流程如下：鎖庫存

創建訂單

用戶支付

扣減庫存

給用戶發送購買短信通知

給用戶增加積分

通知商家發貨

我們以下單成功后，用戶進行支付，支付完成會有個邏輯叫支付回調，在回調里面需要去做一些業務邏輯。首先來看下同步處理需要花費的時間，如下圖：

同步流程

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

異步流程

削峰填谷

削峰填谷指的是在大流量的沖擊下，利用 RocketMQ 可以抗住瞬時的大流量，保護系統的穩定性，提升用戶體驗。

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

削峰填谷

分布式事務最終一致性

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

在電商的業務場景中，交易相關的核心業務一定要確保數據的一致性。通過引入消息隊列 RocketMQ 版的分布式事務，既可以實現系統之間的解耦，又可以保證最終的數據一致性。

數據分發

數據分發指的是可以將原始數據分發到多個需要使用這份數據的系統中，實現數據異構的需求。最常見的有將數據分發到 ES, Redis 中為業務提供搜索，緩存等服務。

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

數據分發

RocketMQ 架構

圖片來源阿里云官方文檔Name Server：是一個幾乎無狀態節點，可集群部署，在消息隊列 RocketMQ 版中提供命名服務，更新和發現 Broker 服務。就是一個注冊中心。

Broker：消息中轉角色，負責存儲消息，轉發消息。分為 Master Broker 和 Slave Broker，一個 Master Broker 可以對應多個 Slave Broker，但是一個 Slave Broker 只能對應一個 Master Broker。Broker 啟動后需要完成一次將自己注冊至 Name Server 的操作；隨后每隔 30s 定期向 Name Server 上報 Topic 路由信息。

生產者：與 Name Server 集群中的其中一個節點(隨機)建立長鏈接(Keep-alive)，定期從 Name Server 讀取 Topic 路由信息，并向提供 Topic 服務的 Master Broker 建立長鏈接，且定時向 Master Broker 發送心跳。

消費者：與 Name Server 集群中的其中一個節點(隨機)建立長連接，定期從 Name Server 拉取 Topic 路由信息，并向提供 Topic 服務的 Master Broker、Slave Broker 建立長連接，且定時向 Master Broker、Slave Broker 發送心跳。Consumer 既可以從 Master Broker 訂閱消息，也可以從 Slave Broker 訂閱消息，訂閱規則由 Broker 配置決定。

RocketMQ 消息類型

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

普通消息

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

同步發送

同步發送：消息發送方發送出去一條消息，會同步得到服務端返回的結果。

異步發送

異步發送：消息發送方發出去一條消息，不用等待服務端返回結果，可以接著發送下一條消息。發送方可以通過回調接口接收服務端響應，并處理響應結果。

單向發送

單向發送：消息發送方只負責發送消息，發送出去后就不管了，這種方式發送速度非常快，存在丟失消息的風險。

順序消息

順序消息是指生產者按照一定的先后順序發布消息；消費者按照既定的先后順序訂閱消息，即先發布的消息一定會先被消費者接收到。

比如數據分發的場景，如果我們訂閱了 Mysql 的 binlog 來進行數據異構。消息要是沒有順序，就會出現數據錯亂問題。

比如新增一條 id=1 的數據，然后馬上刪除。這樣就產生了兩條消息。正常的消費順序是先新增，然后刪除，此時數據是沒有的。如果消息沒有順序，刪除的先被消費了，然后消費新增的，此時數據還在，沒被刪除掉，就會導致不一致。

定時消息

定時消息是指消息具備定時發送的功能，當消息發送到服務端后，不會立即投遞給消費者。而是要等到消息指定的時間后才會投遞給消費者進行消費。

延遲消息也就是定時消息，定時消息是定在某個時間點進行發送，比如 2020-11-11 12:00:00 發送。

延遲消息一般是在當前發送時間的基礎上延遲多久進行發送，比如當前時間是 2020-09-10 12:00:00，延遲 10 分鐘，那么消息發送成功后將在 2020-09-10 12:10:00 進行投遞給消費者。

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

事務消息

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

交互流程：

圖片來源阿里云官方文檔發送方首先發送半事務消息到 RocketMQ 服務端。

RocketMQ 服務端接收到消息，然后將消息持久化成功之后，向發送方返回 Ack 確認消息已經發送成功，此時消息為半事務消息，不會投遞給消費方。

收到半事務消息的 Ack 后，發送方開始執行本地事務邏輯。

發送方根據本地事務執行結果向服務端提交二次確認，如果本地事務執行成則進行消息的 Commit，如果執行失敗則進行消息的 Rollback，服務端收到 Commit 狀態則將半事務消息標記為可投遞，消費方最終將收到該消息；服務端收到 Rollback 狀態則刪除半事務消息，消費方將不會收到該消息。

如果出現意外情況，步驟 4 沒有進行消息的二次確認，等待固定時間后服務端將對該消息發起消息回查。

發送方收到消息回查后，需要檢查對應消息的本地事務執行的最終結果。發送方根據檢查得到的本地事務的最終狀態再次提交二次確認，服務端仍按照步驟 4 對半事務消息進行操作。

最佳實踐

消息重試

消息在消費方消費失敗后，RocketMQ 服務端會重新進行消息的投遞，知道消費者成功消費消息，當然重試有次數限制，默認 16 次。

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

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

最后需要做好對應的監控，如果重試了 4，5 次還是失敗的，基本上后面重試也是失敗的。這個時候需要讓開發人員知道，該人工處理的就人工介入。或者直接監控死信隊列。

消息過濾

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

不同類型的消息有不同的業務處理，我們可以統一定義消息格式，然后通過一個字段去區分消息類型來做不同的業務邏輯。不好的點在于所有消息都會推送到消費方，不能按需消費。

在 RocketMQ 中可以給消息指定 tag，通過 tag 來區分消息類型。消費者可以根據 Tag 在 RocketMQ 服務端完成消息過濾，以確保消費者最終只消費到其關注的消息類型。

我曾經遇到過一個 tag 沒有正確使用的方式，只有一個 MQ 實例，用 tag 來區分環境。所有消息都在一個主題中，測試環境消費測試環境的 tag，線上消費線上的 tag。

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

消息過濾

消費模式

RocketMQ 消費模式有兩種，集群消費和廣播消費。

集群消費：

集群消費

消費者部署了多個實例我們稱之為一個集群，集群消費只會被其中的某一個實例進行消費。

適合大部分的業務場景，大部分的場景我們的消息只允許被消費一次，而且只能有一個消費者去消費，比如支付回調場景，如果一個消息被多個實例同時消費，那么就會出現同時去修改訂單狀態，同時去扣減庫存的情況。

廣播消費：

廣播消費

廣播消費會讓集群中每個實例都消費一次。

比如我們使用了本地緩存，當數據變更的時候，我們需要刷新每個節點本地的緩存，所以每個節點都需要收到消息。

消費冪等

冪等問題，無論是在 API 請求場景還是在消息消費場景，都會遇到。一條消息不能重復消費多次這個肯定是要保證的，因為我們不能保證消息發送方不發送多次，也不能保證消息不重復投遞。

RocketMQ 的 Exactly-Once 投遞語義，就是用于解決冪等問題。Exactly-Once 是指發送到消息系統的消息只能被消費端處理且僅處理一次，即使生產端重試消息發送導致某消息重復投遞，該消息在消費端也只被消費一次。

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

本地事務消息封裝

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

從使用層面來說，每個業務場景都要去實現一個反查的邏輯，有點煩。

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

插入的動作跟本地的業務邏輯是同一個事務，如果本地事務執行成功，消息才會落表成功，才會發送給 MQ, 本地事務失敗，消息數據回滾。

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

本地事務消息

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

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

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

這個場景下，有可能是消費者跟不上生產的速度，消息堆積了，導致消息一直沒被消費。另一種可能就是消息是不是丟失了？

可以獲取對應的消息堆積數據來判斷是否消息堆積了，如果不是就重新發送消息給 MQ，知道消息被消費。

問題是消息被消費了，我怎么知道？

像我使用的云服務，是有對應的 Open API 可以直接查詢消息軌跡。開源的應該也有，沒有仔細去研究，跟商業版應該差不多。

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

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

兩種方式都可以使用，能實現我們要的目的即可。

總結

以上是生活随笔為你收集整理的mq幂等mysql_膜拜！看完这篇你还不懂RocketMQ算我输的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。