必须手动提交 offset 并禁用自动提交，设置 max.poll.interval.ms≥300000、http timeout、异步调用 deepseek、校验消息结构、用业务 id+redis 实现幂等，三者对齐才能避免丢消息与重复消费。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 免费无限量使用 DeepSeek R1 模型☜☜☜
手动提交 offset 是 Kafka 消费者不丢消息的底线
自动提交 enable.auto.commit=true 看似省事，但只要处理逻辑稍慢（比如调用一次 DeepSeek API 耗时超过 AUTO_COMMIT_INTERVAL_MS_CONFIG），Kafka 就会提前把 offset 提交掉——结果是消息还没真正处理完，消费者就挂了，重启后直接跳过这条，永久丢失。
- 必须设为
enable.auto.commit=false，自己控制提交时机 - 提交只能在
process(msg)成功完成后调用consumer.commit()，不能放在 try/except 外面 - 如果处理失败要重试，得先记录失败状态（比如写 Redis 或 DB），再
commit()；否则下次 poll 还会拿到同一条 - 注意
commitSync()会阻塞，超时默认 60 秒；若怕卡死，可用commitAsync()+ 回调校验，但回调里不能做耗时操作
DeepSeek 调用不能堵住 Kafka 消费循环
DeepSeek 接口响应时间不稳定（尤其批量推理或带 RAG 检索时），而 Kafka 消费者线程一旦卡在 requests.post(...) 上，就会触发 max.poll.interval.ms 超时，引发分区再均衡（rebalance）——这会导致重复消费，甚至整个消费组短暂不可用。
- 务必设置
max.poll.interval.ms=300000（5 分钟）以上，给 DeepSeek 留出缓冲时间 - HTTP 请求必须加
timeout=(3, 30)：3 秒 connect，30 秒 read，避免无限 hang - 不要在消费线程里直接跑 DeepSeek 推理；建议用线程池（
concurrent.futures.ThreadPoolExecutor）异步提交，主线程只负责 poll + 分发 - DeepSeek 返回异常（如
429 Too Many Requests或503 Service Unavailable）时，别直接抛错退出，要 sleep 后重试，否则可能触发频繁 rebalance
消息体结构要适配 DeepSeek 的输入格式
Kafka 里存的原始消息（比如 JSON 字符串）不能直接喂给 DeepSeek；字段缺失、嵌套过深、长度超限都会导致模型静默失败或返回乱码，而日志里只看到 status_code=200，误以为成功。
- 消费后先做轻量解析：
json.loads(msg.value.decode("utf-8"))，检查必填字段如"query"、"user_id" - DeepSeek-R1 默认上下文 128K，但实际建议单次输入控制在 8k token 内；超长内容需切片（如按段落或按字段），并加明确分隔符
"---" - 敏感字段（如用户手机号、订单号）建议脱敏后再进模型，避免训练数据污染或合规风险
- 输出要强制指定
response_format={"type": "json_object"}（如果 DeepSeek 支持），否则自由生成文本难解析
重复消费问题不能只靠 Kafka 参数解决
即使关了自动提交、调大超时，网络抖动、服务重启、DeepSeek 侧重试仍可能导致同一条 Kafka 消息被处理两次。Kafka 本身不保证“恰好一次”，得靠业务层兜底。
- 每条消息必须带唯一业务 ID（如
msg.headers.get(b"x-request-id")或从 value 解析出"event_id"） - 处理前先查 Redis：
redis.setex(f"proc:{event_id}", 3600, "1")，如果已存在就跳过 - 别用数据库主键冲突做幂等——插入失败再查，延迟高且并发下仍有概率漏判
- 注意 Redis key 过期时间要比最大处理耗时长至少 2 倍，防止处理中 key 过期，重复入队
真正的难点不在写几行 consumer 代码，而在于把 DeepSeek 的不确定性、Kafka 的语义边界、业务的幂等要求三者对齐——少一个环节，线上就容易出现“查不到记录但日志显示已处理”这种哑巴错误。
