Quartz 5

選擇性消費者

Properties1
tagsintegration, messaging, eip, messaging-endpoints

8 min read

🎯 什麼情境該想到我

當你「只想接收通道中符合特定條件的訊息，其餘留在通道給別人處理」的時候。

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

意圖：讓消費者指定一組選擇條件（selector），只接收符合條件的訊息，不符合的留在通道。

做法 / 結構要點：

  • 消費者宣告選擇條件（通常對訊息標頭 / 屬性做判斷）。
  • 訊息系統只把符合條件的訊息投遞給它。
  • 不符合的訊息留給其他消費者或稍後處理。

★★ 先問：這件事該由 broker 做還是消費者做

① ★★ broker 端過濾（最好）
   RabbitMQ topic／headers exchange、SNS filter policy、NATS 通配
   → ★ 不符條件的訊息根本不會送給你：零頻寬、零 CPU

② ★ 分流成不同通道（Kafka 上唯一的辦法）
   → ★ 這就是 [[資料型別通道]]

③ 消費端過濾（本模式）
   → ★★ 白收所有訊息：頻寬、反序列化 CPU、以及「ack 掉不要的訊息」的成本

★★ Kafka 幾乎沒有 broker 端過濾（只有分區），所以在 Kafka 上
「選擇性消費」實際上只有 ② 和 ③。

③ 的成本可以估算：

2 KB × 10,000 則/秒 × 只有 5% 有用
→ ★ 白收 19 MB/s 網路 + 9,500 則/秒的反序列化
→ ★ 值不值得分流成新的 topic？用這個數字決定，不要憑感覺

★★ 這個模式在點對點通道上的致命問題

★ 一個佇列，兩個選擇性消費者：
  A 只收 region=TW，B 只收 region=JP
  → ★★ 來了一則 region=US 的訊息
     → A 拒絕（nack requeue）→ B 拒絕 → A 又拿到 → ★ 無限迴圈
     → ★ 而且它會一直卡在隊首，把後面全部堵住（poison message）

★★ 這是選擇性消費者最容易踩的坑：
在競爭型（點對點）通道上做選擇性消費，沒被任何人接受的訊息會無限循環。
★ 對策只有兩個：
① 每個消費者的條件必須完全覆蓋所有可能的訊息（★ 而這很難保證）；
② ★ 不要在同一個佇列上放多個選擇性消費者——改用 ① 或 ②（broker 過濾／分流）。

★ Go 的實作：不要的訊息要 ack，不要 nack

func handle(ctx context.Context, msg Message) error {
    var e Envelope
    if err := json.Unmarshal(msg.Body, &e); err != nil { … }
 
    if !interested(e) {
        metrics.Skipped.WithLabelValues(e.Type).Inc()   // ★ 一定要計數
        return msg.Ack()          // ★★ ack 掉，不要 nack —— nack 會造成無限迴圈
    }
    …
}

★★ 「不感興趣就 ack」在發布訂閱通道上是正確的（你有自己的 queue／offset，
ack 只影響你）。
⚠️ 但在共用佇列上，ack 表示「我處理掉了」——別人就看不到這則訊息了。
→ ★ 所以選擇性消費者幾乎只適用於「每個消費者有自己的通道」的情境。

★★ 過濾條件在消費端 vs 在 broker：一個常被忽略的差異

★ 條件在消費端 → 條件改變只需要部署消費者（★ 自己可控）
★ 條件在 broker → 條件改變要改基礎設施（★ 可能要跨團隊、可能要 IaC PR）
  ⚠️ 但也意味著「這個消費者收什麼」是可觀測的（在 broker 的 binding 上看得到）

★ 判準：條件多久改一次、誰擁有它？
自己擁有且常改 → 消費端。
穩定且需要被別人看到（治理、合規）→ broker binding。

★ 兄弟辨析

差別
訊息過濾器★ 獨立元件在中間丟掉訊息；選擇性消費者在消費端自己挑
資料型別通道★ 用通道分開，根本不需要選擇
競爭消費者★ 各消費者都收同一批；★ 混用這兩者會造成無限迴圈
內容路由器在中間依內容分流

🧪 我實際套用的紀錄

  • （待填）

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

  • 選擇通常只能對標頭 / 屬性做，不宜對大 body 內容做（效能）。

  • 條件互斥要設計好，否則某些訊息可能沒人收（卡在通道）。

  • 是在消費端過濾；也可改在來源端用不同通道分流。

  • ★★ 不要在同一個競爭型佇列上放多個選擇性消費者。沒被任何人接受的訊息會無限循環並堵住隊首。

  • ★★ 不感興趣的訊息要 ack，不要 nack。 nack 會造成無限迴圈。

  • ★ 但在共用佇列上 ack 表示「我處理掉了」，別人就看不到了——所以這個模式幾乎只適用於「每個消費者有自己的通道」。

  • ★★ Kafka 幾乎沒有 broker 端過濾。在 Kafka 上只能分流成不同 topic，或消費端全收再丟。

  • ★ 消費端過濾的成本可以估算（訊息大小 × 速率 × 丟棄比例）。用數字決定要不要分流。

  • ★ 跳過的訊息一定要計數，否則「這筆為什麼沒處理」查不出來。

  • ★ 條件放消費端可自己控制、放 broker 則可被觀測。依「誰擁有它、多久改一次」決定。

  • 優先考慮 資料型別通道：用通道分開，根本不需要選擇。

🔗 相關工具

Graph View

Backlinks