Quartz 5

訊息閘道

Properties1
tagsintegration, messaging, eip, messaging-endpoints

8 min read

🎯 什麼情境該想到我

當你「不想讓應用程式碼直接依賴訊息收發 API，希望把訊息操作包成普通的方法呼叫」的時候。

⚙️ 怎麼用（步驟 / 公式）

意圖：用一個類別封裝訊息系統的存取，對外只提供領域相關的方法，讓應用程式不需要直接接觸訊息 API。

做法 / 結構要點：

  • 建立 Gateway 類別，內部處理連線、通道、送收訊息、組 / 解訊息等細節。
  • 對外暴露如 submitOrder(order) 這類業務方法。
  • 應用只呼叫這些方法，完全不感知底層是訊息系統。

★ Go 的實作：介面定義在「使用它的那一側」

// ★ 在業務 package 裡宣告你需要的最小介面（不是在 broker 的 package）
package order
 
type EventPublisher interface {
    PublishOrderPlaced(ctx context.Context, e OrderPlaced) error
}
 
func (s *Service) Place(ctx context.Context, req Request) error {
    // …業務邏輯…
    return s.pub.PublishOrderPlaced(ctx, OrderPlaced{OrderID: o.ID, …})
}
// ★ 閘道的實作在另一個 package，它才認識 kafka
package kafkagw
 
type Gateway struct{ w *kafka.Writer }
 
func (g *Gateway) PublishOrderPlaced(ctx context.Context, e order.OrderPlaced) error {
    b, err := json.Marshal(envelope(e))            // ★ 信封在這裡包（[[信封包裝器]]）
    if err != nil { return fmt.Errorf("marshal: %w", err) }
    return g.w.WriteMessages(ctx, kafka.Message{Key: []byte(e.OrderID), Value: b})
}

★★ 這是 Go 相對其他語言最便宜的一次抽象：隱式滿足讓 kafkagw.Gateway
自動滿足 order.EventPublisher，而 kafkagw 完全不必 import order 的介面宣告
（見 工具-針對介面編程、工具-Go介面小而隱式）。

★★ 收益一：測試不需要 broker

type fakePublisher struct{ sent []order.OrderPlaced }
 
func (f *fakePublisher) PublishOrderPlaced(_ context.Context, e order.OrderPlaced) error {
    f.sent = append(f.sent, e); return nil
}
 
func TestPlace(t *testing.T) {
    pub := &fakePublisher{}
    svc := order.New(repo, pub)
    // …斷言 pub.sent 的內容
}

★ 十行的 fake 取代一個 testcontainer。
⚠️ 但要配至少一個「真的連 broker」的整合測試——
fake 不會告訴你「訊息大小超限」「序列化失敗」「topic 不存在」。

★★ 收益二：橫切關注點有一個統一的地方放

★ 閘道是唯一該放這些東西的地方：
① 信封的組裝（message-id、correlation-id、trace）
② 序列化與 schema 版本
③ 重試與退避（發送端）
④ metrics（發送量、失敗率、延遲）
⑤ ★ Outbox 的寫入（見 [[工具-Transactional-Outbox-Pattern]]）

★ 沒有閘道的後果很具體：這五件事會散落在每一個發送點，
而且一定有幾個地方漏掉其中一項（最常漏的是 correlation-id）。

★★ 閘道的介面該多細：這是唯一需要判斷的地方

// ① ★ 一個方法對一種訊息 —— 型別安全，但方法會很多
PublishOrderPlaced(ctx, OrderPlaced) error
PublishOrderCanceled(ctx, OrderCanceled) error
 
// ② 泛型化 —— 方法少，但呼叫端要自己決定 topic
Publish[T any](ctx context.Context, topic string, msg T) error
 
// ③ ★★ 折衷：泛型 + 型別化的 Topic（見 [[資料型別通道]]）
Publish(ctx, OrderPlacedTopic, OrderPlaced{…})    // ★ 傳錯型別編譯不過

★ 判準：這個 package 發幾種訊息？
少（2–5 種）→ ①，明確好讀。
多 → ③。
⚠️ 不要選 ②：呼叫端拼字串 topic 是拼錯就上線的那種 bug。

★ 消費端也需要閘道

// ★ 業務邏輯只認識「處理一則領域事件」，不認識 broker 的 Message
type OrderPlacedHandler interface {
    HandleOrderPlaced(ctx context.Context, e OrderPlaced) error
}
// ★ 閘道負責：解信封、解序列化、錯誤分類、ack／nack、冪等去重

★★ 消費端閘道更有價值，因為它承載了整個消費迴圈的正確性
（ack 時機、錯誤分類、每則訊息獨立超時——見 訊息端點）。
這些東西寫錯一次就會在生產環境丟訊息，所以只該寫一次。

★ 兄弟辨析

差別
訊息對應器★ 只管「領域物件 ↔ 訊息」的對應；閘道還管 broker API
通道配接器★ 在應用程式外面；閘道在裡面
服務啟動器讓同一個服務同時能被訊息與同步呼叫觸發
閘道★ PoEAA 的一般化版本（封裝任何外部資源）
外觀模式行程內的同一想法

🧪 我實際套用的紀錄

  • （待填）

⚠️ 注意 / 什麼時候不適用

  • 閘道方法可設計成同步（阻塞等回覆）或非同步（帶回呼），要選對。

  • 封裝良好時，抽換訊息中介或改走非訊息呼叫都較容易。

  • 別在閘道裡摻入業務邏輯，它只負責轉接。

  • ★★ 介面要定義在使用它的 package（業務側），不是 broker 的 package。Go 的隱式滿足讓這件事免費。

  • ★★ 消費端閘道比發送端更有價值：ack 時機、錯誤分類、每則訊息獨立超時——寫錯一次就會丟訊息，所以只該寫一次。

  • ★ fake 不能取代整合測試。它不會告訴你訊息超限、序列化失敗、topic 不存在。

  • ★ 不要讓呼叫端拼字串 topic。用型別化的 Topic，拼錯就編譯不過。

  • ★ 閘道是橫切關注點唯一該放的地方（信封、序列化、重試、metrics、Outbox）。散在每個發送點一定會漏掉 correlation-id。

  • 閘道不要塞業務邏輯。「金額大於一萬要多發一則」屬於業務層，不屬於閘道。

  • 閘道的介面就是契約。方法簽章一改，所有使用它的地方都要改——所以介面要小。

🔗 相關工具

Graph View

Backlinks