建立 Kafka 消費者

建立 Kafka 消費者 是做什麼的？

使用時機

• 撰寫或除錯 Kafka 消費者/生產者。
• 選擇位移提交策略與傳遞保證。
• 調整消費者群組平行度、再平衡或死信處理。
• 不適用於串流處理/視窗化（請使用 processing-streaming-data）。

工作流程

- [ ] 將分區大小調整至目標平行度（消費者數量 <= 分區數量）
- [ ] 在成功處理**後**提交位移
- [ ] 讓接收端成為冪等（依事件鍵進行 upsert）
- [ ] 處理再平衡（在回收時提交，避免長時間輪詢間隔）
- [ ] 將毒藥訊息路由至死信主題

1. 分區限制了平行度。 消費者群組只能擴展到分區數量；多餘的消費者會閒置。請根據尖峰吞吐量選擇分區數。
2. 處理後再提交。 只有在工作持久完成後才提交位移，不要提前提交——提前提交會在崩潰時遺失訊息。
3. 冪等接收端。 至少一次意味著重試時會產生重複；透過穩定的事件鍵進行 upsert，讓重新處理不會造成影響。
4. 再平衡一定會發生。 在分區回收時提交位移，並讓 poll() 間隔保持在 max.poll.interval.ms 以下，以免代理程式將消費者踢出。
5. 毒藥訊息應送往死信主題並附上錯誤資訊，這樣一個壞紀錄就不會阻塞整個分區。

模式

處理後手動提交：

consumer = KafkaConsumer("orders", group_id="etl",
                         enable_auto_commit=False,
                         max_poll_records=500)
for msg in consumer:
    try:
        upsert(process(msg))          # 依鍵值進行冪等處理
        consumer.commit()             # 成功後才提交
    except PoisonError:
        send_to_dlq(msg)
        consumer.commit()             # 跳過壞紀錄

冪等 / 交易生產者 — 設定 enable.idempotence=true（去重複重試），並使用交易實現在主題間進行讀取-處理-寫入的恰好一次語意。

吞吐量調整 — 增加 max.poll.records、fetch.min.bytes，並批次處理；保持處理速度快，以避免再平衡踢除。

常見陷阱

• 自動提交 + 慢速處理 — 位移在工作完成前就前進；崩潰會丟失訊息。建議在處理後手動提交。
• 消費者數量多於分區 — 多餘的消費者閒置；請重新分區以擴展。
• 輪詢之間處理時間過長 — 超過 max.poll.interval.ms 會觸發無限再平衡；請在有限的批次中處理或使用背景工作者。
• 沒有死信路徑 — 一個毒藥訊息會阻塞整個分區。
• 依賴恰好一次卻沒有冪等接收端 — 任何至少一次的跳躍都會重新引入重複。