如何在Golang架构中使用Watermill实现领域事件解耦-Golang-PHP中文网

领域事件必须在 domain 层触发且不依赖 kafka，由聚合根或领域服务生成并存入内存通道；domain 层仅定义事件结构和 publisher 接口，application 层实现具体发送逻辑。

如何在golang架构中使用watermill实现领域事件解耦

领域事件必须在 domain 层触发，但不能直接调用 Kafka

领域事件不是“发给消息队列的通知”，而是“业务事实已发生”的声明。它必须由 domain 层的聚合根或领域服务在状态变更后立即生成，比如 Order.Confirm() 成功执行后，调用 o.Publish(OrderConfirmed{ID: o.ID}) —— 这个 Publish 方法只是把事件塞进一个内存通道或切片，不涉及序列化、网络调用或 Kafka 客户端。

常见错误是把 Publish 写成直接调用 kafka.Producer.SendMessage，这会让 domain 层强依赖外部基础设施，破坏分层边界，也导致单元测试无法隔离。

domain 层只定义事件 struct（如 OrderPaidEvent）和 Publisher 接口（如 type Publisher interface { Publish(Event) }）
application 层实现该接口，注入具体的消息发送器（KafkaPublisher 或 InMemoryPublisher）
测试时用内存实现替换，断言事件类型和字段即可，无需启动 Kafka

Router 是 Watermill 的强制入口，别手写 for-range 消费循环

很多人从 _examples/basic/1-your-first-app/ 入门，看到 subscriber.Subscribe("topic") 后直接 for msg := range messages，这在单 topic 场景下看似可行，但会立刻撞上三个硬伤：

无法复用中间件（比如重试、去重、超时）——每个 for 循环得自己裹 context.WithTimeout 和 recover
多个 topic 时要启多个 goroutine，手动管理生命周期极易泄漏；router 内部用 channel 多路复用，更轻量可控
没法统一错误处理：一个 handler panic，整个 subscriber loop 就停摆；而 router 里每个 handler 隔离运行，挂掉不影响其他

正确姿势是初始化 message.NewRouter，再用 router.AddHandler 注册每个 topic 的处理函数。handler 函数签名必须是 HandlerFunc 类型，返回 nil, nil 表示只消费不转发，返回 []*Message, nil 才触发下游发布。

防丢消息不能靠 Publisher 配置，得用 Outbox 模式绑定事务

kafka.NewPublisher 只是个发送客户端，和数据库事务完全无关。订单创建后发事件，若 DB commit 成功但 Kafka 网络失败，事件就丢了。

Golang Naming

Go（Golang）命名规范 — 包括包、构造函数、结构体、接口、常量、枚举、错误、布尔值、接收器、getter/setter、函数等。

下载

Watermill 不提供跨 DB + MQ 的原子提交，Go 也没有两阶段提交支持。Kafka 事务也只能保证“写入 Kafka 和写入自身 topic 原子性”，无法涵盖你的 MySQL/PostgreSQL。

正确做法是“本地消息表 + 定时扫描”或“Kafka 事务 + 业务表同属一个 Kafka cluster”
Outbox 表结构通常为：outbox(id UUID PRIMARY KEY, topic TEXT, payload BYTEA, created_at TIMESTAMPTZ)
事务内先 INSERT 到 outbox，再 COMMIT；独立协程轮询 outbox 并调用 Publisher.Publish，成功后 DELETE 记录

消费者幂等必须靠 event.ID + 状态记录，别用 order_id 做唯一键

“一次处理”不等于“一次送达”。Kafka 可能重复投递，NATS JetStream 也可能重传，所以消费者必须自己判重。

最可靠的方式是以 event.ID（全局唯一，建议用 ULID 或 UUIDv7）为 key，在 DB 或 Redis 中记录“该事件是否已成功处理”。

不要用业务字段（如 order_id）单独做幂等键——同一订单可能触发多个事件（OrderCreated、OrderPaid），混在一起会误判
Redis 方案：SETNX event_id:xxx 1 EX 3600，成功则处理，失败则跳过
DB 方案：建表 event_consumed(event_id TEXT PRIMARY KEY, processed_at TIMESTAMPTZ)，INSERT ON CONFLICT DO NOTHING
避免在事务中查 DB 判重后再发 DB 更新——若事务中途失败，状态未写入，下次又会重试；应先 INSERT 幂等记录，再执行业务逻辑