Kafka 消息未送达的根本原因是 RequiredAcks 默认为 NoResponse 且未显式 Close():必须设为 WaitForAll、确保 replication.factor≥2、设置合理 Timeout 和 Return.Successes,Consumer Group ID 非空,避免复用 config,ReadMessage 需 context 超时控制,advertised.listeners 应配置可路由地址,SyncProducer 需线程安全封装并显式关闭。

Go 用 Kafka 做消息队列,不配对 RequiredAcks 和不显式 Close(),基本等于没发——消息可能根本没进磁盘,连接还越积越多。
sarama.NewSyncProducer 为什么发了却查不到消息
现象是 SendMessage 返回 nil,但 Kafka 消费端收不到,或者 broker 日志里压根没写入记录。根本原因不是网络或 topic 不存在,而是默认配置下 config.Producer.RequiredAcks = sarama.NoResponse:客户端发完 UDP 式一扔就返回,broker 连页缓存都没写。
-
config.Producer.RequiredAcks必须设为sarama.WaitForAll(即 acks=-1),且对应 topic 的replication.factor >= 2,否则会退化为只等 leader,仍可能丢 -
config.Producer.Timeout至少设为10 * time.Second:Kafka broker 默认request.timeout.ms=30000,Go 客户端若超时更短,会提前报错中断,但 broker 端可能还在重试 - 必须开
config.Producer.Return.Successes = true,否则你连发到哪个partition、什么offset都不知道 - 别用
sarama.StringEncoder传二进制数据:含\x00会被截断,统一用sarama.ByteEncoder([]byte(...))
ConsumerGroup 启动就报 group coordinator not available
错误日志里只有这句,没其他线索,十有八九是 config.Group.ID 为空字符串。sarama 对空 ID 非常严格,不会报“ID 不能为空”,而是直接拒绝加入 group,导致 coordinator 无法分配分区。
-
config.Group.ID必须是非空字符串,比如"order-processor-v1",不能是""或只含空白符 - 别复用同一个
config实例启动多个 consumer:group id 冲突会导致 offset 覆盖或消费跳变 - 首次启动时,如果 topic 尚未创建,consumer 会卡在 metadata 请求上,建议先用
ClusterAdmin创建 topic,而非依赖 auto-create
kafka-go ReadMessage 卡住不动的真正原因
不是 Kafka 挂了,也不是网络断了,而是 offset 提交策略和 context 控制没配好。ReadMessage 默认阻塞,但“卡住”往往是因为消费者处理慢 + 没设超时 + offset 提交太懒。
使用 uber-go/dig 在 Golang 中实现依赖注入——基于反射容器,提供 Provide/Invoke、dig.In/dig.Out 参数与结果对象,支持命名依赖等
立即学习“go语言免费学习笔记(深入)”;
- 别依赖
AutoCommit: true:consumer 重启后会从上次自动保存位置开始,可能跳过未处理完的消息 - 批量提交更稳:每处理 10–100 条调一次
reader.CommitMessages(ctx, msgs),避免高频提交拖垮吞吐 - 每次
ReadMessage必须套context.WithTimeout(ctx, 5*time.Second):防死锁或 broker 响应异常导致 goroutine 永久挂起 - 本地调试连不上?90% 是 broker 的
advertised.listeners还设成localhost,Docker 或远程 client 会按这个地址去连,结果连错——该值应设为可路由的 IP,如PLAINTEXT://127.0.0.1:9092
生产环境别在 init() 里初始化 SyncProducer
SyncProducer 不是线程安全的。init() 全局初始化后,多个 goroutine 并发调 SendMessage 会 panic,错误类似 concurrent map read and map write,但堆栈不直观,容易误判为 Kafka 配置问题。
- 封装成结构体,用单例或对象池管理:比如
type KafkaProducer struct { p sarama.SyncProducer },每次发消息前检查是否已初始化 - 必须显式调
producer.Close():不关会导致 goroutine 泄露 + TCP 连接堆积,broker 日志出现Connection reset或too many open files - 高并发场景下,NewSyncProducer 创建成本高,别每次发都 new 一个;但也不能全局复用一个实例扛所有请求——得在可用性与资源间做取舍
最易被忽略的是:RequiredAcks 设对了,但 topic 的 replication-factor 没跟上;Close() 写了,但 defer 在错误分支里没执行到;Group.ID 有值,但不同服务用了同一 ID 导致 offset 相互覆盖。这些点不手工验证,靠日志很难定位。


















