golang如何使用sarama操作Kafka_golang sarama操作Kafka使用技巧-Golang-PHP中文网

必须设 RequiredAcks = sarama.WaitForAll 保证消息落盘，配 Timeout ≥ 10s 防超时，用 ByteEncoder 避免截断，NewClusterAdmin 创建 Topic 更可靠，ConsumerGroup 必须设非空 Group.ID，务必 defer Close() 防连接泄漏。

golang如何使用sarama操作kafka_golang sarama操作kafka使用技巧

同步生产者默认不保证消息落盘，必须显式设 RequiredAcks = sarama.WaitForAll，否则 SendMessage 返回成功 ≠ 消息安全。

为什么 `SendMessage` 成功了消息却丢了

常见错误现象：日志里打印 “partition:0 offset:123”，但 Kafka 集群重启后查不到这条消息；或 leader 切换后消费端跳过该 offset。

config.Producer.RequiredAcks 默认是 sarama.NoResponse，Broker 收到就回包，不等写磁盘也不等副本同步
设成 sarama.WaitForAll 才强制等所有 ISR 副本写入成功，这是生产环境底线配置
同时要配 config.Producer.Timeout ≥ 10s：Kafka broker 默认 request.timeout.ms=30000，Go 客户端超时若更短（比如默认 3s），会提前报 context deadline exceeded，但 broker 可能还在后台重试
别用 sarama.StringEncoder 传二进制数据：遇到 \x00 会截断，改用 sarama.ByteEncoder([]byte{...})

`NewClusterAdmin` 创建 Topic 才可靠

本地用 kafka-topics.sh 创建的 topic，在多节点集群中常出现分区不均、ISR 数不足、甚至 UNKNOWN_TOPIC_OR_PARTITION 错误。

NewClusterAdmin 能实时拉取元数据，创建时校验 ReplicationFactor 是否 ≤ 当前可用 broker 数（设 3 却只启 2 个 broker 会卡住）
自动处理权限/配额类错误，比如 TOPIC_AUTHORIZATION_FAILED，而命令行工具只报模糊异常
版本必须匹配：config.Version = sarama.V2_8_0_0（对应 Kafka 2.8+），低版本 client 设高 version 会直接失败
别复用 AdminClient 实例做高频建 Topic 操作：它内部有连接池，但长期运行需手动调 RefreshMetadata()，否则元数据过期导致创建失败

消费者该用 `ConsumePartition` 还是 `ConsumerGroup`

单机调试打日志用前者，订单/支付等业务流必须用后者——否则无法分摊负载、无法自动提交 offset、重启就重头消费。

Golang Uber Dig

使用 uber-go/dig 在 Golang 中实现依赖注入——基于反射容器，提供 Provide/Invoke、dig.In/dig.Out 参数与结果对象，支持命名依赖等

下载

ConsumePartition 场景：日志采集、离线分析。自己管 offset：启动时选 sarama.OffsetOldest 或 sarama.OffsetNewest，后续靠 pc.HighWaterMarkOffset() 记位置，否则没状态保存
ConsumerGroup 场景：在线业务。config.Group.ID 绝不能为空字符串，空值会导致 consumer 拒绝加入 group，日志只报 group coordinator not available，极难排查
消费循环里禁用 time.Sleep 控速：它会阻塞整个 Messages() channel。正确做法是调大 config.ChannelBufferSize，再对单条消息用 context.WithTimeout 处理超时

`defer p.Close()` 不是可选项

不显式关闭生产者或消费者，会导致 TCP 连接泄漏，短生命周期服务（如 HTTP handler 里临时发消息）容易触发 too many open files。

哪怕只调一次 SendMessage，也要 defer producer.Close()
异步生产者 AsyncProducer 更要注意：它内部有 goroutine 和 channel，不 Close 会永久占用资源
消费者 Consumer 或 ConsumerGroup 同理，defer consumer.Close() 必须写在 main 函数或作用域末尾

真正麻烦的不是写不出代码，而是 RequiredAcks 没设对、Group.ID 留空、或者忘了 Close() —— 这些点线上一出问题，日志里往往只显示很泛的错误码，得翻源码才能定位。