构建Kafka消费者

构建Kafka消费者 是做什么的？

使用时机

• 编写或调试Kafka消费者/生产者时。
• 选择偏移量提交策略和传递保证时。
• 调优消费者组并行性、再平衡或死信处理时。
• 不要用于流处理/窗口化（应使用processing-streaming-data）。

工作流

- [ ] 根据目标并行性调整分区数量（消费者数量 ≤ 分区数量）
- [ ] 成功处理**后**提交偏移量
- [ ] 使接收器幂等（按事件键执行更新插入）
- [ ] 处理再平衡（撤销时提交，避免长轮询间隙）
- [ ] 将有毒消息路由到死信主题

1. 分区限制并行性。 消费者组最多只能扩展到分区数量；多余的消费者会闲置。为达到峰值吞吐量选择合适的分区数。
2. 处理后再提交。 仅当工作持久完成后才提交偏移量，而不是之前——提前提交会在崩溃时丢失消息。
3. 幂等接收器。 至少一次传递意味着重试时会出现重复；通过稳定的事件键执行更新插入，使重复处理无害。
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并触发无休止的再平衡；应使用有界批次处理或后台工作线程。
• 缺少死信路径——一条有毒消息阻塞整个分区。
• 依赖精确一次语义但无幂等接收器——任何至少一次跳都会重新引入重复。