Publisher发消息不丢的关键是Outbox模式：先将事件写入数据库outbox表并绑定业务事务，再由轮询协程异步发布，确保DB提交与消息发出的原子性。
Watermill 不是“配个中间件就能跑”的消息库，它要求你明确区分发布、订阅、路由、事务一致性四个层次。直接调 Publisher.Publish 发消息，在本地开发能过，上线后丢事件是大概率事件。
怎么让 Publisher 发消息不丢（尤其订单/支付类关键事件）
默认的 Publisher 是 fire-and-forget 模式，Kafka 或 AMQP 后端即使网络抖动、broker 不可用，也不会报错或重试。生产环境必须叠加 Outbox 模式。
- 先建一张
outbox表，字段至少包含id、topic、payload、created_at、published_at - 用
outbox.NewPublisher(db, realPublisher, logger)封装原始Publisher，它会把消息写进outbox表并标记未发布 -
outboxPub.Publish(ctx, "order_topic", msg)必须在数据库事务tx.Commit()之前调用——否则事务回滚时，outbox 记录也丢了 - 另起一个轮询协程（如每 100ms 查一次
published_at IS NULL），调用realPublisher.Publish并更新published_at
别试图自己手写事务内 Publish + defer rollback，Watermill 的 outbox 包已处理幂等、失败跳过、并发安全等细节。
Subscriber 卡住或重复消费的常见配置陷阱
现象是日志里反复出现同一条消息，或某段时间完全没日志——大概率不是代码逻辑问题，而是 Subscriber 初始化参数没对齐中间件语义。
- Kafka 场景：必须显式设置
ConsumerGroup，否则每次重启都从 earliest 开始消费；SessionTimeout要大于 handler 最长执行时间，否则触发 rebalance - AMQP（RabbitMQ）场景：
AckMode默认是amqp.ManualAck，但 handler 里忘了调msg.Ack()就会无限重投；若业务允许丢失，可设为amqp.AutoAck - 所有场景下，handler 函数返回
nil, nil表示成功消费且不转发；返回[]*message.Message才触发 Router 下游分发
错误信息 context deadline exceeded 几乎总和 timeout 或 ack 配置有关，不是网络问题。
如何用 Router 统一管理多个 Topic 而不写一堆 goroutine
手写 for range subscriber.Messages() 套路容易漏 defer、难统一加中间件、无法共享上下文。Router 是 Watermill 的核心抽象，不是可选组件。
- 每个 handler 注册时指定
subscribeTopic（如"order.created"）和publishTopic（可为空） - 所有 handler 共享同一个
Subscriber实例，Router 内部用 channel 分流，避免连接爆炸 - 中间件如
retryMiddleware.Middleware、middleware.CorrelationID统一加在 Router 上，不用每个 handler 单独配 - handler 函数签名必须是
func(ctx context.Context, msg *message.Message) ([]*message.Message, error)
Router 启动后调 router.Run(context.Background())，它会自动拉取消息、分发、重试、记录关联 ID——别再自己写 for-loop。
本地开发用 GoChannel，上线切 Kafka/RabbitMQ 的关键切换点
gochannel.NewGoChannel 只适合单机调试，它不持久、不支持多消费者、无 offset 管理。切换时有三个硬性检查项：
- 替换初始化代码：从
gochannel.NewPublisher改为kafka.NewPublisher或amqp.NewPublisher，传入真实 broker 地址 - Subscriber 必须用对应实现：
kafka.NewSubscriber要传ConsumerGroup，amqp.NewSubscriber要传QueueName和AckMode - Outbox 模式仍需保留——GoChannel 本身不提供事务保障，上线后更不能依赖内存通道保消息
最易忽略的是：GoChannel 的 Messages() 返回的是内存 channel，而 Kafka/AMQP 的 Messages() 是阻塞读取 broker，超时、重连、offset 提交全由底层驱动控制，不能假设行为一致。
