企业官网建设流程全解析

我不会起名字322· 后端 / 算法 / 数据库

📘 技术栈 ｜ 🧩 力扣 Hot100 ｜ 🐹 Go 项目 ｜ 🟥 Redis ｜ 🐬 MySQL

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

一、先把根因说死

Kafka 里"重复消费"只有一个本质原因：消息已经被处理了，但 offset 没有（或者没能）提交。

Broker 侧什么都不知道，它只认 offset。只要 offset 还停在上一个位置，下一次消费者启动或者分区被重新分配时，那段消息就一定会被再读一遍。

所以排查重复消费，不是去查 Kafka，而是去查**"处理完成"和"提交 offset"这两个动作之间，到底哪里断了**。

二、错位 1：自动提交的时间窗口

自动提交是默认配置（enable.auto.commit=true），客户端每隔auto.commit.interval.ms（默认 5 秒）把"上一次 poll 回来的最大 offset"提交掉。

问题在于：提交是定时发生的，和你的业务处理进度无关。

时间线大概是这样：

t=0.0s poll 回来 offset 100~200（100 条） t=4.9s 业务才处理到第 60 条 t=5.0s 自动提交触发 → 提交的却是 offset 200（poll 回来的最大值） t=5.1s 进程崩溃

重启之后从 offset 200 继续，第 61~200 条消息永久丢失——这是自动提交更可怕的一面：它不仅会重复消费，还会丢消息。

反过来，如果提交发生在处理之前、崩溃发生在提交之后但处理之前，就会出现重复消费。

结论：只要你在意不丢消息，就关掉自动提交。这不是风格问题，是正确性问题。

// sarama 里关掉自动提交config:=sarama.NewConfig()config.Consumer.Offsets.AutoCommit.Enable=falseconfig.Consumer.Offsets.AutoCommit.Interval=0

三、错位 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，还没提交，就被判定超时踢出了组。分区重新分配给消费者 B，B 从上一次已提交的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 事务 + 外部存储的原子提交（成本高，多数业务不值得）。

六、带走物

排查重复消费的按顺序三问：

1. 是不是自动提交？是 → 关掉，改手动。enable.auto.commit=false。
2. 提交点在处理之后吗？异步/批量的场景尤其要检查"是否提交了失败消息的 offset"。
3. 退出时序对吗？取消消费 → 等处理完 → 提交 offset → Close()，四步顺序不能乱。

最后一句结论可以直接写进设计文档：Kafka 只提供 at-least-once，重复消费是常态而不是异常；幂等是消费端的必选项，不是可选项。