Kafka 重复消费：自动提交位点与手动提交的 3 个错位

我不会起名字322· 后端 / 算法 / 数据库 技术栈 力扣 Hot100 Go 项目 Redis MySQL文章目录Kafka 重复消费自动提交位点与手动提交的 3 个错位一、先把根因说死二、错位 1自动提交的时间窗口三、错位 2先提交后处理或反过来四、错位 3退出时序错了五、rebalance 时的重复消费六、带走物Kafka 重复消费自动提交位点与手动提交的 3 个错位一、先把根因说死Kafka 里重复消费只有一个本质原因消息已经被处理了但 offset 没有或者没能提交。Broker 侧什么都不知道它只认 offset。只要 offset 还停在上一个位置下一次消费者启动或者分区被重新分配时那段消息就一定会被再读一遍。所以排查重复消费不是去查 Kafka而是去查**处理完成和提交 offset这两个动作之间到底哪里断了**。二、错位 1自动提交的时间窗口自动提交是默认配置enable.auto.committrue客户端每隔auto.commit.interval.ms默认 5 秒把上一次 poll 回来的最大 offset提交掉。问题在于提交是定时发生的和你的业务处理进度无关。时间线大概是这样t0.0s poll 回来 offset 100~200100 条 t4.9s 业务才处理到第 60 条 t5.0s 自动提交触发 → 提交的却是 offset 200poll 回来的最大值 t5.1s 进程崩溃重启之后从 offset 200 继续第 61~200 条消息永久丢失——这是自动提交更可怕的一面它不仅会重复消费还会丢消息。反过来如果提交发生在处理之前、崩溃发生在提交之后但处理之前就会出现重复消费。提交时机 vs 处理进度结果提交 实际处理进度然后崩了丢消息提交 实际处理进度然后崩了重复消费两者一致正常at least once结论只要你在意不丢消息就关掉自动提交。这不是风格问题是正确性问题。// sarama 里关掉自动提交config:sarama.NewConfig()config.Consumer.Offsets.AutoCommit.Enablefalseconfig.Consumer.Offsets.AutoCommit.Interval0三、错位 2先提交后处理或反过来关掉自动提交之后第二个坑就来了——手动提交的位置写错了。常见的错误写法长这样伪代码formsg:rangeclaim.Messages(){gofunc(m*sarama.ConsumerMessage){handle(m)// 异步处理}(msg)session.MarkMessage(msg,)// 立刻标记还没处理完就标记了}MarkMessage只是标记真正提交要等Commit()。但这里的根本问题是标记发生在处理完成之前。异步任务失败了offset 也已经提交了消息就丢了。正确的顺序必须保证处理成功是提交的前置条件formsg:rangeclaim.Messages(){iferr:handle(msg);err!nil{// 处理失败不要标记让这一批重来log.Printf(handle failed, offset%d: %v,msg.Offset,err)continue}session.MarkMessage(msg,)// 只有成功了才标记}还有一个更隐蔽的变体批量处理时提交了整个批次的最后一个 offset但批次中间有失败的。msgs:batch()// 一次拉 100 条for_,m:rangemsgs{handle(m)// 第 37 条失败但没中断}session.MarkMessage(msgs[len(msgs)-1],)// 提交了第 100 条的 offset第 37 条失败了offset 却提交到了 100。这是最典型的看起来跑通了、实际在静默丢消息。正确做法记录批次内失败的最小 offset只提交到那个位置之前。四、错位 3退出时序错了第三个错位发生在关闭流程里。看这段真实存在过的代码// 错误示例consumer.Unsubscribe()// 先退订consumer.Close()// 再关闭Unsubscribe()之后还在缓冲里、还没来得及提交的 offset 就没有机会提交了——因为客户端已经不再参与这个组coordinator 也不再接受它的提交。正确顺序是先提交、再关闭// 正确先确保 offset 落盘再断开handler.Flush()// 或者手动 Commit 一次iferr:consumer.Close();err!nil{log.Printf(close consumer: %v,err)}在 K8s 环境里这个坑还会被放大Pod 收到 SIGTERM 后只有terminationGracePeriodSeconds默认 30 秒的时间。如果你的关闭流程里没有把提交 offset放在最前面那么每一次发布都会制造一批重复消费。一个能用的优雅退出骨架funcmain(){ctx,cancel:context.WithCancel(context.Background())wg:sync.WaitGroup{}wg.Add(1)gofunc(){deferwg.Done()consumeLoop(ctx)// 消费主循环}()sig:make(chanos.Signal,1)signal.Notify(sig,syscall.SIGTERM,syscall.SIGINT)-sigcancel()// 1. 先停消费循环wg.Wait()// 2. 等在手的消息处理完commitPendingOffsets()// 3. 再提交 offsetconsumer.Close()// 4. 最后关闭}顺序不能反。先Close()再提交提交会失败先Unsubscribe()再提交提交会被拒。五、rebalance 时的重复消费前面三类都是提交时机的问题还有一类是分区易主带来的。消费者 A 处理完 offset 500还没提交就被判定超时踢出了组。分区重新分配给消费者 BB 从上一次已提交的offset 450 开始读——450 到 500 这 50 条被消费了两次。这一类完全无法靠提交时机避免因为崩溃/超时是外部事件。它只能靠消费端幂等兜底// 用业务主键做幂等而不是靠 Kafka 的 offsetfunchandle(m*sarama.ConsumerMessage)error{varevt Event json.Unmarshal(m.Value,evt)// 唯一键去重处理过了就跳过ok,err:dedup.TryMark(ctx,evt.BizID)// 比如 Redis SETNX / 唯一索引iferr!nil{returnerr}if!ok{returnnil// 已处理过直接返回成功}returndoBusiness(evt)}记住一条Kafka 的 offset 机制保证的是至少一次不是恰好一次。要恰好一次的效果必须在业务侧做幂等或者用 Kafka 事务 外部存储的原子提交成本高多数业务不值得。六、带走物排查重复消费的按顺序三问是不是自动提交是 → 关掉改手动。enable.auto.commitfalse。提交点在处理之后吗异步/批量的场景尤其要检查是否提交了失败消息的 offset。退出时序对吗取消消费 → 等处理完 → 提交 offset → Close()四步顺序不能乱。最后一句结论可以直接写进设计文档Kafka 只提供 at-least-once重复消费是常态而不是异常幂等是消费端的必选项不是可选项。