kratos 中集成 kafka 需封装生产者为可注入组件并注册到 di 容器，消费者组须禁用自动提交 offset 并显式提交，分区保序需统一 key 序列化且单 goroutine 处理每 partition，panic 应通过 setup/cleanup 控制 rebalance 而非仅 recover。
在 Kratos 微服务中直接集成 Kafka，不是“配个客户端就能用”的事——它涉及生产者可靠性、消费者组生命周期管理、错误重试边界、以及与 Kratos 依赖注入和中间件链的耦合方式。不处理好这些，消息会丢、消费会卡、服务启停时会 panic。
kratos 中如何注册 Kafka 生产者而不破坏 DI 容器
Kratos 的 service.Init 阶段不适合直接 new 一个 sarama.SyncProducer 并全局变量持有：它无法被容器管理，也无法参与 graceful shutdown。正确做法是把生产者封装为可注入的组件：
- 定义接口如
type Producer interface { SendMessage(ctx context.Context, msg *sarama.ProducerMessage) (int32, int64, error) } - 实现 struct 包含
*sarama.SyncProducer和配置字段，并实现Close()方法 - 在
internal/biz/producer.go中注册为 singleton：用app.Register或di.NewSet注入到容器 - 务必在
app.Run的app.WithServer之外、app.WithService之前完成初始化，否则可能早于配置加载
漏掉 Close() 实现或未在 app shutdown hook 中调用，会导致连接泄漏，Kafka broker 端积累大量 stale connection。
消费者组（Consumer Group）必须手动管理 offset 提交时机
使用 sarama.NewConsumerGroup 时，默认 config.Consumer.Offsets.AutoCommit.Enable = true，看似省事，但在 Kratos 场景下极易出错：
- auto-commit 是后台 goroutine 异步提交，无法保证“消息处理成功后再 commit”
- 若 handler 中 panic 或 context 被 cancel（比如服务重启），已处理但未 commit 的 offset 会丢失
- Kratos 的 middleware（如 logging、tracing）若依赖 context.Value，而 auto-commit 在 handler 外部触发，trace ID 可能为空
应设为 false，并在业务 handler 显式调用 session.CommitContext(ctx)，且只在无 error 且逻辑真正完成时调用。注意：该操作本身可能失败，需判断返回 error 并决定是否重试或告警，不能忽略。
Go语言（Golang）1.26.0版本提供 Go 官方 Windows amd64 MSI 安装包下载入口，版本号 1.26.0，可用于旧项目维护、兼容性测试和指定版本开发环境配置。
分区策略与消息顺序：别迷信 NewHashPartitioner
想让同一用户 ID 的消息落到同一 partition 以保序？sarama.NewHashPartitioner 看似合理，但它对 key 做的是 Murmur2 哈希，且不支持自定义 hash 函数。问题在于：
- 如果 key 是 string 类型但内容含不可见字符（如 BOM、\u200b），哈希结果不稳定
- Go 的
string和 Java 客户端对同一字符串的哈希值可能不同（Murmur2 实现细节差异） - 若业务要求“同 user_id 消息严格 FIFO”，必须确保所有生产者都用相同 key 序列化逻辑，且 consumer group 内只有一个 consumer 实例（否则 partition 内有序，跨 partition 无序）
更稳妥的做法是：key 固定为 user_id 字符串，禁用 config.Producer.Idempotent（避免幂等性干扰 hash），并在 consumer 端对每个 partition 单 goroutine 处理（用 sync.Map + channel 控制 per-partition worker 数量）。
panic 时的 recover 不足以保护 Kafka 消费循环
很多人在 ConsumeClaim 的 for-loop 里套一层 defer func() { recover() }()，以为能兜住 panic。但这是错的：
- recover 只捕获当前 goroutine 的 panic，而 sarama 的 consumer group 会为每个 partition 启一个 goroutine；一个 panic 不影响其他 partition，但该 partition 的消费会永久停滞
- panic 后 session 已失效，继续调用
session.MarkOffset会 panic，且不会自动 rebalance - 正确做法是：在 handler 入口做
ctx.Err()检查，所有外部调用（DB、HTTP、log）都带超时；用errors.Is(err, context.Canceled)区分主动退出和异常；panic 只允许出现在不可恢复的编程错误中，并由上层监控捕获
真正关键的不是 recover，而是让每个 partition 的消费 goroutine 在退出前明确通知 consumer group 主循环，触发 rebalance —— 这需要你重写 ConsumerGroupHandler 的 Setup/Cleanup 方法，而不是依赖默认行为。
golang免费学习笔记（深入）：立即使用
在学习笔记中，你将探索golang的核心概念和高级技巧！
