rocketmq 快速入門

請參考我們的各種配置（podman），瞭解如何安裝。

概念介紹

RocketMQ 是阿里巴巴開源、Apache 頂級專案的分散式訊息中介軟體（middleware），核心元件（component）：

• NameServer：服務發現與路由
• Broker：訊息儲存、投遞、拉取
• Producer：訊息生產者（傳送訊息）
• Consumer：訊息消費者（訂閱並消費訊息）
• Topic/Tag：主題/標籤，用於訊息分組與過濾

生產與消費模型：Producer 將訊息傳送到某個 Topic；Broker 進行持久化並供 Consumer 拉取；Consumer 以叢集（cluster）或廣播模式消費。

程式碼範例本章以 Go 為例（偽程式碼/示意），不同 SDK 方法名稱略有差異，請以實際版本為準。

依據傳送特性分類

1. 同步傳送

同步傳送會等待 Broker 返回傳送結果，適合對可靠性有要求的場景（例如下單、建立訂單事件）。

// 同步發送
msg := rocketmq.NewMessage("OrderTopic", []byte("order-created"))
res, err := producer.SendSync(context.Background(), msg)
if err != nil {
    // 失敗處理/重試
}
log.Printf("SendOK: %v", res)

2. 非同步傳送

非同步傳送不會阻塞主執行緒（thread），透過回呼（callback）取得結果，適合鏈路較長或吞吐量（throughput）要求高的場景。

// 異步發送
msg := rocketmq.NewMessage("LogTopic", []byte("user-action"))
producer.SendAsync(context.Background(), msg, func(res *SendResult, err error) {
    if err != nil {
        // 記錄失敗，後續重試
        return
    }
    log.Printf("AsyncSendOK: %v", res)
})

3. 單向傳送（OneWay）

單向傳送只負責「盡力而為」地發出訊息，不關心結果，適用於日誌收集、埋點等對可靠性要求低的場景。

// 單向發送
_ = producer.SendOneWay(context.Background(), rocketmq.NewMessage("TraceTopic", []byte("trace")))

依據使用功能特性分類

1. 普通訊息（訂閱）

最常見的發布/訂閱模型。消費者可採用叢集（cluster）模式（負載平衡）或廣播模式（每個消費者都收到）。

// 消費者訂閱普通消息
consumer.Subscribe("OrderTopic", rocketmq.FilterByTag("created"), func(msg *MessageExt) ConsumeResult {
    // 冪等處理
    // 業務邏輯...
    return ConsumeSuccess
})

要點：

• 冪等性：使用業務唯一鍵或去重表避免重複消費
• 重試與死信：失敗返回重試，超過閾值進入 DLQ

2. 順序訊息

順序訊息分為全域順序和分區順序。常見做法是按業務鍵（例如訂單號）將訊息路由到同一個佇列（queue），保證「同一訂單」的訊息有序。

// 生產者按業務鍵選擇隊列（示意）
shardingKey := orderID
msg := rocketmq.NewMessage("OrderSeqTopic", []byte("status-changed"))
msg.WithShardingKey(shardingKey)
_, _ = producer.SendSync(ctx, msg)

注意：要保證同一業務鍵落在同一個佇列（queue），消費者通常單執行緒（single thread）或按佇列（queue）串列處理。

3. 延遲訊息（定時/延遲）

用於在指定時間後再投遞給消費者，例如「訂單逾時取消」、「支付結果稍後檢查」等。

// 發送 30s 後可見的延時消息（不同 SDK 可用 delayLevel 或 deliverTime）
msg := rocketmq.NewMessage("DelayTopic", []byte("close-order"))
msg.SetDelay(time.Second * 30)
_, _ = producer.SendSync(ctx, msg)

實踐要點：

• 合理的延遲等級/絕對投遞時間
• 消費端仍需冪等與補償

4. 事務訊息（分散式事務）

用於保證「本地事務 + 訊息」最終一致。流程：傳送半訊息 → 執行本地事務 → 依據結果 Commit/Rollback；Broker 未收到確認會回查業務狀態。

sequenceDiagram
  participant P as Producer
  participant MQ as RocketMQ
  participant DB as LocalDB
  P->>MQ: 發送半消息
  P->>DB: 執行本地事務
  alt 成功
    P->>MQ: Commit
    MQ->>C: 投遞正式消息
  else 失敗
    P->>MQ: Rollback
  end
  MQ->>P: 回查未確認事務

更多細節可參考本儲存庫（repository） 013.md 中「事務訊息」與「TCC/本地訊息表」等章節。

生產者與消費者快速範例

// Producer 初始化（示意）
producer, _ := rocketmq.NewProducer(rocketmq.ProducerConfig{
    NameServer: []string{"127.0.0.1:9876"},
    Group:      "demo-producer-group",
})
defer producer.Shutdown()

// Consumer 初始化（示意）
consumer, _ := rocketmq.NewPushConsumer(rocketmq.ConsumerConfig{
    NameServer: []string{"127.0.0.1:9876"},
    Group:      "demo-consumer-group",
    Model:      rocketmq.Clustering, // 或 Broadcasting
})
defer consumer.Shutdown()

分散式事務訊息的優勢

• 解耦：上下游透過事件協作，降低強耦合
• 彈性與可擴展：非同步削峰，支援高併發
• 可靠性：訊息持久化，失敗可重試/對帳
• 最終一致：在 AP 取捨下透過補償與回查達到一致

適用場景：訂單建立/支付、庫存扣減、積分/優惠券發放、資金記帳、狀態同步等。

常見實踐建議

• 消費端冪等：唯一業務鍵、去重表、樂觀鎖
• 失敗重試與死信佇列（DLQ）配置
• 監控與告警：積壓、失敗率、耗時
• 結合延遲訊息實現「逾時關閉/回查」
• 事務訊息只在關鍵鏈路使用，其餘用本地訊息表或最大努力通知