?? RocketMQ 消息（Message）詳解

在 Apache RocketMQ 中，消息（Message） 是數(shù)據(jù)傳輸?shù)淖钚卧?，是生產(chǎn)者與消費(fèi)者之間通信的“載體”。理解 Message 的結(jié)構(gòu)、屬性、生命周期和使用方式，是掌握 RocketMQ 的核心基礎(chǔ)。

推薦閱讀：深入理解Apache RocketMQ 中Message 消息的核心概念

一、什么是 Message？

? 定義：

Message 是 RocketMQ 中封裝實際業(yè)務(wù)數(shù)據(jù)的對象，包含消息體（Body）和一系列元數(shù)據(jù)（如 Topic、Tag、Key、Properties 等），用于在生產(chǎn)者與消費(fèi)者之間傳遞信息。

類比：就像一封信，信紙是內(nèi)容（Body），信封上寫著收件人（Topic）、標(biāo)簽（Tag）、編號（Key）等信息。

二、Message 的核心結(jié)構(gòu)

一個 Message 對象主要由以下幾個部分組成：

三、Message 各字段詳解

1.Topic（主題）

• 消息的邏輯分類，決定消息被發(fā)送到哪個隊列。
• 必須提前創(chuàng)建或允許自動創(chuàng)建。
• 示例：ORDER_TOPIC, USER_LOG_TOPIC

new Message("ORDER_TOPIC", ...);

2.Body（消息體）

• 實際傳輸?shù)臄?shù)據(jù)，必須是字節(jié)數(shù)組。
• 通常通過 JSON、Protobuf、Hessian 等序列化框架編碼。

String content = "{\"orderId\":\"1001\",\"userId\":10086}";
Message msg = new Message(topic, tag, content.getBytes(StandardCharsets.UTF_8));

?? 注意：

• 單條消息大小默認(rèn)最大 4MB（可配置）
• 過大消息會影響性能，建議拆分或使用外部存儲（如上傳文件后傳 URL）

3.Tags（標(biāo)簽）

• 用于對同一 Topic 下的消息進(jìn)行二次分類。
• 消費(fèi)者可通過 subscribe("Topic", "TagA || TagB") 進(jìn)行過濾。

// 發(fā)送
new Message("ORDER_TOPIC", "CREATE", "創(chuàng)建訂單".getBytes());
new Message("ORDER_TOPIC", "PAY", "支付完成".getBytes());
// 訂閱 CREATE 類型消息
consumer.subscribe("ORDER_TOPIC", "CREATE");

? 優(yōu)勢：輕量級過濾，避免消費(fèi)者接收無關(guān)消息。

?? 注意：Tags 是字符串匹配，不支持正則（但支持 * 通配和 || 多選）

4.Keys（消息鍵）

• 為消息設(shè)置唯一標(biāo)識或業(yè)務(wù)主鍵（如訂單號、用戶ID）。
• 支持通過 mqadmin queryMsgByKey 命令查詢消息。
• 支持索引，便于問題排查。

Message msg = new Message(...);
msg.setKeys("ORDER_20240501001");

? 建議：關(guān)鍵業(yè)務(wù)消息務(wù)必設(shè)置 Keys，便于追蹤。

5.Properties（屬性）

• 鍵值對形式的擴(kuò)展字段，可用于：
  • 存儲自定義上下文（如 traceId、tenantId）
  • RocketMQ 內(nèi)部使用（如 RECONSUME_TIME、DELAY、TRAN_MSG）

msg.putUserProperty("traceId", "abc123");
msg.putUserProperty("source", "web");

?? 注意：系統(tǒng)屬性以 PREFIX_SYS_PROP 開頭，不要沖突。

6.DelayTimeLevel（延遲級別）

• 設(shè)置消息延遲投遞時間，實現(xiàn)“定時任務(wù)”功能。
• 取值范圍：1~18，對應(yīng)不同延遲時間：

Message msg = new Message("DELAY_TOPIC", "TAG", "延遲消息".getBytes());
msg.setDelayTimeLevel(5); // 延遲1分鐘
producer.send(msg);

?? 注意：延遲消息不保證精確時間，存在輕微誤差。

四、Message 的生命周期

1. 生產(chǎn)者創(chuàng)建 Message 對象
   ↓
2. 發(fā)送到 Broker（寫入 CommitLog）
   ↓
3. 構(gòu)建 ConsumeQueue 和 IndexFile
   ↓
4. 消費(fèi)者拉取消息（根據(jù) Topic + Queue）
   ↓
5. 處理成功 → 提交 Offset
   ↓
6. 消息過期（默認(rèn) 72 小時）→ 被刪除

? 消息是持久化存儲的，即使消費(fèi)者未上線，消息也不會丟失。

五、Message 的存儲機(jī)制

雖然 Message 是邏輯對象，但在 Broker 端有嚴(yán)格的物理存儲結(jié)構(gòu)：

1.CommitLog

• 所有消息按到達(dá)順序追加寫入 CommitLog 文件（順序?qū)懀咝阅埽?/li>
• 每個消息包含：Topic、Queue、Body、Properties 等完整信息

2.ConsumeQueue

• 每個 Topic 的每個 MessageQueue 對應(yīng)一個 ConsumeQueue
• 存儲消息的邏輯偏移量、大小、物理位置，用于快速定位消息

ConsumeQueue/{Topic}/{QueueId}/
   ├── 00000000000000000000
   └── ...

3.IndexFile

• 可選索引文件，支持通過 Keys 或時間范圍 查詢消息
• 用于排查問題（如“查找某個訂單的消息”）

IndexFile/index_1714567890000

六、Message 的發(fā)送方式回顧

所有方式發(fā)送的都是 Message 對象。

七、最佳實踐與注意事項

八、常見問題排查

? 總結(jié)：Message 核心要點(diǎn)

?? 一句話總結(jié)：

Message 是 RocketMQ 的“數(shù)據(jù)包” —— 它不僅是業(yè)務(wù)數(shù)據(jù)的載體，更是路由、過濾、追蹤、延遲、事務(wù)等功能的基礎(chǔ)。
設(shè)計好 Message 的結(jié)構(gòu)與屬性，才能讓消息系統(tǒng)真正高效、可靠、易維護(hù)。

掌握 Message，你就掌握了 RocketMQ 的“語言”。

到此這篇關(guān)于RocketMQ 消息Message的結(jié)構(gòu)和使用方式詳解的文章就介紹到這了,更多相關(guān)RocketMQ 消息Message內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論