好的，我們來(lái)深入理解一下 Apache RocketMQ 中 Message (消息) 這個(gè)核心概念。這份文檔詳細(xì)闡述了消息的定義、在模型中的位置、內(nèi)部屬性、約束和使用建議。

推薦閱讀：RocketMQ 消息Message的結(jié)構(gòu)和使用方式詳解

你可以將 Message 看作是 RocketMQ 系統(tǒng)中數(shù)據(jù)傳輸和處理的最小原子單位。它承載了業(yè)務(wù)數(shù)據(jù)，并附帶了豐富的元信息，是生產(chǎn)者、Broker 和消費(fèi)者之間通信的載體。

1.Message的本質(zhì)定義

• 最小傳輸單元 (Smallest Unit of Data Transmission)：
  • Message 是 RocketMQ 中數(shù)據(jù)傳輸?shù)?strong>基本單元。生產(chǎn)者將業(yè)務(wù)數(shù)據(jù)（負(fù)載）和擴(kuò)展屬性封裝成 Message，發(fā)送給 Broker；Broker 再根據(jù)訂閱關(guān)系將 Message 傳遞給消費(fèi)者。
• 核心特性 (Characteristics)：
  • 不可變性 (Immutability)：
    • 消息一旦生成，其內(nèi)容（特別是系統(tǒng)屬性和負(fù)載）在傳輸和存儲(chǔ)過(guò)程中不會(huì)改變。它被視為一個(gè)“已發(fā)生的事件”。
    • 消費(fèi)者獲取到的消息是只讀的 (read-only)。在 5.x 版本中，這是強(qiáng)約束；在 3.x/4.x 版本中雖無(wú)強(qiáng)約束，但最佳實(shí)踐也建議不要修改。
    • 最佳實(shí)踐：如果需要基于收到的消息發(fā)送新消息，應(yīng)該創(chuàng)建一個(gè)新消息（例如 MessageBuilder.buildFrom(m)），而不是直接修改原消息。
  • 持久性 (Persistence)：
    • 默認(rèn)情況下，RocketMQ 會(huì)將接收到的消息持久化存儲(chǔ)在 Broker 的存儲(chǔ)文件中。這是保證消息不丟失、支持消息追溯和系統(tǒng)故障恢復(fù)的基礎(chǔ)。

2.Message在模型中的位置

• 生命周期流程：
  1. 生產(chǎn) (Produced)：由生產(chǎn)者 (Producer) 創(chuàng)建并初始化。
  2. 發(fā)送 (Sent)：生產(chǎn)者將消息發(fā)送到 Apache RocketMQ Broker。
  3. 存儲(chǔ) (Stored)：Broker 接收到消息后，將其按接收順序存儲(chǔ)在特定 Topic 的某個(gè) Queue 中。
  4. 消費(fèi) (Consumed)：消費(fèi)者 (Consumer) 根據(jù)訂閱關(guān)系，從 Broker 的相應(yīng) Queue 中拉取 (pull) 消息進(jìn)行消費(fèi)。

3.Message的核心內(nèi)部屬性

這些屬性分為系統(tǒng)保留屬性 (System retention attributes) 和可選屬性 (Optional attributes)，以及負(fù)載 (Load)。

系統(tǒng)保留屬性 (由系統(tǒng)或生產(chǎn)者設(shè)置)

• Topic 名稱(chēng) (Topic name)：
  • 作用：標(biāo)識(shí)該消息屬于哪個(gè)邏輯主題。在集群內(nèi)必須唯一。
  • 來(lái)源：由生產(chǎn)者 SDK 設(shè)置。

• 消息類(lèi)型 (Message type)：

  • 作用：定義消息的語(yǔ)義和處理方式。RocketMQ 支持多種類(lèi)型：
    • Normal：普通消息，無(wú)特殊語(yǔ)義。
    • FIFO：順序消息，保證同一消息組 (Message Group) 內(nèi)的消息按發(fā)送順序被消費(fèi)。實(shí)現(xiàn)依賴(lài)于 Queue 的有序性。
    • Delay：延遲消息，可以指定延遲時(shí)間（最大40天），延遲時(shí)間到后才對(duì)消費(fèi)者可見(jiàn)。
    • Transaction：事務(wù)消息，用于實(shí)現(xiàn)分布式事務(wù)，確保本地?cái)?shù)據(jù)庫(kù)操作和消息發(fā)送的最終一致性。

• 消息隊(duì)列 (Message queue)：

  • 作用：指明該消息最終被存儲(chǔ)在哪個(gè)具體的 Queue 中（屬于哪個(gè) Topic 的哪個(gè) Queue）。
  • 來(lái)源：由 Broker 在消息到達(dá)后，根據(jù)路由策略（如輪詢(xún)、哈希等）指定并填充。

• 消息 Offset (Message offset)：

  • 作用：標(biāo)識(shí)該消息在其所屬 Queue 內(nèi)部的物理存儲(chǔ)位置（偏移量）。是 Broker 內(nèi)部管理消息順序和消費(fèi)者消費(fèi)進(jìn)度的關(guān)鍵。
  • 來(lái)源：由 Broker 指定并填充。從 0 開(kāi)始遞增。

• 消息 ID (Message ID)：

  • 作用：消息的全局唯一標(biāo)識(shí)符。在集群內(nèi)絕對(duì)唯一，用于消息追蹤、排查問(wèn)題。
  • 來(lái)源：由生產(chǎn)者客戶端自動(dòng)生成（通常是 32 位的數(shù)字和大寫(xiě)字母組成的字符串）。

可選屬性 (由生產(chǎn)者設(shè)置)

• (可選) 消息 Keys (Message keys)：
  • 作用：為消息設(shè)置一個(gè)或多個(gè)索引鍵。主要用于消息查詢(xún)（通過(guò) Message ID 或 Message Key 在控制臺(tái)或通過(guò) API 查找特定消息）和去重（結(jié)合業(yè)務(wù)邏輯）。
  • 來(lái)源：由生產(chǎn)者客戶端定義。
• (可選) 消息 Tag (Message tag)：
  • 作用：消息的標(biāo)簽，用于消費(fèi)者進(jìn)行消息過(guò)濾。消費(fèi)者可以訂閱特定的 Tag，從而只接收帶有該 Tag 的消息，實(shí)現(xiàn)簡(jiǎn)單的消息分類(lèi)。
  • 來(lái)源：由生產(chǎn)者客戶端定義。
  • 約束：每個(gè)消息只能設(shè)置一個(gè) Tag。
• (可選) 定時(shí)時(shí)間 (Scheduled time)：
  • 作用：配合 Delay 消息類(lèi)型使用，指定消息延遲的具體時(shí)間戳（毫秒級(jí)），而不是延遲時(shí)長(zhǎng)。
  • 來(lái)源：由消息生產(chǎn)者定義。
  • 約束：最大延遲時(shí)間 40 天。

時(shí)間戳屬性

• 消息發(fā)送時(shí)間 (Message sending time)：
  • 作用：記錄消息在生產(chǎn)者客戶端本地被發(fā)送出去的時(shí)間戳（毫秒級(jí)）。
  • 來(lái)源：由生產(chǎn)者客戶端填充。
  • 注意：這是客戶端時(shí)間，可能與 Broker 時(shí)間有偏差。
• 消息存儲(chǔ)時(shí)間 (Message store timestamp)：
  • 作用：記錄消息被Broker 成功寫(xiě)入存儲(chǔ)（落盤(pán)）的時(shí)間戳（毫秒級(jí)）。對(duì)于延遲消息和事務(wù)消息，消費(fèi)者感知到的“有效時(shí)間”通?；诖藭r(shí)間。
  • 來(lái)源：由Broker 填充。
  • 注意：這是 Broker 時(shí)間，是消息在服務(wù)端的“出生”時(shí)間。

重試與自定義

• 重試次數(shù) (Retry times)：
  • 作用：記錄該消息被 Broker 重新投遞給消費(fèi)者的次數(shù)。每次消費(fèi)失敗觸發(fā)重試，次數(shù)加一。第一次消費(fèi)時(shí)為 0。
  • 來(lái)源：由 Broker 標(biāo)記。消費(fèi)者可以獲取此信息以進(jìn)行冪等處理或特殊邏輯。
• 自定義屬性 (Custom attributes)：
  • 作用：生產(chǎn)者可以添加任意的 Key-Value (字符串類(lèi)型) 對(duì)作為擴(kuò)展信息，供業(yè)務(wù)邏輯使用。
  • 來(lái)源：由生產(chǎn)者根據(jù)需要指定。
• 消息負(fù)載 (Message load)：
  • 作用：消息的實(shí)際業(yè)務(wù)數(shù)據(jù)內(nèi)容，即有效載荷 (Payload)。
  • 來(lái)源：由生產(chǎn)者序列化成二進(jìn)制字節(jié)流后設(shè)置。
  • 約束：大小不能超過(guò)系統(tǒng)限制。

4.Message的行為約束

• 大小限制 (Size Limit)：
  • 核心約束：?jiǎn)螚l消息的大小不能超過(guò)上限，否則發(fā)送會(huì)失敗。
  • 默認(rèn)限制：4 MB。這是非常重要的參數(shù)，直接影響網(wǎng)絡(luò)傳輸、存儲(chǔ)和處理性能。

5.Message的使用建議與最佳實(shí)踐

• 避免單條消息過(guò)大 (Overloaded transmission)：
  • 原因：RocketMQ 是事件驅(qū)動(dòng)的中間件。過(guò)大的消息會(huì)：
    • 加重網(wǎng)絡(luò)傳輸負(fù)擔(dān)，增加延遲。
    • 影響錯(cuò)誤重試：重試大消息成本高。
    • 影響流控 (Throttling)：流控粒度可能不夠精細(xì)。
  • 建議：
    1. 嚴(yán)格控制單條消息的數(shù)據(jù)量，使其盡可能小。
    2. 如果業(yè)務(wù)上必須傳輸大量數(shù)據(jù)，強(qiáng)烈建議：
       • 拆分消息：將大數(shù)據(jù)按固定大小拆分成多條小消息。
       • 使用外部存儲(chǔ)：將實(shí)際數(shù)據(jù)（如文件、圖片）存放到對(duì)象存儲(chǔ)（如 OSS）、文件系統(tǒng)或數(shù)據(jù)庫(kù)中，然后在消息的 load 或 custom attributes 中只傳遞數(shù)據(jù)的訪問(wèn)鏈接 (URL) 或 ID。
• 遵守消息不可變性原則 (Immutability)：
  • 正確做法：收到消息后，如果需要轉(zhuǎn)發(fā)或基于它生成新消息，使用 MessageBuilder.buildFrom(m) 這樣的方法創(chuàng)建一個(gè)新消息實(shí)例，然后修改新實(shí)例的屬性（如 Topic, Tag, Load 等）再發(fā)送。
  • 錯(cuò)誤做法：直接調(diào)用 m.update() 修改收到的消息 m 的內(nèi)容，然后發(fā)送。這違反了不可變性原則，可能導(dǎo)致不可預(yù)知的行為或在 5.x 版本中被拒絕。

總結(jié)與核心理解

1. Message 是原子單元：它封裝了業(yè)務(wù)數(shù)據(jù)和元信息，是 RocketMQ 傳輸?shù)淖钚挝弧?/li>
2. Message 是不可變的：內(nèi)容一旦產(chǎn)生，在傳遞過(guò)程中不應(yīng)被修改。最佳實(shí)踐是“讀取-創(chuàng)建-發(fā)送”新消息。
3. Message 是持久化的：默認(rèn)落盤(pán)存儲(chǔ)，保證可靠性。
4. Message 擁有豐富的屬性：
   • Topic/Queue/Offset 定義了其在系統(tǒng)中的位置和順序。
   • Message ID 提供全局唯一標(biāo)識(shí)。
   • Keys/Tag 支持查詢(xún)和過(guò)濾。
   • Message Type 定義了語(yǔ)義（普通、順序、延遲、事務(wù)）。
   • Sending Time/Store Timestamp 記錄了關(guān)鍵時(shí)間點(diǎn)。
   • Retry Times 協(xié)助處理消費(fèi)失敗。
   • Custom Attributes 提供擴(kuò)展能力。
   • Load 承載實(shí)際業(yè)務(wù)數(shù)據(jù)。
5. Message 有嚴(yán)格的大小限制：默認(rèn) 4MB。避免大消息是關(guān)鍵設(shè)計(jì)原則，應(yīng)通過(guò)拆分或外鏈方式處理大數(shù)據(jù)。
6. 最佳實(shí)踐：
   • 合理使用 Tag 進(jìn)行消息過(guò)濾。
   • 利用 Keys 進(jìn)行消息追蹤。
   • 遵守不可變性，通過(guò) buildFrom 創(chuàng)建新消息。
   • 絕對(duì)不要發(fā)送超過(guò) 4MB 的消息，采用拆分或外鏈方案。

簡(jiǎn)而言之，Message 是 RocketMQ 的“信封”和“信件”本身。理解其結(jié)構(gòu)、屬性、約束和最佳實(shí)踐，對(duì)于設(shè)計(jì)高效、可靠、可維護(hù)的消息系統(tǒng)至關(guān)重要。記?。?strong>小消息、不可變、善用屬性、規(guī)避大負(fù)載。

最新評(píng)論