直接用 sarama 容易踩坑因其原生 api 过于底层：连接不自动管理、重试需手写、offset 提交时机难控；配置散落导致行为不一致；需统一决策如启用 successes、校验 rebalance 重试、禁用 consumepartition。
为什么直接用 sarama 容易踩坑
因为 sarama 原生 API 太底层：连接管理不自动、重试逻辑要手写、消费者组 offset 提交时机难控制。很多团队封装失败，不是因为功能没做全，而是把 config := sarama.NewConfig() 这类配置散落在业务代码里，导致不同服务的超时、重试、序列化行为不一致。
真正要封装的不是“怎么发消息”，而是“怎么让发消息这件事在分布式环境下可观察、可降级、可灰度”。比如：sarama.SyncProducer 在 broker 不可用时会卡住 goroutine，而 sarama.AsyncProducer 又默认丢消息（config.Producer.Return.Successes = false），这些必须在封装层统一决策。
- 强制所有生产者启用
config.Producer.Return.Successes = true，失败时通过 channel 显式反馈 - 消费者启动前检查
config.Consumer.Group.Rebalance.Retry.Max是否设为 4（Sarama 默认 0，会导致首次 rebalance 失败后直接退出） - 禁止业务代码直接调用
consumer.ConsumePartition—— 它绕过消费者组协调，无法自动分配分区
如何设计可插拔的消息序列化器
Kafka 只传字节，但业务需要 JSON、Protobuf、甚至带 traceID 的结构化 payload。如果每个 handler 都自己 json.Marshal，就丧失了统一加 header、打日志、做 schema 校验的机会。
推荐用函数选项模式注入序列化器，而不是接口：
type MessageEncoder func(context.Context, interface{}) ([]byte, error)
<p>func WithJSONEncoder() MessageEncoder {
return func(ctx context.Context, v interface{}) ([]byte, error) {
return json.Marshal(v)
}
}</p><p>// 使用时
producer := NewKafkaProducer(WithJSONEncoder())
</p>
- 避免定义
Encoder接口——Go 中函数类型更轻量，且方便单元测试 mock - 序列化器必须接收
context.Context：某些 encoder 可能需要从 ctx 提取 traceID 或 tenantID 注入 header - 不要在 encoder 里做压缩（如 gzip）：压缩应在网络层或 broker 级别开启，否则不同语言消费者无法兼容
消费者组重启时如何避免重复消费
关键不在“不重复”，而在“重复可接受但必须幂等”。Sarama 的 config.Consumer.Offsets.AutoCommit.Enable = true 是陷阱：它按固定间隔提交 offset，若 consumer crash 在 commit 前，重启后会重复拉取。
正确做法是关掉自动提交，改用手动控制：
- 只在消息处理成功后调用
consumer.CommitOffsets，且传入当前 partition + offset + metadata - 使用
config.Consumer.Group.Session.Timeout = 45 * time.Second（不能低于 10s），避免短暂 GC 导致误踢出组 - 为每个 topic/partition 维护一个内存 offset map，避免并发 commit 冲突；不要依赖
consumer.GetOffset—— 它返回的是本地缓存，可能滞后
如果业务允许少量重复，比强一致性更容易落地。强行追求“恰好一次”往往引入额外存储依赖（如 Kafka transaction），反而增加运维复杂度。
怎么让模块支持优雅关闭和健康检查
容器编排场景下，SIGTERM 到来时，生产者要清空缓冲区，消费者要完成正在处理的消息再提交 offset。硬 kill 会导致消息丢失或重复。
封装层必须暴露 Close() 方法，并在内部阻塞等待：
func (p *Producer) Close() error {
p.asyncProducer.AsyncClose() // 非阻塞，清空 pending queue
- 消费者
Close()要先调用consumer.Close()，再等待consumer.Events()channel 关闭，确保所有事件（包括ConsumerError）被消费 - 健康检查不应只 ping broker，而应尝试发送一条测试消息并等待 success callback —— 这才能真实反映生产链路是否通畅
- 不要用
runtime.SetFinalizer做兜底关闭：它不可控，且可能在 GC 时触发，破坏 shutdown 顺序
最常被忽略的是：消费者组实例的 Close() 必须在所有 handler goroutine 结束后才调用，否则会 panic “closing closed channel”。这需要业务层显式通知 handler 退出，而不是靠 context.Done() 简单判断。
golang免费学习笔记（深入）：立即使用
在学习笔记中，你将探索golang的核心概念和高级技巧！
