Quartz 5

聚合器

Properties1
tagsintegration, messaging, eip, message-routing

9 min read

🎯 什麼情境該想到我

當你「要把一組相關的多則訊息，收集、關聯並合併成單一訊息」的時候。

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

意圖：把多則相關訊息組合成一則。

做法要點（聚合器是有狀態的元件）：

  • 關聯（correlation）：用某個依據把屬於同一組的訊息歸攏在一起。
  • 完成條件：判斷何時算收齊（收滿數量 / 滿足某條件 / 逾時）。
  • 聚合策略：把收齊的訊息合併成一則輸出。
  • 常搭配逾時，避免永遠等不到而卡住。

★★ 這是整個 EIP 型錄裡最難正確實作的模式

理由：它是有狀態的、需要逾時、需要記憶體上限、需要重啟後恢復——
四件事每一件都能讓它在生產環境爆掉。

★ 三個必須決定的策略

① ★ 關聯（correlation）：哪些訊息屬於同一組？
   → 通常是 [[關聯識別碼]] 或 [[分割器]] 給的 SequenceID

② ★★ 完成條件：什麼時候算收齊？—— 最難的一個
   a. 等到指定數量（Total）  ← ★ 需要分割器告訴你總數
   b. 等到 IsLast 標記        ← ★ 總數事先未知時（串流）
   c. 等到某個條件成立        ← 「已經有三家報價就夠了」
   d. ★★ 等到逾時            ← 永遠要有這一條作為後盤
   e. 外部訊號               ← 收到「結束」命令

③ ★ 聚合策略：怎麼合？
   全部串起來 / 取最好的一個 / 求和 / 只取最新

★★ (d) 逾時不是「可選的第五種」，是「其他四種的必要後盤」：
缺漏的那一則可能永遠不來（進了 DLQ、被判無效、生產者崩潰）。
→ 沒有逾時的聚合器會永久累積半成品，最後 OOM。

★★ Go 的實作要點：狀態必須在行程外

// ❌ 狀態放記憶體 —— 重啟就全丟，而且無法水平擴展
var pending = map[string][]Message{}
 
// ✅ 狀態放資料庫或 Redis
type AggregateState struct {
    CorrelationID string
    Received      map[int]json.RawMessage   // position → payload
    ExpectedTotal int                        // ★ -1 表示未知
    FirstSeenAt   time.Time                  // ★ 逾時的基準
    CompletedAt   *time.Time
}

三個必須做的事：

① ★★ 併發安全的「加入 + 檢查完成」必須原子
   兩個 goroutine 同時加入最後兩則 → ★ 都認為自己是最後一個 ⇒ 聚合輸出兩次
   → ★ 用資料庫交易 + SELECT ... FOR UPDATE，或 Redis 的 Lua script

② ★★ 一定要有獨立的「掃逾時」排程器
   靠訊息驅動無法發現逾時 —— 逾時的定義就是「訊息沒來」
   → 一個 cron 掃 FirstSeenAt + timeout < now 的項目

③ ★ 要有數量與大小上限
   一個壞的生產者發十萬個不同 correlation id → ★ 你的狀態儲存爆掉
   → 上限 + 超過時拒絕新的組並告警

★★ 不完整的聚合怎麼輸出：這個決定必須明確

// ★ 逾時時的三種處置，要由業務決定，不能預設
switch policy {
case DiscardIncomplete:
    // ★ 丟掉 —— ★★ 一定要記錄「丟了什麼、缺了哪幾則」，否則是靜默資料遺失
case EmitPartial:
    // ★ 輸出部分結果，並標明 incomplete: true 與缺失的 position
    //   → 下游必須能處理「不完整」這件事，這是契約的一部分
case FailLoudly:
    // ★ 送 [[無效訊息通道]] 並告警，等人處理
}

★ 最常見的錯誤是「靜默丟棄不完整的組」——
比價流程少了一家報價，結果直接用兩家的價格成交，而沒有任何人知道。

★ 能不要用聚合器嗎？（先試這三條路）

① ★★ 讓下游自己處理「部分到達」
   不要等齊，收到一則就更新一次讀模型（配版本號防舊蓋新）
   → ★ 無狀態、可並行、沒有逾時問題。這是最好的答案
② ★ 用資料庫當聚合器
   每則訊息 UPSERT 一列，用一個查詢判斷「是否已收齊」
   → ★ 狀態、併發、持久化全部交給資料庫，不要自己寫
③ ★ 用現成的 stream processing
   Kafka Streams 的 windowed aggregation、Flink —— 它們把狀態管理做好了

★★ ② 是實務上最划算的做法：一張 received_parts 表 + 唯一索引 (correlation_id, position)，
加上一個「count(*) = expected_total」的判斷。
你不需要自己實作一個有狀態的分散式元件。

★ 兄弟辨析

差別
分割器★ 反向：1 → N
重排序器★ 同樣有狀態，但只排序不合併（輸出仍是 N 則）
組合訊息處理器分割 + 處理 + 聚合的完整組合
散發-聚集送給多方 + 用聚合器收回覆
流程管理器★ 更一般化：可分支、可多回合、可補償

🧪 我實際套用的紀錄

  • （待填）

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

  • 有狀態，需管理記憶體與逾時。

  • 完成條件的設計是關鍵（等齊 / 等到條件 / 等到逾時）。

  • 是分割器、散發-聚集等模式的收斂端。

  • ★★ 逾時不是可選的。缺漏的那一則可能永遠不來（進 DLQ、被判無效、生產者崩潰）。沒有逾時的聚合器會 OOM。

  • ★★ 「加入 + 檢查完成」必須原子。否則兩個 goroutine 都認為自己是最後一個 ⇒ 聚合輸出兩次。

  • ★★ 必須有獨立的掃逾時排程器。靠訊息驅動發現不了逾時——逾時的定義就是「訊息沒來」。

  • ★★ 不完整聚合的處置必須明確，而且不可以靜默丟棄。「比價少一家就直接成交」是真實的事故樣貌。

  • ★ 狀態必須在行程外，否則重啟全丟、無法水平擴展。

  • ★ 要有組數與大小上限。壞的生產者發十萬個不同 correlation id 會撐爆狀態儲存。

  • ★★ 先試「讓下游處理部分到達」（配版本號）或「用資料庫當聚合器」——不要自己寫一個有狀態的分散式元件。

  • 空集合的情況要處理：分割器發了 Total=0，聚合器要能立刻完成。

🔗 相關工具

Graph View

Backlinks