beego 集成 kafka 应选用 kafka-go 而非 sarama，因其避免 goroutine 卡死、支持 context、简化 offset 提交；需在 appstart/appstop 中统一管理 reader 生命周期，并正确配置 minbytes、maxwait、commitinterval 等 5 个关键参数。
Beego 本身不内置 Kafka 支持，集成 Kafka 的关键不在 Beego 框架层，而在于 Go 客户端选型与生命周期管理——用 kafka-go 替代 sarama，并在 Beego 的 AppStart 和 AppStop 钩子中正确启停消费者/生产者。
为什么别用 sarama 在 Beego 里连 Kafka
sarama 在 Beego 这类 Web 框架中容易引发隐性故障：
-
sarama.AsyncProducer.Errors()是一个无缓冲 channel，不持续消费就会卡住 goroutine，Beego 进程无法优雅退出 -
sarama.ConsumerGroup的Setup()和Cleanup()是同步阻塞的，若里面做 DB 初始化或 HTTP 调用，会拖慢整个 rebalance，导致消费者组长时间失联 - offset 提交必须显式调用
session.CommitOffsets()，handler panic 或提前 return 时极易漏提交，造成重复消费 - 没有原生
context.Context支持，超时控制、链路追踪都要手动包一层，Beego 的ctx.Input.Ctx很难自然注入
kafka-go Reader 必须显式配置的 5 个参数
kafka-go.Reader 默认配置只适合本地调试，上生产前必须重设：
-
MinBytes: 1：避免小流量下因等待凑够默认 10KB 而卡住，尤其在 Beego 处理事件驱动型请求时 -
MaxWait: 100 * time.Millisecond：平衡实时性与吞吐，设成 1s 会导致消息延迟飙升 -
CommitInterval: 1 * time.Second：必须设置！否则依赖ReadMessage自动提交，handler panic 后 offset 就丢了 -
PartitionWatchInterval: 30 * time.Second：防止 broker 重启或 topic 动态扩缩容时每秒都触发 rebalance -
StartOffset显式设为kafka.FirstOffset或kafka.LastOffset，别依赖默认值（它可能随版本变）
在 Beego 中启动和关闭 kafka-go Reader 的正确姿势
不能在 controller 里临时 new Reader，必须作为全局资源在 Beego 生命周期中统一管理：
- 在
models/init.go或独立kafka/client.go中定义全局*kafka.Reader变量 - 在
beego.AppConfig读取kafka.brokers、kafka.topic等配置项，而非硬编码 - 在
beego.AppStart钩子里启动 goroutine 拉取消息，并用signal.Notify监听os.Interrupt和syscall.SIGTERM - 在
beego.AppStop钩子里调用reader.Close()，并wait消费 goroutine 退出（用sync.WaitGroup或context.WithTimeout） - handler 函数内务必用
defer func() { if r := recover(); r != nil { log.Error(r) } }()，否则 panic 会跳过 commit 逻辑
真正难的不是把消息从 Kafka 读出来，而是当 broker 暂时不可用、consumer panic 后恢复、或流量突增时，你的 offset 是否还准、消息是否仍在预期顺序里、下游会不会被同一条消息打穿——这些全靠 CommitInterval、MaxWait、错误 recover 和 AppStop 里的 close 时机共同决定。
