Go 写 Kafka 消费者：sarama 的 3 个坑与优雅退出写法

我不会起名字322· 后端 / 算法 / 数据库 技术栈 力扣 Hot100 Go 项目 Redis MySQL文章目录Go 写 Kafka 消费者sarama 的 3 个坑与优雅退出写法一、先说结论用 ConsumerGroup不要用 Consumer二、坑 1没处理错误通道三、坑 2在 ConsumeClaim 里开 goroutine四、坑 3offset 提交时机五、优雅退出完整写法六、还有三个细节1. 必须指定 Kafka 版本2. Offsets.Initial 的选择3. 设置合理的超时七、带走物Go 写 Kafka 消费者sarama 的 3 个坑与优雅退出写法一、先说结论用 ConsumerGroup不要用 Consumersarama 有两套消费者 APIAPI语义适用Consumer手动指定 topic/partition/offset特殊场景要精确控制位点ConsumerGroup自动加入组、自动分配分区、自动 rebalance99% 的场景用Consumer手写我要消费分区 0、1、2是最常见的错误——它不支持 rebalance消费者挂了分区就没人消费扩容也不会自动接管。// ✅ 正确用 ConsumerGroupconfig:sarama.NewConfig()config.Consumer.Group.Rebalance.Strategysarama.BalanceStrategyRoundRobin config.Consumer.Offsets.Initialsarama.OffsetNewest client,err:sarama.NewConsumerGroup(brokers,order-consumer-group,config)iferr!nil{returnerr}handler:consumerHandler{}// 这个 Consume 是阻塞的要在循环里跑rebalance 后会返回for{iferr:client.Consume(ctx,[]string{order-topic},handler);err!nil{log.Printf(consume error: %v,err)}ifctx.Err()!nil{return}}注意Consume的循环调用它不是调用一次就一直消费而是每次 rebalance 结束就会返回所以必须放在for循环里重新调用。忘了这层循环rebalance 之后消费者就消失了。二、坑 1没处理错误通道sarama 的异步生产者默认关闭返回通道必须显式打开否则失败全静默config:sarama.NewConfig()config.Producer.Return.Successestrueconfig.Producer.Return.Errorstrue// 不开这个发送失败永远不知道producer,_:sarama.NewAsyncProducer(brokers,config)// 必须有 goroutine 消费这两个 channel否则会阻塞gofunc(){forerr:rangeproducer.Errors(){log.Printf(发送失败: %v,err)// 这里要决定重试 / 落库补偿 / 告警}}()消费者这边也有一个容易漏的ConsumerGroup的Errors()gofunc(){forerr:rangeclient.Errors(){log.Printf(consumer group error: %v,err)}}()三、坑 2在ConsumeClaim里开 goroutineConsumeClaim的语义是每个分区一个 goroutine。所以默认写法天然就是分区内串行 分区间并行。在里面再开 goroutine 会破坏顺序性而且会让 offset 提交变得不可控// ❌ 错误破坏分区内顺序且 MarkMessage 的语义被破坏func(h*handler)ConsumeClaim(sess sarama.ConsumerGroupSession,claim sarama.ConsumerGroupClaim)error{formsg:rangeclaim.Messages(){gofunc(m*sarama.ConsumerMessage){handle(m)sess.MarkMessage(m,)// 并发调用顺序乱}(msg)}returnnil}// ✅ 正确分区内串行func(h*handler)ConsumeClaim(sess sarama.ConsumerGroupSession,claim sarama.ConsumerGroupClaim)error{formsg:rangeclaim.Messages(){iferr:handle(ctx,msg);err!nil{log.Printf(处理失败 offset%d: %v,msg.Offset,err)continue// 或者按策略决定要不要中断}sess.MarkMessage(msg,)// 只有成功才标记}returnnil}如果单分区真的处理不过来正确做法是加分区数而不是在分区内并发。四、坑 3offset 提交时机sarama 的ConsumerGroup默认自动提交Consumer.Offsets.AutoCommit.Enable默认true间隔 1 秒。这个默认行为的风险在重复消费那篇讲过提交的是已处理的最大 offset但业务可能只处理了一半。进程崩溃时既可能重复消费也可能丢消息。改成手动config:sarama.NewConfig()config.Consumer.Offsets.AutoCommit.Enablefalse// 关掉自动提交// 在处理循环里自己控制提交时机func(h*handler)ConsumeClaim(sess sarama.ConsumerGroupSession,claim sarama.ConsumerGroupClaim)error{formsg:rangeclaim.Messages(){iferr:handle(ctx,msg);err!nil{returnerr// 返回错误会触发重新消费}sess.MarkMessage(msg,)}returnnil}关掉自动提交后MarkMessage只是标记真正提交需要sess.Commit()。不过 sarama 在ConsumeClaim返回前会自动提交一次标记过的 offset。如果要精确控制比如每 100 条提交一次count:0formsg:rangeclaim.Messages(){iferr:handle(ctx,msg);err!nil{returnerr}sess.MarkMessage(msg,)ifcount;count%1000{sess.Commit()}}sess.Commit()// 退出前再提交一次五、优雅退出完整写法这是最值得抄的一段。要点是顺序不能反funcmain(){ctx,cancel:context.WithCancel(context.Background())defercancel()config:sarama.NewConfig()config.Versionsarama.V2_8_0_0// 必须指定版本config.Consumer.Group.Rebalance.Strategysarama.BalanceStrategyRoundRobin config.Consumer.Offsets.Initialsarama.OffsetNewest config.Consumer.Offsets.AutoCommit.Enablefalseconfig.Consumer.Return.Errorstrueclient,err:sarama.NewConsumerGroup(brokers,groupID,config)iferr!nil{log.Fatal(err)}deferclient.Close()// 最后关gofunc(){forerr:rangeclient.Errors(){log.Printf(group error: %v,err)}}()handler:consumerHandler{ready:make(chanbool)}wg:sync.WaitGroup{}wg.Add(1)gofunc(){deferwg.Done()for{// rebalance 后 Consume 会返回所以要循环调用iferr:client.Consume(ctx,topics,handler);err!nil{log.Printf(consume: %v,err)}ifctx.Err()!nil{return}}}()-handler.ready log.Println(消费者已就绪)// 等退出信号sigterm:make(chanos.Signal,1)signal.Notify(sigterm,syscall.SIGINT,syscall.SIGTERM)-sigterm log.Println(收到退出信号开始优雅退出)cancel()// 1. 取消 ctxConsume 循环退出wg.Wait()// 2. 等在手的消息处理完ConsumeClaim 里的循环会自然结束log.Println(退出完成)}Handler 里也要配合 ctxtypeconsumerHandlerstruct{readychanbool}func(h*consumerHandler)Setup(sarama.ConsumerGroupSession)error{close(h.ready)returnnil}func(h*consumerHandler)Cleanup(sarama.ConsumerGroupSession)error{log.Println(分区被收回Cleanup)returnnil}func(h*consumerHandler)ConsumeClaim(sess sarama.ConsumerGroupSession,claim sarama.ConsumerGroupClaim)error{// 关键监听 claim.Messages() 的关闭以及 session 的 ctxfor{select{casemsg,ok:-claim.Messages():if!ok{returnnil// 通道关闭正常退出}iferr:handle(sess.Context(),msg);err!nil{log.Printf(handle offset%d: %v,msg.Offset,err)continue}sess.MarkMessage(msg,)case-sess.Context().Done():// session 被取消比如 rebalance 或者主流程 cancelreturnnil}}}select里一定要加sess.Context().Done()否则cancel()之后消费者不会退出wg.Wait()会永远卡住最后被 SIGKILL——那和没有优雅退出一样。六、还有三个细节1. 必须指定 Kafka 版本config.Versionsarama.V2_8_0_0不指定的话 sarama 会用一个很老的默认值某些新特性比如静态成员、增量 rebalance不生效而且可能报奇怪的协议错误。2.Offsets.Initial的选择config.Consumer.Offsets.Initialsarama.OffsetNewest// 从最新开始默认config.Consumer.Offsets.Initialsarama.OffsetOldest// 从最早开始注意这个只在没有已提交 offset时生效第一次启动或者 offset 过期。一旦有提交记录就从提交的位置继续。新业务用OffsetNewest避免启动时灌进一堆历史消息要补数据的场景用OffsetOldest。3. 设置合理的超时config.Consumer.Group.Session.Timeout10*time.Second config.Consumer.Group.Heartbeat.Interval3*time.Second config.Consumer.MaxProcessingTime5*time.Minute// 对应 max.poll.interval.msMaxProcessingTime就是max.poll.interval.ms在 sarama 里的对应项默认 5 分钟。处理慢的话要么减小批量要么调大它详见 rebalance 那篇。七、带走物一份可直接用的配置骨架config:sarama.NewConfig()config.Versionsarama.V2_8_0_0// 必须指定版本config.Consumer.Group.Rebalance.Strategysarama.BalanceStrategyRoundRobin config.Consumer.Offsets.Initialsarama.OffsetNewest config.Consumer.Offsets.AutoCommit.Enablefalse// 手动提交config.Consumer.Return.Errorstrue// 打开错误通道config.Consumer.Group.Session.Timeout10*time.Second config.Consumer.Group.Heartbeat.Interval3*time.Second config.Consumer.MaxProcessingTime5*time.Minute三个坑的速查坑症状修法没开错误通道发送/消费失败静默Return.Errors true goroutine 消费它Consume没循环调用rebalance 后消费者消失放在for循环里分区内开 goroutine顺序乱、offset 不可控ConsumeClaim里串行处理优雅退出顺序1. cancel() 停止 Consume 循环 2. wg.Wait() 等在手消息处理完 3. ConsumeClaim 里监听 sess.Context().Done()及时返回 4. defer client.Close() 最后关闭客户端一句话结论sarama 的坑大多集中在异步通道没消费“Consume 没循环调用”优雅退出没等到位这三处把这三处写对剩下的就是 offset 提交策略的选择问题。