訊息端點

🎯 什麼情境該想到我

當你「要讓一個應用程式連上訊息系統來收發訊息，又不想讓訊息 API 的細節滲進商業邏輯」的時候。

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

意圖：用一段專門的膠合程式碼（端點），把應用程式接上訊息通道，負責把應用資料與訊息互相轉換、以及實際收發。

做法 / 結構要點：

端點是應用與訊息系統之間的接口：一端講應用的語言（方法呼叫、領域物件），一端講訊息系統的語言（訊息、通道）。
發送端：把應用資料包成訊息送上通道；接收端：從通道取訊息、拆出資料交給應用處理。
把訊息收發的細節封裝在端點，應用其餘部分不必知道訊息系統存在。

★★ 端點的核心紀律：應用程式不該認識 broker 的 API

// ❌ 業務邏輯裡出現 broker 的型別 —— 從此無法在沒有 Kafka 的情況下測試
func PlaceOrder(o Order, w *kafka.Writer) error {
    b, _ := json.Marshal(o)
    return w.WriteMessages(ctx, kafka.Message{Value: b})    // ★ 業務碼依賴 kafka
}
 
// ✅ 端點把 broker 隔在外面（這就是 [[訊息閘道]]）
type OrderPublisher interface {                     // ★ 介面定義在消費方
    Publish(ctx context.Context, e OrderPlaced) error
}
func PlaceOrder(ctx context.Context, o Order, pub OrderPublisher) error {
    return pub.Publish(ctx, OrderPlaced{OrderID: o.ID, …})
}

★ 收益：測試用一個十行的 fake，不必起 broker（見工具-打破依賴以便測試）。
Go 的隱式介面讓這件事幾乎免費——介面定義在使用它的 package，broker 的套件完全不知道它存在
（見工具-針對介面編程）。

★ 端點家族的三個決定（照這個順序問）

① 怎麼收？   拉（[[輪詢式消費者]]）還是推（[[事件驅動消費者]]）
             ★ Kafka／SQS 是拉；RabbitMQ 的 basic.consume 是推
             ★ 拉天生有背壓；推需要自己限制 in-flight 數量（prefetch）

② 幾個人收？ 一個（[[點對點通道]]）還是每人一份（[[發布訂閱通道]]）
             要水平擴展 → [[競爭消費者]]
             ★ Kafka：並行度上限 = 分區數（多的消費者會閒置）

③ 可靠性？   ★ [[交易式用戶端]]（ack 時機）＋ [[冪等接收者]]（必備）
             ＋ [[死信通道]]（失敗的去處）

★★ 消費端的正確骨架（Go）

func Run(ctx context.Context, c Consumer, h Handler) error {
    for {
        msg, err := c.Fetch(ctx)                       // ★ 拉模式天生有背壓
        if err != nil {
            if errors.Is(err, context.Canceled) { return nil }   // ★ 優雅關閉
            return fmt.Errorf("fetch: %w", err)
        }
 
        // ★ 每則訊息一個獨立的處理超時 —— 不要用外層的 ctx
        mctx, cancel := context.WithTimeout(ctx, 30*time.Second)
        err = h.Handle(mctx, msg)
        cancel()
 
        switch {
        case err == nil:
            _ = msg.Ack()                              // ★★ ack 在成功「之後」
        case errors.Is(err, ErrPermanent):
            _ = msg.Nack(false)                        // ★ 不可重試 → 直接進 DLQ
        default:
            _ = msg.Nack(true)                         // ★ 可重試 → 重送
        }
    }
}

三個必做的事：

★ ack 在處理成功之後（先 ack 再處理 = 失敗就丟訊息）。
★ 區分「可重試」與「不可重試」。ErrInvalidArgument 重試一萬次也一樣，直接進 DLQ。
★ 每則訊息獨立超時。用外層 ctx → 一則卡住會拖垮整個消費迴圈。

★ 兩個最常被忽略的細節

// ① ★★ 處理時間超過「可見性超時／ack 期限」→ 訊息被重送，於是同一則被處理兩次
//    SQS 預設 visibility timeout = 30 秒（查證 2026-09-25，最大 12 小時）
//    ★ 處理要 5 分鐘 → 要嘛調高，要嘛週期性延長（heartbeat）
//    → 這是「明明沒有重送卻出現重複」的頭號原因
 
// ② ★ 優雅關閉：收到 SIGTERM 之後
//    停止 Fetch → 等手上那則處理完 → ack → 才退出
//    ★ 直接 os.Exit 會讓手上的訊息重送（如果還沒 ack），或遺失（如果已 ack）

★ 兄弟辨析（端點家族 11 個一次看懂）

分類	模式	一句話
封裝	訊息閘道	★ 把 broker API 藏在應用程式介面後面
	訊息對應器	領域物件 ↔ 訊息的對應
	服務啟動器	★ 讓同一個服務同時能被訊息與同步呼叫觸發
收取方式	輪詢式消費者	主動拉（★ 天生有背壓）
	事件驅動消費者	被推（★ 要自己限流）
	選擇性消費者	只收符合條件的
擴展	競爭消費者	多個消費者搶同一個通道
	訊息分派器	一個消費者收、分給多個 worker
可靠性	交易式用戶端	ack／交易邊界
	冪等接收者	★ 重複投遞只生效一次（必備）
	持久訂閱者	離線期間的訊息不丟

🧪 我實際套用的紀錄

（待填）

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

端點是專為某通道與某應用量身打造的，通常不可通用、會綁定雙方。
收發策略（輪詢/事件驅動、交易、併發）需依需求選擇，設計不當會成為瓶頸或漏訊息。
★★ ack 一定在處理成功之後。先 ack 再處理 = 處理失敗就丟訊息。
★★ 處理時間超過可見性超時／ack 期限會導致重送。SQS 預設 30 秒（2026-09-25 查證）——這是「明明沒重送卻出現重複」的頭號原因。
★ 業務邏輯不該認識 broker 的型別。用訊息閘道隔開，測試才能用 fake。
★ 每則訊息要有獨立的處理超時。用外層 ctx，一則卡住會拖垮整個消費迴圈。
★ 要區分可重試與不可重試的錯誤。InvalidArgument 重試一萬次結果一樣，直接進 DLQ。
★ 優雅關閉要等手上那則處理完。直接退出會造成重送或遺失。
★ Kafka 的並行度上限是分區數。加消費者超過分區數只是讓它們閒置。
推模式要自己限制 in-flight 數量（prefetch／concurrency），否則會被灌爆。

🔗 相關工具

訊息通道 Message Channel（端點所連接的管道）
訊息 Message（端點負責封裝與拆解的資料封包）
通道配接器 Channel Adapter（一種特殊端點：把不用訊息的既有系統接上通道）
企業整合模式
訊息閘道、訊息對應器、服務啟動器 —— 端點的封裝
輪詢式消費者、事件驅動消費者、選擇性消費者 —— 怎麼收
競爭消費者、訊息分派器 —— 怎麼擴展
交易式用戶端、冪等接收者、持久訂閱者 —— 可靠性
工具-MQ消費端防禦三原則 —— 消費端的完整防禦
工具-打破依賴以便測試、工具-針對介面編程 —— 為什麼介面該定義在消費方

Quartz 5

Explorer