企业官网建设流程全解析

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

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

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

一、先说结论：用 ConsumerGroup，不要用 Consumer

sarama 有两套消费者 API：

用Consumer手写"我要消费分区 0、1、2"是最常见的错误——它不支持 rebalance，消费者挂了分区就没人消费，扩容也不会自动接管。

// ✅ 正确：用 ConsumerGroupconfig:=sarama.NewConfig()config.Consumer.Group.Rebalance.Strategy=sarama.BalanceStrategyRoundRobin config.Consumer.Offsets.Initial=sarama.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.Successes=trueconfig.Producer.Return.Errors=true// 不开这个，发送失败永远不知道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里开 goroutine

ConsumeClaim的语义是：每个分区一个 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}

如果单分区真的处理不过来，正确做法是加分区数，而不是在分区内并发。

四、坑 3：offset 提交时机

sarama 的ConsumerGroup默认自动提交（Consumer.Offsets.AutoCommit.Enable默认true，间隔 1 秒）。

这个默认行为的风险在重复消费那篇讲过：提交的是"已处理的最大 offset"，但业务可能只处理了一半。进程崩溃时，既可能重复消费也可能丢消息。

改成手动：

config:=sarama.NewConfig()config.Consumer.Offsets.AutoCommit.Enable=false// 关掉自动提交// 在处理循环里自己控制提交时机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%100==0{sess.Commit()}}sess.Commit()// 退出前再提交一次

五、优雅退出：完整写法

这是最值得抄的一段。要点是顺序不能反：

funcmain(){ctx,cancel:=context.WithCancel(context.Background())defercancel()config:=sarama.NewConfig()config.Version=sarama.V2_8_0_0// 必须指定版本config.Consumer.Group.Rebalance.Strategy=sarama.BalanceStrategyRoundRobin config.Consumer.Offsets.Initial=sarama.OffsetNewest config.Consumer.Offsets.AutoCommit.Enable=falseconfig.Consumer.Return.Errors=trueclient,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. 取消 ctx，Consume 循环退出wg.Wait()// 2. 等在手的消息处理完（ConsumeClaim 里的循环会自然结束）log.Println("退出完成")}

Handler 里也要配合 ctx：

typeconsumerHandlerstruct{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 或者主流程 cancel）returnnil}}}

select里一定要加sess.Context().Done()：否则cancel()之后消费者不会退出，wg.Wait()会永远卡住，最后被 SIGKILL——那和没有优雅退出一样。

六、还有三个细节

1. 必须指定 Kafka 版本

config.Version=sarama.V2_8_0_0

不指定的话 sarama 会用一个很老的默认值，某些新特性（比如静态成员、增量 rebalance）不生效，而且可能报奇怪的协议错误。

2.Offsets.Initial的选择

config.Consumer.Offsets.Initial=sarama.OffsetNewest// 从最新开始（默认）config.Consumer.Offsets.Initial=sarama.OffsetOldest// 从最早开始

注意这个只在"没有已提交 offset"时生效（第一次启动或者 offset 过期）。一旦有提交记录，就从提交的位置继续。

• 新业务用OffsetNewest（避免启动时灌进一堆历史消息）；
• 要补数据的场景用OffsetOldest。

3. 设置合理的超时

config.Consumer.Group.Session.Timeout=10*time.Second config.Consumer.Group.Heartbeat.Interval=3*time.Second config.Consumer.MaxProcessingTime=5*time.Minute// 对应 max.poll.interval.ms

MaxProcessingTime就是max.poll.interval.ms在 sarama 里的对应项，默认 5 分钟。处理慢的话要么减小批量，要么调大它（详见 rebalance 那篇）。

七、带走物

一份可直接用的配置骨架：

config:=sarama.NewConfig()config.Version=sarama.V2_8_0_0// 必须指定版本config.Consumer.Group.Rebalance.Strategy=sarama.BalanceStrategyRoundRobin config.Consumer.Offsets.Initial=sarama.OffsetNewest config.Consumer.Offsets.AutoCommit.Enable=false// 手动提交config.Consumer.Return.Errors=true// 打开错误通道config.Consumer.Group.Session.Timeout=10*time.Second config.Consumer.Group.Heartbeat.Interval=3*time.Second config.Consumer.MaxProcessingTime=5*time.Minute

三个坑的速查：

优雅退出顺序：

1. cancel() 停止 Consume 循环 2. wg.Wait() 等在手消息处理完 3. ConsumeClaim 里监听 sess.Context().Done()，及时返回 4. defer client.Close() 最后关闭客户端

一句话结论：sarama 的坑大多集中在"异步通道没消费"“Consume 没循环调用”"优雅退出没等到位"这三处；把这三处写对，剩下的就是 offset 提交策略的选择问题。