訊息分派器

🎯 什麼情境該想到我

當你「想用一個消費者統一從通道收訊息，再把訊息分派給多個處理者（performer）並行處理，以集中協調分工」的時候。

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

意圖：以單一消費者接收通道上的所有訊息，再依規則把每則訊息交給對應的 performer 處理，藉此協調並行與控制。

做法 / 結構要點：

Dispatcher 從通道收訊息。
依內容 / 類型決定交給哪個 performer（可用執行緒池並行）。
performer 負責實際處理；Dispatcher 專責收與派。

★ 和競爭消費者的界線（這兩個最常被混為一談）

★ [[競爭消費者]]：N 個「消費者」（通常是 N 個 process／pod）搶同一個通道
★ 訊息分派器  ：1 個消費者從通道取訊息，在「行程內」分給 N 個 worker

	競爭消費者	訊息分派器
並行單位	process／pod	★ goroutine／thread
Kafka 並行上限	★ 分區數	★ 不受分區數限制
故障隔離	★ 好（一個 pod 掛不影響其他）	★ 差（行程掛掉手上全部訊息受影響）
連線數	★ N 倍	★ 1 份（共用連線池）
ack／offset 管理	★ 簡單（broker 幫你分配）	★★ 困難（見下）

★ 選它的唯一好理由：Kafka 的分區數不夠，而你需要更高並行度。
（或是要共用昂貴的連線／快取。）

★★ Go 的實作，以及 offset 提交的陷阱

// ★ 簡單版：worker pool + semaphore
sem := make(chan struct{}, concurrency)
for {
    msg, err := consumer.Fetch(ctx)
    if err != nil { return err }
    sem <- struct{}{}
    go func(m Message) {
        defer func() { <-sem }()
        if err := handle(ctx, m); err == nil { m.Ack() }
    }(msg)
}

★★ 這段在 Kafka 上是錯的，因為 Kafka 的 ack 是「提交 offset」，而 offset 是單調的：

拉到 offset 100–110，10 個 worker 並行處理
★ offset 105 先完成 → 若提交 105，就等於宣告「100–105 都處理完了」
★★ 程式此刻崩潰 → 100–104 永遠不會被重讀 ⇒ 靜默遺失

// ✅ 正確做法：維護 in-flight 集合，只提交「連續完成的最小未完成 offset - 1」
type offsetTracker struct {
    mu        sync.Mutex
    inFlight  map[int64]bool   // ★ 已派出但未完成
    committed int64
}
 
func (t *offsetTracker) done(off int64) (commitTo int64, ok bool) {
    t.mu.Lock(); defer t.mu.Unlock()
    delete(t.inFlight, off)
    next := t.committed + 1
    for !t.inFlight[next] && next <= t.maxSeen {   // ★ 往前推進到第一個未完成的
        t.committed = next; next++
    }
    return t.committed, true
}

★★ 這是實務上最容易寫錯、而且錯了不會有任何錯誤訊息的一段。
★ 務實建議：如果不是真的被分區數卡住，用「多 pod + 加分區」simpler 十倍。

★ RabbitMQ／SQS 上簡單得多

// ★ 它們是 per-message ack，沒有 offset 單調性的問題
//   → worker 各自 ack 自己的訊息即可
//   ⚠️ 但仍要注意：prefetch 必須 ≥ concurrency，否則 worker 會餓死

★ 一個常見的設定錯誤：concurrency = 20 但 prefetch = 5
→ 只有 5 個 worker 有事做，另外 15 個永遠閒置。

★★ 三個必須處理的問題

① ★★ 順序完全消失
   同一個實體的兩則訊息被兩個 worker 並行處理 → ★ 後到的可能先寫入
   → ★ 需要順序時：用實體 ID 做「行程內分片」
      worker := hash(entityID) % concurrency  ← ★ 同一實體固定給同一個 worker
      ⚠️ 代價：熱門實體變成熱點

② ★★ 優雅關閉
   收到 SIGTERM → ★ 必須等所有 in-flight 的 worker 完成並 ack
   → 直接退出：未 ack 的重送（可接受）；已 ack 未完成的 ★★ 永久遺失

③ ★ 一個 worker 卡住會吃掉一個並行額度
   → ★ 每則訊息必須有獨立超時，否則並行度會被慢訊息慢慢吃光

★ 兄弟辨析

	差別
競爭消費者	★ 多個消費者實例，broker 負責分配
輪詢式消費者／事件驅動消費者	收取方式；分派器是「收到之後怎麼分」
選擇性消費者	依條件挑訊息，不是分派
工具-靠通訊共享記憶體	Go 的 worker pool 寫法

🧪 我實際套用的紀錄

（待填）

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

與競爭消費者達成類似並行效果，但控制點集中在 Dispatcher（可決定分派策略、維持部分順序）。
Dispatcher 本身可能成為單點瓶頸。
分派與處理分離，複雜度較高。
★★ Kafka 上的 offset 提交極易寫錯：只能提交「連續完成的最小未完成 offset - 1」，否則會靜默遺失訊息。
★★ 不是真的被分區數卡住，就用「多 pod + 加分區」——simpler 十倍，而且故障隔離更好。
★★ 順序完全消失。需要順序就用 hash(entityID) % concurrency 做行程內分片（代價是熱點）。
★★ 優雅關閉必須等所有 in-flight worker 完成並 ack。 已 ack 但未完成的訊息會永久遺失。
★ prefetch 必須 ≥ concurrency，否則多數 worker 永遠閒置。
★ 每則訊息要有獨立超時，否則慢訊息會慢慢吃光並行額度。
故障隔離比競爭消費者差：行程掛掉時手上所有訊息都受影響。
在 RabbitMQ／SQS 上簡單得多（per-message ack，沒有 offset 單調性問題）。

🔗 相關工具

競爭消費者 Competing Consumers（同家族，達成類似並行的另一種做法）
選擇性消費者 Selective Consumer（同家族，改由消費端依條件挑訊息）
企業整合模式
競爭消費者 —— ★ 多實例的對照
輪詢式消費者、事件驅動消費者 —— 收取方式
訊息序列 —— 讓處理與順序無關
交易式用戶端 —— ack 時機
工具-靠通訊共享記憶體 —— Go 的 worker pool
工具-資料分區 —— 分區數與並行度

Quartz 5

Explorer