?? RocketMQ 消息（Message）詳解

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

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

一、什么是 Message？

? 定義：

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

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

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

一個(gè) Message 對(duì)象主要由以下幾個(gè)部分組成：

三、Message 各字段詳解

1.Topic（主題）

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

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

2.Body（消息體）

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

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

?? 注意：

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

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

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

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

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

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

4.Keys（消息鍵）

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

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

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

5.Properties（屬性）

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

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

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

6.DelayTimeLevel（延遲級(jí)別）

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

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

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

四、Message 的生命周期

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

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

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

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

1.CommitLog

• 所有消息按到達(dá)順序追加寫(xiě)入 CommitLog 文件（順序?qū)?，高性能?/li>
• 每個(gè)消息包含：Topic、Queue、Body、Properties 等完整信息

2.ConsumeQueue

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

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

3.IndexFile

• 可選索引文件，支持通過(guò) Keys 或時(shí)間范圍 查詢(xún)消息
• 用于排查問(wèn)題（如“查找某個(gè)訂單的消息”）

IndexFile/index_1714567890000

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

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

七、最佳實(shí)踐與注意事項(xiàng)

八、常見(jiàn)問(wèn)題排查

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

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

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

掌握 Message，你就掌握了 RocketMQ 的“語(yǔ)言”。

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

最新評(píng)論