事件驅動消費者

🎯 什麼情境該想到我

當你「希望訊息一抵達就立刻被處理，不想自己寫輪詢迴圈」的時候。

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

意圖：讓應用註冊一個回呼，訊息系統在訊息到達時主動呼叫它，消費者不需主動輪詢。

做法 / 結構要點：

向訊息系統註冊消費者 / 監聽器（callback）。
訊息到達時由系統觸發回呼，執行處理邏輯。
通常由訊息系統的執行緒管理併發。

★★ 推模式的核心風險：沒有背壓，你會被灌爆

★ broker 主動推 → 消費者無法說「等一下」
→ ① 記憶體暴增（未處理的訊息堆在 client 的緩衝區）
   ② ★★ 處理不完 ⇒ 大量訊息超過 ack 期限 ⇒ 重送 ⇒ 更多訊息 ⇒ 惡性循環

★★ 對策是「in-flight 上限」，而它在每個 broker 上都是一個容易忘記設的參數：

broker	限流參數	★ 預設
RabbitMQ	`basic.qos(prefetch_count)`	★★ 預設不限制——一個消費者會把整個佇列抓進記憶體
NATS push consumer	`max_ack_pending`	有預設但通常要調
AWS Lambda（SQS 觸發）	`batch size` + 保留併發數	★ 併發會自動擴，可能打爆下游 DB

★★ RabbitMQ 的 prefetch 預設不限制，是這個模式最常見的生產事故：
三個消費者，第一個連上就把 10 萬則訊息全抓走 →
另兩個閒置，而第一個 OOM。
★ 實務起點：prefetch = 併發 worker 數（或略多一點）。

★ Go 的推模式骨架（RabbitMQ）

if err := ch.Qos(prefetch, 0, false); err != nil { … }   // ★★ 第一件事就是設 QoS
 
deliveries, err := ch.Consume(queue, consumerTag, false /* autoAck=false */, …)
if err != nil { … }
 
sem := make(chan struct{}, prefetch)
for d := range deliveries {
    sem <- struct{}{}
    go func(d amqp.Delivery) {
        defer func() { <-sem }()
        mctx, cancel := context.WithTimeout(ctx, perMessageTimeout)
        defer cancel()
        if err := handle(mctx, d.Body); err != nil {
            _ = d.Nack(false, retriable(err))    // ★ 第二個參數是 requeue
            return
        }
        _ = d.Ack(false)                          // ★★ ack 在處理成功之後
    }(d)
}

★★ autoAck=false 是不可妥協的。
autoAck=true 表示「broker 送出就算成功」——處理失敗就丟訊息，而且完全靜默。

★ 這是推模式比拉模式更危險的地方：autoAck 這個參數讓「丟訊息」只差一個 bool。

★★ 優雅關閉在推模式下更難

★ 收到 SIGTERM 之後要做的事（順序很重要）：
① ★ 先 Cancel 消費者（basic.cancel）—— 告訴 broker 不要再送了
② ★ 等手上的 goroutine 全部處理完並 ack（有超時上限）
③ ★ 才關閉 channel 與連線

⚠️ 直接關連線 → 所有未 ack 的訊息立刻重送給別人（RabbitMQ 的行為），
而你手上可能正在處理它們 ⇒ 重複處理。
★ 這正是「消費端必須冪等」在推模式下更迫切的原因。

★ 推 vs 拉：什麼時候選推

選推	選輪詢式消費者
★ 延遲敏感（訊息一到就要處理）	吞吐優先
★ 流量平穩、可預測	★ 流量會突增（拉有天然背壓）
broker 只支援推（RabbitMQ）	broker 是拉（Kafka、SQS）
★ 訊息量小、處理快	★ 處理慢、需要精細控制節奏

★ 判準：處理速度會不會落後於到達速度？
可能會 → 拉更安全。推模式在那個情況下會變成「用重送把自己打死」。

★ 兄弟辨析

	差別
輪詢式消費者	★ 拉模式：天生有背壓
競爭消費者	多實例；推模式下 prefetch 決定負載怎麼分配
訊息分派器	單一消費者內部分派給 worker
服務啟動器	讓訊息與同步呼叫共用同一個服務入口

🧪 我實際套用的紀錄

（待填）

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

處理速度受制於系統推送，較難自行限流（需搭配其他機制）。
回呼中若處理過慢會累積壓力。
相對輪詢式消費者，延遲低但對節奏的掌控較少。
★★ RabbitMQ 的 prefetch 預設不限制——一個消費者會抓光整個佇列導致 OOM，其他消費者閒置。實務起點是「prefetch = 併發 worker 數」。
★★ autoAck=true 表示「送出就算成功」，處理失敗就靜默丟訊息。永遠用 autoAck=false。
★★ 推模式沒有背壓。處理不完 → 超過 ack 期限 → 重送 → 更多訊息 → 惡性循環。
★ 優雅關閉順序：先 cancel 消費者 → 等手上處理完 → 才關連線。 直接關連線會讓未 ack 訊息立刻重送給別人，而你正在處理它們。
★ 判準：處理速度會落後於到達速度嗎？ 可能會 → 用拉模式更安全。
★ Lambda 的 SQS 觸發會自動擴併發，可能打爆下游資料庫——要設保留併發數。
每則訊息要有獨立超時，否則一則卡住會佔住一個 prefetch 額度。
推模式下冪等更迫切，因為關閉與網路抖動都會造成重送。

🔗 相關工具

輪詢式消費者 Polling Consumer（同家族，改為主動拉取）
競爭消費者 Competing Consumers（同家族，多監聽器並行分擔）
訊息分派器 Message Dispatcher（同家族，收下後再分派給多個處理者）
企業整合模式
輪詢式消費者 —— ★ 拉模式，天生有背壓
競爭消費者 —— 多實例與負載分配
訊息分派器 —— 內部 worker pool
冪等接收者 —— 推模式下更迫切
交易式用戶端 —— ack 時機
工具-MQ消費端防禦三原則 —— 背壓的完整做法

Quartz 5

Explorer