Watermill本身不提供现成“事务收件箱网关”,需用outbox.NewPublisher将事件写入数据库outbox表,与业务事务原子性绑定,并由outbox.NewOutbox启动轮询协程异步投递,确保业务提交成功则事件必落地。

Watermill 本身不提供“事务收件箱网关”这个现成组件,但你可以用它组合出符合该语义的实现:本地事务 + Outbox 表 + 消息发布解耦。关键不是找一个叫“收件箱网关”的包,而是把数据库写入、Outbox 轮询、消息投递三者串成一条可控链路。
为什么不能直接用 Publisher.Publish 做事务收件箱?
因为 Publisher.Publish 是纯异步发送,和数据库事务完全隔离。订单创建成功后调 publisher.Publish,若 Kafka 网络抖动或 broker 不可用,事件就丢了——这违反“事务收件箱”最核心要求:**业务提交成功,事件必须可靠落地**。
常见错误写法:
tx, _ := db.Begin()
_, _ := tx.Exec("INSERT INTO orders ...")
publisher.Publish(ctx, "order.created", msg) // ⚠️ 这里失败,事务却可能 commit 成功
tx.Commit()正确路径只有一条:把事件先落库(outbox 表),再由独立协程读取并发布。
立即学习“go语言免费学习笔记(深入)”;
如何用 outbox.NewPublisher 绑定事务?
outbox.NewPublisher 是 Watermill 提供的封装,它把事件写入数据库的 outbox 表,并确保和业务事务原子性一致。它不是替代 Publisher,而是包装它。
使用 uber-go/dig 在 Golang 中实现依赖注入——基于反射容器,提供 Provide/Invoke、dig.In/dig.Out 参数与结果对象,支持命名依赖等
-
outbox.NewPublisher构造时需传入*sql.DB、底层Publisher(如kafka.NewPublisher)和logger - 调用
outboxPub.Publish时,必须在tx.Commit()之前执行;它内部会插入一条outbox记录,且和你的业务 SQL 在同一个事务中 -
outbox表结构需包含:id、topic、payload(JSON)、created_at、processed_at(NULL 初始值) - 别手动建轮询 goroutine;Watermill 的
outbox.NewOutbox提供了Start方法,它会启动后台协程持续扫描未处理记录
示例关键行:
outboxPub := outbox.NewPublisher(db, kafkaPub, logger)
err := outboxPub.Publish(ctx, "order.created", msg) // ✅ 必须在 tx.Commit() 前
if err != nil {
tx.Rollback()
return err
}
tx.Commit() // 此时 outbox 记录已写入,且和订单记录一起持久化如何让收件箱真正“松耦合”?
松耦合 ≠ 不依赖数据库,而是指:业务代码不感知消息投递细节,也不承担重试、死信、幂等逻辑。这些应由 Outbox 轮询器统一处理。
- 轮询器(
outbox.NewOutbox)应独立于 HTTP handler 启动,通常放在main或服务初始化阶段 - 轮询间隔建议设为 100–500ms,太短增加 DB 压力,太长影响事件延迟
- 轮询器必须配置重试策略和死信 Topic:失败消息(如 Kafka 不可达)应发到
outbox.dlq,而非无限重试阻塞后续 - 避免在轮询器中做复杂业务逻辑——它只负责“读 outbox → 发 Kafka/RabbitMQ → 标记 processed_at”。业务逻辑仍在 Subscriber 的 Handler 中
注意:outbox.NewOutbox 默认不自动清理已处理记录,需另配定时任务或在应用层定期 DELETE FROM outbox WHERE processed_at IS NOT NULL AND created_at 。
Router 和 Subscriber 怎么配合收件箱语义?
收件箱只管“投递”,而“消费”仍由标准 Watermill Router 承担。你不需要为 outbox 表单独写订阅逻辑——所有事件都走统一消息通道(如 Kafka topic),Router 按 topic 分发即可。
- 所有业务事件(
order.created、payment.succeeded)都通过outboxPub.Publish写入,最终出现在 Kafka 对应 topic -
Router.AddHandler注册的 handler 处理这些事件,返回nil, nil表示仅消费;返回新*Message可触发下游转发 - 若需跨服务幂等,应在 handler 开头查
processed_events表(基于 event_id + topic 唯一索引),而不是依赖 outbox 表——outbox 是发件箱,不是去重依据
真正容易被忽略的点:outbox 表的 processed_at 字段更新时机,必须严格对应底层 Publisher.Publish 成功回调。Watermill 的 outbox 包默认使用 msg.Ack() 触发更新,这意味着你必须用 ManualAck 模式,并在 handler 中显式调 msg.Ack(),否则 outbox 记录永远卡在未处理状态。

















