輪詢式消費者

🎯 什麼情境該想到我

當你「想由應用主動、按自己的節奏去通道拉取訊息來處理，而不是被動等訊息推來」的時候。

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

意圖：應用明確地去通道詢問是否有訊息可處理，有就取來處理，沒有就等待或稍後再問；由消費者控制取用速度。

做法 / 結構要點：

消費者主動呼叫接收（可阻塞或帶逾時）。
取得訊息後處理，處理完再拉下一則。
可依自身負載調整輪詢頻率與並行度。

★★ 最大的優點：它天生有背壓

// ★ 拉模式：你控制節奏。手上沒處理完就不去拉下一批
for {
    msgs, err := consumer.Fetch(ctx, maxBatch)   // ★ 你決定一次拿幾則
    if err != nil { … }
    for _, m := range msgs { handle(ctx, m) }    // ★ 處理完才回到 Fetch
}

★★ 這是拉模式相對推模式最重要的優勢：
消費者永遠不會被灌爆——它只在自己準備好時才要下一批。
推模式必須自己實作等效的機制（RabbitMQ 的 prefetch），而那個設定很容易忘記。

★ 今天的主流其實都是拉模式

broker	模式
Kafka	★ 拉（`poll()`／`FetchMessage`）
SQS	★ 拉（`ReceiveMessage`，配長輪詢）
RabbitMQ	★ 推（`basic.consume`）為主；`basic.get` 是拉但效率很差，不要用在迴圈裡
NATS JetStream	兩種都有（pull consumer／push consumer）

★ EIP 書把輪詢式消費者寫成「比較笨的那個選項」，因為 2003 年的 JMS 以推為主。
今天反過來——Kafka 和 SQS 都是拉，而且拉是主流。

★★ 空輪詢的成本：長輪詢是必須的

// ❌ 短輪詢 + sleep —— ★ 延遲與成本都差
for {
    msgs, _ := sqs.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
        MaxNumberOfMessages: 10,
        // ★ 沒設 WaitTimeSeconds → 立刻回空 → 你只能 sleep
    })
    if len(msgs) == 0 { time.Sleep(time.Second); continue }   // ★ 平均多 500ms 延遲
}
 
// ✅ 長輪詢 —— ★ 有訊息立刻回，沒訊息就掛在那裡等
ReceiveMessageInput{ MaxNumberOfMessages: 10, WaitTimeSeconds: 20 }

★★ SQS 的短輪詢有兩個具體代價：
① 延遲：你的 sleep 間隔就是額外延遲。
② ★ API 呼叫費用：SQS 按請求計費，空輪詢也算——一個 sleep 100ms 的迴圈
一天就是 86 萬次請求。
→ ★ 一律用長輪詢（SQS 最長 20 秒；Kafka 的 FetchMaxWait）。

★★ 批次的兩難

★ 批次大 → 吞吐高、API 呼叫少
★ 批次大 → ① 一則壞訊息可能影響整批的處理
            ② ★★ 處理整批的時間可能超過 visibility timeout ⇒ 後面幾則被重送
            ③ 延遲變高（等湊滿一批）

// ★ 務實做法：批次抓取，但逐則 ack，而且每則獨立超時
for _, m := range msgs {
    mctx, cancel := context.WithTimeout(ctx, perMessageTimeout)
    err := handle(mctx, m)
    cancel()
    if err == nil { m.Ack() } else { m.Nack(retriable(err)) }   // ★ 逐則決定
}

★ 批次大小的上限應該由「整批處理時間 < visibility timeout 的一半」反推，
而不是憑感覺設 100。

★ 並行：拉模式下怎麼加速

// ★ 方案一：多個消費者實例（[[競爭消費者]]）—— ★ Kafka 受分區數限制
// ★ 方案二：一個消費者拉、內部 worker pool 處理（[[訊息分派器]]）
sem := make(chan struct{}, concurrency)
for _, m := range msgs {
    sem <- struct{}{}
    go func(m Message) {
        defer func() { <-sem }()
        if err := handle(ctx, m); err == nil { m.Ack() }
    }(m)
}
// ★★ 代價：順序沒了；而且「什麼時候算處理完這一批」變複雜
//    → ★ 而且 Kafka 的 offset 提交會變得棘手（見 [[訊息分派器]]）

★ 兄弟辨析

	差別
事件驅動消費者	★ 推模式：broker 主動送；★ 需要自己限流
競爭消費者	多個消費者實例搶同一通道
訊息分派器	一個消費者拉、內部分給多個 worker
選擇性消費者	只拉符合條件的

🧪 我實際套用的紀錄

（待填）

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

空輪詢會浪費資源；逾時 / 退避策略要設好。
相對事件驅動消費者，延遲較高但節奏可控。
適合需要限流、批次或明確控制處理時機的場景。
★★ 拉模式天生有背壓，這是它最重要的優勢。推模式必須自己設 prefetch，而那很容易忘。
★★ 一律用長輪詢。SQS 短輪詢有延遲與 API 費用雙重代價（sleep 100ms 的迴圈一天 86 萬次請求）。
★★ 批次大小上限應由「整批處理時間 < visibility timeout 的一半」反推，不是憑感覺設 100。
★ 批次抓取但逐則 ack、每則獨立超時。 整批一起 ack 會讓一則壞訊息影響整批。
★ 內部 worker pool 並行會讓順序消失，而且 Kafka 的 offset 提交變棘手。
★ Kafka 的並行度上限是分區數，加實例超過分區數只是閒置。
RabbitMQ 的 basic.get 效率很差，不要放在迴圈裡當拉模式用。
優雅關閉要等手上那批處理完，否則造成重送或遺失。

🔗 相關工具

事件驅動消費者 Event-Driven Consumer（同家族，改為被推送觸發）
競爭消費者 Competing Consumers（同家族，多個消費者並行拉取）
企業整合模式
事件驅動消費者 —— 推模式的對照
競爭消費者 —— 多實例擴展
訊息分派器 —— 單消費者內部並行
選擇性消費者 —— 只拉部分訊息
交易式用戶端 —— ack 時機
訊息端點 —— 完整的消費迴圈骨架

Quartz 5

Explorer