watermill 是 go 语言中用于构建事件驱动架构的抽象层，而非 kafka 客户端；它通过统一接口封装多种消息中间件（如 kafka/rabbitmq），提供标准化消息结构、中间件链和可插拔的发布/订阅语义，适用于多消费者、事务性消费及未来协议切换场景。
Watermill 是什么，为什么选它而不是直接用 sarama
Watermill 不是 Kafka 客户端，而是事件驱动架构的抽象层，watermill 本身不处理网络通信，它依赖底层驱动（比如 watermill-kafka）来对接 Kafka。如果你已经用 sarama 手写消费者组、手动提交 offset、自己做重试和死信路由，那 Watermill 能帮你把这部分逻辑标准化——它提供统一的 Message 结构、内置中间件链、可插拔的发布/订阅语义，且不强制依赖任何框架。
但注意：watermill 不是“更轻量”的替代品，它是更高一层的封装；真正轻量的是你不用重复实现消息生命周期管理。如果项目只需要发几条日志，直接用 sarama-sync-producer 更合适；但一旦涉及多消费者、失败重试、事务性消费、或未来要切换到 RabbitMQ/NATS，Watermill 的抽象就开始值回票价。
初始化 Kafka 驱动时必须设置的 3 个关键配置项
Watermill 的 Kafka 驱动（watermill-kafka）默认不自动创建 topic，也不自动 commit offset，所有行为都靠配置驱动。漏掉任意一项，大概率出现「消息不消费」「重复消费」「启动卡住」。
-
ConsumerGroup：必须非空字符串，否则kafka.NewSubscriber会 panic —— 即使你只用单实例消费，也要设成类似"svc-order-processor" -
InitialOffset：默认是sarama.OffsetNewest，意味着只消费启动后的新消息。调试时建议显式设为sarama.OffsetOldest，否则可能以为「没消息」其实是跳过了历史数据 -
Brokers：必须是[]string，不能是逗号分隔的单字符串。写成"localhost:9092"会静默失败；正确写法是[]string{"localhost:9092"}
示例片段：
cfg := kafka.DefaultKafkaConfig()
cfg.Brokers = []string{"localhost:9092"}
cfg.ConsumerGroup = "my-service"
cfg.InitialOffset = sarama.OffsetOldest
subscriber, err := kafka.NewSubscriber(cfg, watermillLogger)
如何避免 Message 处理失败后无限重试
Watermill 默认对 panic 或返回 error 的 Handler 进行无限重试，这在 Kafka 场景下极易导致 partition 卡死（因为 offset 不提交）。必须主动干预重试策略。
Go 配置库，使用 spf13/viper — 分层优先级（flag > env >file > KV > default），提供 BindPFlag/BindPFlags、SetEnvPrefix + SetEnvKeyReplace 等功能。
- 用
middleware.Retry{MaxRetries: 3}替代全局重试，它会在第 3 次失败后把Message标记为已处理（即提交 offset），但不会丢弃消息 —— 你需要额外监听FailedMessagestopic 或写入 DLQ - 不要在 handler 里直接
panic，而应返回errors.New("validation failed")，否则 middleware 无法捕获 - 若需死信投递，必须手动调用
msg.SetMetadata("dead_letter_topic", "dlq.orders")，然后在 middleware 中检查并转发，Watermill 不自动创建或路由 DLQ
Go module 依赖冲突常见于 sarama 版本不匹配
Watermill 的 watermill-kafka v1.4+ 锁定 sarama v1.33+，但很多老项目还在用 sarama v1.27（因兼容旧 Kafka broker）。go mod tidy 时会出现 cannot load sarama.Client 类错误。
解决方式只有两个：
- 升级整个项目到
sarama v1.33+（推荐，v1.33 开始支持 Kafka 3.0+ 的 SCRAM-SHA-512 认证） - 降级
watermill-kafka到 v1.3.0（对应sarama v1.27），但会丢失Assignor自定义和IsolationLevel支持
验证方法：运行 go list -m all | grep sarama，确保输出唯一一行且版本一致。混用不同 major 版本的 sarama 几乎必然导致 interface conversion: interface {} is *sarama.ConsumerGroup, not *sarama.consumerGroup 这类类型断言失败。
实际项目里，最常被忽略的是 broker 版本与 sarama 版本的隐式绑定——Kafka 2.8 要求 sarama >= v1.28，否则 FetchVersion 请求直接被拒绝，但错误日志只显示「context deadline exceeded」，看不出根源。
golang免费学习笔记（深入）：立即使用
在学习笔记中，你将探索golang的核心概念和高级技巧！
