Apache RocketMQ 5.0 筆記

RocketMQ 5.0：雲原生“消息、事件、流”實時數據處理平臺，覆蓋雲邊端一體化數據處理場景。

核心特性

• 雲原生：生與雲，長與雲，無限彈性擴縮，K8s友好
• 高吞吐：萬億級吞吐保證，同時滿足微服務與大數據場景
• 流處理：提供輕量、高擴展、高性能和豐富功能的流計算引擎
• 金融級：金融級的穩定性，廣泛用於交易核心鏈路
• 架構極簡：零外部依賴，Shared-nothing 架構
• 生態友好：無縫對接微服務、實時計算、數據湖等周邊生態

1. 基本概念

1、消息由生產者初始化併發送到Apache RocketMQ 服務端。

2、消息按照到達Apache RocketMQ 服務端的順序存儲到主題的指定隊列中。

3、消費者按照指定的訂閱關係從Apache RocketMQ 服務端中獲取消息並消費。

1.1. 消息

消息是 Apache RocketMQ 中的最小數據傳輸單元。生產者將業務數據的負載和拓展屬性包裝成消息發送到 Apache RocketMQ 服務端，服務端按照相關語義將消息投遞到消費端進行消費。

RocketMQ 消息構成非常簡單，如下所示：

• topic：表示要發送的消息的主題
• body：表示消息的存儲內容
• properties：表示消息屬性
• transactionId：會在事務消息中使用

消息內部屬性

系統預設的消息最大限制如下：

• 普通和順序消息：4 MB
• 事務和定時或延時消息：64 KB

1.2. Tag

Topic 與 Tag 都是業務上用來歸類的標識，區別在於 Topic 是一級分類，而 Tag 可以理解為是二級分類。使用 Tag 可以實現對 Topic 中的消息進行過濾。

提示：

• Topic：消息主題，通過 Topic 對不同的業務消息進行分類。
• Tag：消息標簽，用來進一步區分某個 Topic 下的消息分類，消息從生產者發出即帶上的屬性。

Topic 和 Tag 的關係如下圖所示：

什麼時候該用 Topic，什麼時候該用 Tag？

可以從以下幾個方面進行判斷：

• 消息類型是否一致：如普通消息、事務消息、定時（延時）消息、順序消息，不同的消息類型使用不同的 Topic，無法通過 Tag 進行區分。
• 業務是否相關聯：沒有直接關聯的消息，如淘寶交易消息，京東物流消息使用不同的 Topic 進行區分；而同樣是天貓交易消息，電器類訂單、女裝類訂單、化妝品類訂單的消息可以用 Tag 進行區分。
• 消息優先順序是否一致：如同樣是物流消息，盒馬必須小時內送達，天貓超市 24 小時內送達，淘寶物流則相對會慢一些，不同優先順序的消息用不同的 Topic 進行區分。
• 消息量級是否相當：有些業務消息雖然量小但是實時性要求高，如果跟某些萬億量級的消息使用同一個 Topic，則有可能會因為過長的等待時間而“餓死”，此時需要將不同量級的消息進行拆分，使用不同的 Topic。

通常情況下，不同的 Topic 之間的消息沒有必然的聯繫，而 Tag 則用來區分同一個 Topic 下相互關聯的消息，例如全集和子集的關係、流程先後的關係。

1.3. Keys

Apache RocketMQ 每個消息可以在業務層面的設置唯一標識碼 keys 欄位，方便將來定位消息丟失問題。 Broker 端會為每個消息創建索引（哈希索引），應用可以通過 topic、key 來查詢這條消息內容，以及消息被誰消費。由於是哈希索引，請務必保證 key 儘可能唯一，這樣可以避免潛在的哈希衝突。

// 訂單Id
String orderId = "20034568923546";
message.setKeys(orderId);

1.4. 隊列

一個 Topic 可能有多個隊列，並且可能分佈在不同的 Broker 上。

隊列天然具備順序性，即消息按照進入隊列的順序寫入存儲，同一隊列間的消息天然存在順序關係，隊列頭部為最早寫入的消息，隊列尾部為最新寫入的消息。消息在隊列中的位置和消息之間的順序通過位點（Offset）進行標記管理。

Apache RocketMQ 預設提供消息可靠存儲機制，所有發送成功的消息都被持久化存儲到隊列中，配合生產者和消費者客戶端的調用可實現至少投遞一次的可靠性語義。

Apache RocketMQ 隊列模型和Kafka的分區（Partition）模型類似。在 Apache RocketMQ 消息收發模型中，隊列屬於主題的一部分，雖然所有的消息資源以主題粒度管理，但實際的操作實現是面向隊列。例如，生產者指定某個主題，向主題內發送消息，但實際消息發送到該主題下的某個隊列中。

Apache RocketMQ 中通過修改隊列數量，以此實現橫向的水平擴容和縮容。

一般來說一條消息，如果沒有重覆發送（比如因為服務端沒有響應而進行重試），則只會存在在 Topic 的其中一個隊列中，消息在隊列中按照先進先出的原則存儲，每條消息會有自己的位點，每個隊列會統計當前消息的總條數，這個稱為最大位點 MaxOffset；隊列的起始位置對應的位置叫做起始位點 MinOffset。隊列可以提升消息發送和消費的併發度。

註意：按照實際業務消耗設置隊列數，隊列數量的設置應遵循少用夠用原則，避免隨意增加隊列數量。

1.5. 生產者

生產者（Producer）就是消息的發送者，Apache RocketMQ 擁有豐富的消息類型，可以支持不同的應用場景，在不同的場景中，需要使用不同的消息進行發送。比如在電商交易中超時未支付關閉訂單的場景，在訂單創建時會發送一條延時消息。這條消息將會在 30 分鐘以後投遞給消費者，消費者收到此消息後需要判斷對應的訂單是否已完成支付。如支付未完成，則關閉訂單。如已完成支付則忽略，此時就需要用到延遲消息；電商場景中，業務上要求同一訂單的消息保持嚴格順序，此時就要用到順序消息。在日誌處理場景中，可以接受的比較大的發送延遲，但對吞吐量的要求很高，希望每秒能處理百萬條日誌，此時可以使用批量消息。在銀行扣款的場景中，要保持上游的扣款操作和下游的簡訊通知保持一致，此時就要使用事務消息。

註意：不要在同一個主題內使用多種消息類型

生產者通常被集成在業務系統中，將業務消息按照要求封裝成 Apache RocketMQ 的消息（Message）併發送至服務端。

生產者和主題的關係為多對多關係，即同一個生產者可以向多個主題發送消息，同一個主題也可以接收多個生產者的消息。

註意：不建議頻繁創建和銷毀生產者

Producer p = ProducerBuilder.build();
for (int i =0;i<n;i++){
    Message m= MessageBuilder.build();
    p.send(m);
}
p.shutdown();

1.6. 消費者與消費者組

如果多個消費者設置了相同的Consumer Group，我們認為這些消費者在同一個消費組內。同一個消費組的多個消費者必須保持消費邏輯和配置一致，共同分擔該消費組訂閱的消息，實現消費能力的水平擴展。

在 Apache RocketMQ 有兩種消費模式，分別是：

• 集群消費模式：當使用集群消費模式時，RocketMQ 認為任意一條消息只需要被消費組內的任意一個消費者處理即可。
• 廣播消費模式：當使用廣播消費模式時，RocketMQ 會將每條消息推送給消費組所有的消費者，保證消息至少被每個消費者消費一次。

負載均衡

RocketMQ的負載均衡策略與Kafka極其類似，幾乎一毛一樣

集群模式下，同一個消費組內的消費者會分擔收到的全量消息，這裡的分配策略是怎樣的？如果擴容消費者是否一定能提升消費能力？

Apache RocketMQ 提供了多種集群模式下的分配策略，包括平均分配策略、機房優先分配策略、一致性hash分配策略等，可以通過如下代碼進行設置相應負載均衡策略。

consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragely());

預設的分配策略是平均分配，這也是最常見的策略。平均分配策略下消費組內的消費者會按照類似分頁的策略均攤消費。

在平均分配的演算法下，可以通過增加消費者的數量來提高消費的並行度。比如下圖中，通過增加消費者來提高消費能力。

但也不是一味地增加消費者就能提升消費能力的，比如下圖中Topic的總隊列數小於消費者的數量時，消費者將分配不到隊列，即使消費者再多也無法提升消費能力。

1.7. 消費者分類

如上圖所示， Apache RocketMQ 的消費者處理消息時主要經過以下階段：消息獲取--->消息處理--->消費狀態提交。

針對以上幾個階段，Apache RocketMQ 提供了不同的消費者類型： PushConsumer 、SimpleConsumer 和 PullConsumer。這幾種類型的消費者通過不同的實現方式和介面可滿足您在不同業務場景下的消費需求。具體差異如下：

註：在實際使用場景中，PullConsumer 僅推薦在流處理框架中集成使用，大多數消息收發場景使用 PushConsumer 和 SimpleConsumer 就可以滿足需求。

PushConsumer

PushConsumers是一種高度封裝的消費者類型，消費消息僅通過消費監聽器處理業務並返回消費結果。消息的獲取、消費狀態提交以及消費重試都通過 Apache RocketMQ 的客戶端SDK完成。

SimpleConsumer

SimpleConsumer 是一種介面原子型的消費者類型，消息的獲取、消費狀態提交以及消費重試都是通過消費者業務邏輯主動發起調用完成。

補充：

rocketmq-client中定義的：

• DefaultMQProducer
• DefaultMQPushConsumer
• DefaultLitePullConsumer

rocketmq-client-java中定義的：

• Producer
• PushConsumer
• SimpleConsumer

1.8. 消費位點

消息是按到達Apache RocketMQ 服務端的先後順序存儲在指定主題的多個隊列中，每條消息在隊列中都有一個唯一的Long類型坐標，這個坐標被定義為消息位點。一條消息被某個消費者消費完成後不會立即從隊列中刪除，Apache RocketMQ 會基於每個消費者分組記錄消費過的最新一條消息的位點，即消費位點。

如上圖所示，在Apache RocketMQ中每個隊列都會記錄自己的最小位點、最大位點。針對於消費組，還有消費位點的概念，在集群模式下，消費位點是由客戶端提給交服務端保存的，在廣播模式下，消費位點是由客戶端自己保存的。一般情況下消費位點正常更新，不會出現消息重覆，但如果消費者發生崩潰或有新的消費者加入群組，就會觸發重平衡，重平衡完成後，每個消費者可能會分配到新的隊列，而不是之前處理的隊列。為了能繼續之前的工作，消費者需要讀取每個隊列最後一次的提交的消費位點，然後從消費位點處繼續拉取消息。但在實際執行過程中，由於客戶端提交給服務端的消費位點並不是實時的，所以重平衡就可能會導致消息少量重覆。

1.9. 訂閱關係

一個訂閱關係指的是指定某個消費者分組對於某個主題的訂閱。

不同消費者分組對於同一個主題的訂閱相互獨立如下圖所示，消費者分組Group A和消費者分組Group B分別以不同的訂閱關係訂閱了同一個主題Topic A，這兩個訂閱關係互相獨立，可以各自定義，不受影響。

同一個消費者分組對於不同主題的訂閱也相互獨立如下圖所示，消費者分組Group A訂閱了兩個主題Topic A和Topic B，對於Group A中的消費者來說，訂閱的Topic A為一個訂閱關係，訂閱的Topic B為另外一個訂閱關係，且這兩個訂閱關係互相獨立，可以各自定義，不受影響。

2. 消息類型

1、順序消息（FIFO）：這類消息必須設置 message group，這種類型的消息需要與FIFO消費者組一起使用

2、延遲消息（DELAY）：消息被髮送後不會立即對消費者可見，這種類型的消息必須設置delivery timestamp以決定對消費者可見的時間；

3、事務消息（TRANSACTIONAL）：將一個或多個消息的發佈包裝到一個事務中，提供提交/回滾方法來決定消息的可見性；

4、普通消息（NORMAL）：預設類型

不同的類型是互斥的，當意味著要發佈的消息不能同時是FIFO類型和DELAY類型。實際上，主題的類型決定了消息的類型。例如，FIFO主題不允許發佈其他類型的消息。

2.1. 普通消息

普通消息一般應用於微服務解耦、事件驅動、數據集成等場景，這些場景大多數要求數據傳輸通道具有可靠傳輸的能力，且對消息的處理時機、處理順序沒有特別要求。

典型場景一：微服務非同步解耦

如上圖所示，以線上的電商交易場景為例，上游訂單系統將用戶下單支付這一業務事件封裝成獨立的普通消息併發送至Apache RocketMQ服務端，下游按需從服務端訂閱消息並按照本地消費邏輯處理下游任務。每個消息之間都是相互獨立的，且不需要產生關聯。

典型場景二：數據集成傳輸

如上圖所示，以離線的日誌收集場景為例，通過埋點組件收集前端應用的相關操作日誌，並轉發到 Apache RocketMQ 。每條消息都是一段日誌數據，Apache RocketMQ 不做任何處理，只需要將日誌數據可靠投遞到下游的存儲系統和分析系統即可，後續功能由後端應用完成。

2.2. 順序消息

應用場景

在有序事件處理、撮合交易、數據實時增量同步等場景下，異構系統間需要維持強一致的狀態同步，上游的事件變更需要按照順序傳遞到下游進行處理。在這類場景下使用 Apache RocketMQ 的順序消息可以有效保證數據傳輸的順序性。

典型場景一：撮合交易

以證券、股票交易撮合場景為例，對於出價相同的交易單，堅持按照先出價先交易的原則，下游處理訂單的系統需要嚴格按照出價順序來處理訂單。

典型場景二：數據實時增量同步

以資料庫變更增量同步場景為例，上游源端資料庫按需執行增刪改操作，將二進位操作日誌作為消息，通過 Apache RocketMQ 傳輸到下游搜索系統，下游系統按順序還原消息數據，實現狀態數據按序刷新。如果是普通消息則可能會導致狀態混亂，和預期操作結果不符，基於順序消息可以實現下游狀態和上游操作結果一致。

功能原理

順序消息是 Apache RocketMQ 提供的一種高級消息類型，支持消費者按照發送消息的先後順序獲取消息，從而實現業務場景中的順序處理。 相比其他類型消息，順序消息在發送、存儲和投遞的處理過程中，更多強調多條消息間的先後順序關係。

Apache RocketMQ 順序消息的順序關係通過消息組（MessageGroup）判定和識別，發送順序消息時需要為每條消息設置歸屬的消息組，相同消息組的多條消息之間遵循先進先出的順序關係，不同消息組、無消息組的消息之間不涉及順序性。

基於消息組的順序判定邏輯，支持按照業務邏輯做細粒度拆分，可以在滿足業務局部順序的前提下提高系統的並行度和吞吐能力。

如何保證消息的順序性？

Apache RocketMQ 的消息的順序性分為兩部分，生產順序性和消費順序性。

1、生產順序性

如需保證消息生產的順序性，則必須滿足以下條件：

• 單一生產者：消息生產的順序性僅支持單一生產者，不同生產者分佈在不同的系統，即使設置相同的消息組，不同生產者之間產生的消息也無法判定其先後順序。
• 串列發送：Apache RocketMQ 生產者客戶端支持多線程安全訪問，但如果生產者使用多線程並行發送，則不同線程間產生的消息將無法判定其先後順序。

滿足以上條件的生產者，將順序消息發送至 Apache RocketMQ 後，會保證設置了同一消息組的消息，按照發送順序存儲在同一隊列中。服務端順序存儲邏輯如下：

• 相同消息組的消息按照先後順序被存儲在同一個隊列。
• 不同消息組的消息可以混合在同一個隊列中，且不保證連續。

2、消費順序性

如需保證消息消費的順序性，則必須滿足以下條件：

• 投遞順序：Apache RocketMQ 通過客戶端SDK和服務端通信協議保障消息按照服務端存儲順序投遞，但業務方消費消息時需要嚴格按照接收---處理---應答的語義處理消息，避免因非同步處理導致消息亂序。
• 有限重試：Apache RocketMQ 順序消息投遞僅在重試次數限定範圍內，即一條消息如果一直重試失敗，超過最大重試次數後將不再重試，跳過這條消息消費，不會一直阻塞後續消息處理。對於需要嚴格保證消費順序的場景，請務設置合理的重試次數，避免參數不合理導致消息亂序。

生產順序性和消費順序性組合

如果消息需要嚴格按照先進先出（FIFO）的原則處理，即先發送的先消費、後發送的後消費，則必須要同時滿足生產順序性和消費順序性。

一般業務場景下，同一個生產者可能對接多個下游消費者，不一定所有的消費者業務都需要順序消費，您可以將生產順序性和消費順序性進行差異化組合，應用於不同的業務場景。例如發送順序消息，但使用非順序的併發消費方式來提高吞吐能力。更多組合方式如下表所示：

2.3. 定時/延時消息

註：定時消息和延時消息本質相同，都是服務端根據消息設置的定時時間在某一固定時刻將消息投遞給消費者消費。

應用場景

在分散式定時調度觸發、任務超時處理等場景，需要實現精準、可靠的定時事件觸發。使用 Apache RocketMQ 的定時消息可以簡化定時調度任務的開發邏

輯，實現高性能、可擴展、高可靠的定時觸發能力。

典型場景一：分散式定時調度

在分散式定時調度場景下，需要實現各類精度的定時任務，例如每天5點執行文件清理，每隔2分鐘觸發一次消息推送等需求。基於 Apache RocketMQ 的定時消息可以封裝出多種類型的定時觸發器。

典型場景二：任務超時處理

以電商交易場景為例，訂單下單後暫未支付，此時不可以直接關閉訂單，而是需要等待一段時間後才能關閉訂單。使用 Apache RocketMQ 定時消息可以實現超時任務的檢查觸發。

基於定時消息的超時任務處理具備如下優勢：

• 精度高、開發門檻低：基於消息通知方式不存在定時階梯間隔。可以輕鬆實現任意精度事件觸發，無需業務去重。
• 高性能可擴展：傳統的資料庫掃描方式較為複雜，需要頻繁調用介面掃描，容易產生性能瓶頸。 Apache RocketMQ 的定時消息具有高併發和水平擴展的能力。

功能原理

定時時間設置原則

Apache RocketMQ 定時消息設置的定時時間是一個預期觸發的系統時間戳，延時時間也需要轉換成當前系統時間後的某一個時間戳，而不是一段延時時長。

投遞等級

Apache RocketMQ 一共支持18個等級的延遲投遞，具體時間如下：

2.4. 事務消息

以電商交易場景為例，用戶支付訂單這一核心操作的同時會涉及到下游物流發貨、積分變更、購物車狀態清空等多個子系統的變更。當前業務的處理分支包括：

• 主分支訂單系統狀態更新：由未支付變更為支付成功。
• 物流系統狀態新增：新增待發貨物流記錄，創建訂單物流記錄。
• 積分系統狀態變更：變更用戶積分，更新用戶積分表。
• 購物車系統狀態變更：清空購物車，更新用戶購物車記錄。

使用普通消息和訂單事務無法保證一致的原因，本質上是由於普通消息無法像單機資料庫事務一樣，具備提交、回滾和統一協調的能力。而基於 RocketMQ 的分散式事務消息功能，在普通消息基礎上，支持二階段的提交能力。將二階段提交和本地事務綁定，實現全局提交結果的一致性。

事務消息發送分為兩個階段。第一階段會發送一個半事務消息，半事務消息是指暫不能投遞的消息，生產者已經成功地將消息發送到了 Broker，但是Broker 未收到生產者對該消息的二次確認，此時該消息被標記成“暫不能投遞”狀態，如果發送成功則執行本地事務，並根據本地事務執行成功與否，向 Broker 半事務消息狀態（commit或者rollback），半事務消息只有 commit 狀態才會真正向下游投遞。如果由於網路閃斷、生產者應用重啟等原因，導致某條事務消息的二次確認丟失，Broker 端會通過掃描發現某條消息長期處於“半事務消息”時，需要主動向消息生產者詢問該消息的最終狀態（Commit或是Rollback）。這樣最終保證了本地事務執行成功，下游就能收到消息，本地事務執行失敗，下游就收不到消息。總而保證了上下游數據的一致性。（PS：重點是兩階段提交）

事務消息處理流程

1、生產者將消息發送至Apache RocketMQ服務端。

2、Apache RocketMQ服務端將消息持久化成功之後，向生產者返回Ack確認消息已經發送成功，此時消息被標記為"暫不能投遞"，這種狀態下的消息即為半事務消息。

3、生產者開始執行本地事務邏輯。

4、生產者根據本地事務執行結果向服務端提交二次確認結果（Commit或是Rollback），服務端收到確認結果後處理邏輯如下：

• 二次確認結果為Commit：服務端將半事務消息標記為可投遞，並投遞給消費者。
• 二次確認結果為Rollback：服務端將回滾事務，不會將半事務消息投遞給消費者。

5、在斷網或者是生產者應用重啟的特殊情況下，若服務端未收到發送者提交的二次確認結果，或服務端收到的二次確認結果為Unknown未知狀態，經過固定時間後，服務端將對消息生產者即生產者集群中任一生產者實例發起消息回查。

6、生產者收到消息回查後，需要檢查對應消息的本地事務執行的最終結果。

7、生產者根據檢查到的本地事務的最終狀態再次提交二次確認，服務端仍按照步驟4對半事務消息進行處理。

3. 機制

3.1. 消息發送重試機制

Apache RocketMQ 客戶端連接服務端發起消息發送請求時，可能會因為網路故障、服務異常等原因導致調用失敗。為保證消息的可靠性， Apache RocketMQ 在客戶端SDK中內置請求重試邏輯，嘗試通過重試發送達到最終調用成功的效果。

同步發送和非同步發送模式均支持消息發送重試。

重試觸發條件：

• 客戶端消息發送請求調用失敗或請求超時
• 網路異常造成連接失敗或請求超時
• 服務端節點處於重啟或下線等狀態造成連接失敗
• 服務端運行慢造成請求超時
• 服務端返回失敗錯誤碼

重試流程：

生產者在初始化時設置消息發送最大重試次數，當出現上述觸發條件的場景時，生產者客戶端會按照設置的重試次數一直重試發送消息，直到消息發送成功或達到最大重試次數重試結束，併在最後一次重試失敗後返回調用錯誤響應。

• 同步發送：調用線程會一直阻塞，直到某次重試成功或最終重試失敗，拋出錯誤碼和異常。
• 非同步發送：調用線程不會阻塞，但調用結果會通過異常事件或者成功事件返回。

重試間隔

• 除服務端返回系統流控錯誤場景，其他觸發條件觸發重試後，均會立即進行重試，無等待間隔。
• 若由於服務端返迴流控錯誤觸發重試，系統會按照指數退避策略進行延遲重試。指數退避演算法通過以下參數控制重試行為：
  • INITIAL_BACKOFF： 第一次失敗重試前後需等待多久，預設值：1秒
  • MULTIPLIER ：指數退避因數，即退避倍率，預設值：1.6
  • JITTER ：隨機抖動因數，預設值：0.2
  • MAX_BACKOFF ：等待間隔時間上限，預設值：120秒
  • MIN_CONNECT_TIMEOUT ：最短重試間隔，預設值：20秒 

3.2. 消息流控機制

消息流控指的是系統容量或水位過高， Apache RocketMQ 服務端會通過快速失敗返迴流控錯誤來避免底層資源承受過高壓力。

觸發條件

• 存儲壓力大：消費者分組的初始消費位點為當前隊列的最大消費位點。
• 服務端請求任務排隊溢出：若消費者消費能力不足，導致隊列中有大量堆積消息，當堆積消息超過一定數量後會觸發消息流控，減少下游消費系統壓力。

流控行為

當系統觸發消息發送流控時，客戶端會收到系統限流錯誤和異常，錯誤碼信息如下：

• reply-code：530
• reply-text：TOO_MANY_REQUESTS

3.3. 消費重試

消費者出現異常，消費某條消息失敗時， Apache RocketMQ 會根據消費重試策略重新投遞該消息。消費重試主要解決的是業務處理邏輯失敗導致的消費完整性問題，是一種為業務兜底的策略，不應該被用做業務流程式控制制。

推薦使用消息重試場景如下：

• 業務處理失敗，且失敗原因跟當前的消息內容相關，比如該消息對應的事務狀態還未獲取到，預期一段時間後可執行成功。
• 消費失敗的原因不會導致連續性，即當前消息消費失敗是一個小概率事件，不是常態化的失敗，後面的消息大概率會消費成功。此時可以對當前消息進行重試，避免進程阻塞。

消費重試策略

消費重試指的是，消費者在消費某條消息失敗後，Apache RocketMQ 服務端會根據重試策略重新消費該消息，超過一次定數後若還未消費成功，則該消息將不再繼續重試，直接被髮送到死信隊列中。

消息重試的觸發條件

• 消費失敗，包括消費者返回消息失敗狀態標識或拋出非預期異常。
• 消息處理超時，包括在PushConsumer中排隊超時。

重試策略差異

3.4. 消費進度

消息位點（Offset）

消息是按到達服務端的先後順序存儲在指定主題的多個隊列中，每條消息在隊列中都有一個唯一的Long類型坐標，這個坐標被定義為消息位點。

任意一個消息隊列在邏輯上都是無限存儲，即消息位點會從0到Long.MAX無限增加。通過主題、隊列和位點就可以定位任意一條消息的位置，具體關係如下圖所示：

Apache RocketMQ 定義隊列中最早一條消息的位點為最小消息位點（MinOffset）；最新一條消息的位點為最大消息位點（MaxOffset）。雖然消息隊列邏輯上是無限存儲，但由於服務端物理節點的存儲空間有限， Apache RocketMQ 會滾動刪除隊列中存儲最早的消息。因此，消息的最小消費位點和最大消費位點會一直遞增變化。

消費位點（ConsumerOffset）

Apache RocketMQ 領域模型為發佈訂閱模式，每個主題的隊列都可以被多個消費者分組訂閱。若某條消息被某個消費者消費後直接被刪除，則其他訂閱了該主題的消費者將無法消費該消息。

因此，Apache RocketMQ 通過消費位點管理消息的消費進度。每條消息被某個消費者消費完成後不會立即在隊列中刪除，Apache RocketMQ 會基於每個消費者分組維護一份消費記錄，該記錄指定消費者分組消費某一個隊列時，消費過的最新一條消息的位點，即消費位點。

當消費者客戶端離線，又再次重新上線時，會嚴格按照服務端保存的消費進度繼續處理消息。如果服務端保存的歷史位點信息已過期被刪除，此時消費位點向前移動至服務端存儲的最小位點。

註：消費位點的保存和恢復是基於 Apache RocketMQ 服務端的存儲實現，和任何消費者無關。

隊列中消息位點MinOffset、MaxOffset和每個消費者分組的消費位點ConsumerOffset的關係如下：

ConsumerOffset≤MaxOffset：

• 當消費速度和生產速度一致，且全部消息都處理完成時，最大消息位點和消費位點相同，即ConsumerOffset=MaxOffset
• 當消費速度較慢小於生產速度時，隊列中會有部分消息未消費，此時消費位點小於最大消息位點，即ConsumerOffset<MaxOffset，兩者之差就是該隊列中堆積的消息量

ConsumerOffset≥MinOffset：

• 正常情況下有效的消費位點ConsumerOffset必然大於等於最小消息位點MinOffset。消費位點小於最小消息位點時是無效的，相當於消費者要消費的消息已經從隊列中刪除了，是無法消費到的，此時服務端會將消費位點強制糾正到合法的消息位點。

消費位點初始值

消費位點初始值指的是消費者分組首次啟動消費者消費消息時服務端保存的消費位點的初始值。Apache RocketMQ 定義消費位點的初始值為消費者首次獲取消息時，該時刻隊列中的最大消息位點。相當於消費者將從隊列中最新的消息開始消費。

3.5. 消息存儲機制

Apache RocketMQ 使用存儲時長作為消息存儲的依據，即每個節點對外承諾消息的存儲時長。在存儲時長範圍內的消息都會被保留，無論消息是否被消費；超過時長限制的消息則會被清理掉。

4. 架構

4.1. 技術架構

RocketMQ架構上主要分為四部分，如上圖所示：

• Producer：消息發佈的角色，支持分散式集群方式部署。Producer通過MQ的負載均衡模塊選擇相應的Broker集群隊列進行消息投遞，投遞的過程支持快速失敗並且低延遲。
• Consumer：消息消費的角色，支持分散式集群方式部署。支持以push推，pull拉兩種模式對消息進行消費。同時也支持集群方式和廣播方式的消費，它提供實時消息訂閱機制，可以滿足大多數用戶的需求。
• NameServer：NameServer是一個非常簡單的Topic路由註冊中心，其角色類似Dubbo中的zookeeper，支持Broker的動態註冊與發現。主要包括兩個功能：Broker管理，NameServer接受Broker集群的註冊信息並且保存下來作為路由信息的基本數據。然後提供心跳檢測機制，檢查Broker是否還存活；路由信息管理，每個NameServer將保存關於Broker集群的整個路由信息和用於客戶端查詢的隊列信息。然後Producer和Consumer通過NameServer就可以知道整個Broker集群的路由信息，從而進行消息的投遞和消費。NameServer通常也是集群的方式部署，各實例間相互不進行信息通訊。Broker是向每一臺NameServer註冊自己的路由信息，所以每一個NameServer實例上面都保存一份完整的路由信息。當某個NameServer因某種原因下線了，Broker仍然可以向其它NameServer同步其路由信息，Producer和Consumer仍然可以動態感知Broker的路由的信息。
• BrokerServer：Broker主要負責消息的存儲、投遞和查詢以及服務高可用保證，為了實現這些功能，Broker包含了以下幾個重要子模塊。

4.2. 部署架構

RocketMQ 網路部署特點：

• NameServer是一個幾乎無狀態節點，可集群部署，節點之間無任何信息同步。
• Broker部署相對複雜，Broker分為Master與Slave，一個Master可以對應多個Slave，但是一個Slave只能對應一個Master，Master與Slave 的對應關係通過指定相同的BrokerName，不同的BrokerId 來定義，BrokerId為0表示Master，非0表示Slave。Master也可以部署多個。每個Broker與NameServer集群中的所有節點建立長連接，定時註冊Topic信息到所有NameServer。 註意：當前RocketMQ版本在部署架構上支持一Master多Slave，但只有BrokerId=1的從伺服器才會參與消息的讀負載。

• Producer與NameServer集群中的其中一個節點（隨機選擇）建立長連接，定期從NameServer獲取Topic路由信息，並向提供Topic 服務的Master建立長連接，且定時向Master發送心跳。Producer完全無狀態，可集群部署。

• Consumer與NameServer集群中的其中一個節點（隨機選擇）建立長連接，定期從NameServer獲取Topic路由信息，並向提供Topic服務的Master、Slave建立長連接，且定時向Master、Slave發送心跳。Consumer既可以從Master訂閱消息，也可以從Slave訂閱消息，消費者在向Master拉取消息時，Master伺服器會根據拉取偏移量與最大偏移量的距離（判斷是否讀老消息，產生讀I/O），以及從伺服器是否可讀等因素建議下一次是從Master還是Slave拉取。

結合部署架構圖，描述集群工作流程：

• 啟動NameServer，NameServer起來後監聽埠，等待Broker、Producer、Consumer連上來，相當於一個路由控制中心。
• Broker啟動，跟所有的NameServer保持長連接，定時發送心跳包。心跳包中包含當前Broker信息(IP+埠等)以及存儲所有Topic信息。註冊成功後，NameServer集群中就有Topic跟Broker的映射關係。
• 收發消息前，先創建Topic，創建Topic時需要指定該Topic要存儲在哪些Broker上，也可以在發送消息時自動創建Topic。
• Producer發送消息，啟動時先跟NameServer集群中的其中一臺建立長連接，並從NameServer中獲取當前發送的Topic存在哪些Broker上，輪詢從隊列列表中選擇一個隊列，然後與隊列所在的Broker建立長連接從而向Broker發消息。
• Consumer跟Producer類似，跟其中一臺NameServer建立長連接，獲取當前訂閱Topic存在哪些Broker上，然後直接跟Broker建立連接通道，開始消費消息。

5. 客戶端

在編寫客戶端代碼時，首先準備一個簡單的環境，可以選用Local模式。這裡不多介紹，只說一句，啟動broker的時候可以-c指定配置文件，啟動完以後通過jps查看進程

通過mqadmin命令創建並查看主題

mqadmin updateTopic -n localhost:9876 -b 172.16.52.116:10911 -t TEST_TOPIC
mqadmin topicList -n localhost:9876

具體命令參數，參見  https://rocketmq.apache.org/zh/docs/deploymentOperations/16admintool/

也可以通過RocketMQ Dashboard創建主題

5.1. rocketmq-client

引入依賴

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>5.0.0</version>
</dependency>

代碼片段

public class AppTest extends TestCase {

    private String producerGroupName = "MyProducerGroup";
    private String consumerGroupName = "MyConsumerGroup";

    /**
     * 發送同步消息
     */
    @Test
    public void testSyncProducer() throws Exception {
        // 實例化消息生產者Producer
        DefaultMQProducer producer = new DefaultMQProducer(producerGroupName);
        // 設置NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 啟動Producer實例
        producer.start();
        // 發送消息
        Message message = new Message("TEST_TOPIC", "A", "UserID12345", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult sendResult = producer.send(message);
        System.out.println(sendResult);
        // 關閉Producer實例
        producer.shutdown();
    }

    /**
     * 發送非同步消息
     */
    @Test
    public void testAsyncProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(producerGroupName);
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);

        Message msg = new Message("TEST_TOPIC", "B", "OrderID12346", "Hello World".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // SendCallback接收非同步返回結果的回調
        producer.send(msg, new SendCallback() {
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult);
            }

            public void onException(Throwable e) {
                e.printStackTrace();
            }
        });
        //  等待5秒
        TimeUnit.SECONDS.sleep(5);
    }

    /**
     * 單向發送消息
     * 這種方式主要用在不特別關心發送結果的場景，例如日誌發送。
     */
    @Test
    public void testOnewayProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(producerGroupName);
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        Message msg = new Message("TEST_TOPIC", "C", "OrderID12348", "Hello World".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // 發送單向消息，沒有任何返回結果
        producer.sendOneway(msg);
    }

    /**
     * 消費消息
     */
    @Test
    public void testConsumer() throws Exception {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(consumerGroupName);
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("TEST_TOPIC", "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.start();

        while (true) {
            List<MessageExt> messageExts = consumer.poll();
            if (messageExts.isEmpty()) {
                continue;
            }
            messageExts.forEach(msg -> {
                System.out.println(String.format("MsgId: %s, MsgBody: %s", msg.getMsgId(), new String(msg.getBody())));
            });
            consumer.commitSync();
        }
    }
}

5.2. rocketmq-spring-boot-starter

依賴

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.2</version>
</dependency>

application.yml

配置項詳見 org.apache.rocketmq.spring.autoconfigure.RocketMQProperties

rocketmq:
  name-server: localhost:9876
  producer:
    group: MyProducerGroup
    send-message-timeout: 10000
  consumer:
    group: MyConsumerGroup

發送消息

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Author: ChengJianSheng
 * @Date: 2023/1/18
 */
@RestController
@RequestMapping("/message")
public class MessageController {

    private String springTopic = "SPRING_TOPIC";
    private String userTopic = "USER_TOPI