Beego 集成 Kafka 必须通过 sarama 手动实现，禁止在 controller 中裸写逻辑；应在 main.go 的 beego.Run() 前初始化全局 SyncProducer 并设 RequiredAcks=WaitForAll、Idempotent=true；消费者须用 ConsumerGroup 在独立 goroutine 中长期运行，禁用 auto-commit，手动调用 session.CommitContext 确保 offset 持久化。
Beego 本身不内置 Kafka 支持，必须通过 Go 客户端库（如 sarama）手动集成。直接在 Beego 的 controllers 或 models 中裸写生产/消费逻辑极易引发 goroutine 泄漏、配置硬编码、offset 管理混乱等问题——这不是“能不能连上”的问题，而是“连上了怎么不出错”的问题。
如何在 Beego 启动时初始化 Kafka 生产者
Beego 的 main.go 中的 beego.Run() 前是唯一安全的初始化时机。不能把 sarama.SyncProducer 实例放在 controller 方法里每次 new，否则连接池失控、内存暴涨。
-
sarama.NewSyncProducer是重量级对象，应作为全局变量或注入到AppConfig中复用 - 必须显式设置
config.Producer.RequiredAcks = sarama.WaitForAll，否则单节点故障就丢消息 - 启用幂等性：
config.Producer.Idempotent = true，但要求config.Net.MaxOpenRequests = 1，否则 panic - 别忽略错误回调：设
config.Producer.Return.Errors = true，否则SendMessages失败静默吞掉
示例片段：
var kafkaProducer sarama.SyncProducer
func initKafka() error {
cfg := sarama.NewConfig()
cfg.Producer.RequiredAcks = sarama.WaitForAll
cfg.Producer.Idempotent = true
cfg.Net.MaxOpenRequests = 1
cfg.Producer.Return.Errors = true
p, err := sarama.NewSyncProducer([]string{"localhost:9092"}, cfg)
if err != nil {
return err
}
kafkaProducer = p
return nil
}
为什么不能在 Beego Controller 里直接消费 Kafka
Beego 的 HTTP 请求生命周期短，而 Kafka 消费者需长期运行、自动 rebalance、管理 offset。把 sarama.ConsumerGroup 塞进 controller 会导致：请求一结束 consumer 就被 GC；多个请求并发触发重复启动；无法响应 Rebalance 事件。
- Kafka 消费必须作为独立 goroutine 或子进程启动，与 HTTP server 并行运行
- 推荐在
main.go初始化后，用go func() { ... }()启动消费者组，监听ctx.Done()做优雅退出 - 务必使用
sarama.ConsumerGroup而非sarama.Consumer，否则无法支持多实例水平扩展 - offset 提交策略选
auto.commit风险极高——服务重启瞬间未处理完的消息就被标记为已消费
关键配置项：config.Consumer.Offsets.AutoCommit.Enable = false，后续在业务处理成功后手动调用 session.CommitContext。
Beego 应用中如何安全传递 Kafka 消息上下文
HTTP handler 和 Kafka consumer 属于不同执行流，不能共享 request context。常见错误是把 Beego 的 context.Context 直接传给 consumer 回调，导致 cancel 信号误杀消费逻辑。
- consumer 内部应使用自己独立的
context.WithCancel控制生命周期 - 若需关联请求 ID 追踪（如订单创建事件 → Kafka → 库存扣减），应在生产消息时把 traceID 写入
msg.Headers，而非依赖外部 context - 避免在 consumer 中调用
beego.Controller.Ctx相关方法——它只在 HTTP 请求期间有效 - 日志打点统一用结构化字段，例如
log.Printf("kafka_consume topic=%s partition=%d offset=%d", msg.Topic, msg.Partition, msg.Offset)
Beego + Kafka 上线前必须验证的三件事
很多团队卡在“本地能跑，线上丢消息”，本质是环境差异没对齐。
- 确认 Kafka broker 地址用的是集群内网 DNS（如
kafka-headless:9092），不是localhost或宿主机 IP - 检查
advertised.listeners配置是否正确——Docker/K8s 环境下若只配了PLAINTEXT://localhost:9092，外部消费者永远连不上 - Beego 应用的
ulimit -n必须 ≥ 65536，sarama默认每个 broker 维护多个长连接，文件描述符不够会静默断连
真正棘手的从来不是“怎么发”，而是“发了之后谁来保活、谁来记位置、谁来兜底重试”。Kafka 在 Beego 里不是插件，是需要单独设计生命周期的基础设施组件。
