用Outbox模式保障Publisher不丢数据：先写outbox表再发消息，与业务事务一致；Subscriber需按中间件特性配Ack、timeout；Router统一分发多topic；SSE需SSERouter+手动管理连接生命周期。
怎么用 Publisher 发消息不丢数据？
Watermill 的 Publisher 默认是“发完即忘”，但生产环境里订单、支付这类事件绝不能丢。关键不是换中间件，而是加一层事务一致性保障。
- 本地开发用
gochannel.NewPublisher没问题，但上线必须切到支持持久化的后端（如kafka.NewPublisher或amqp.NewPublisher） - 真正防丢的核心是
Outbox模式：把事件先写进数据库的outbox表，再由单独的轮询协程读取并发布——这样和业务数据库操作在同一个事务里 - 别直接调
publisher.Publish后就 commit 事务；要用 Watermill 提供的OutboxPublisher封装，它会自动绑定事务对象
示例片段：outboxPub := outbox.NewPublisher(db, publisher, logger)err := outboxPub.Publish(ctx, "order_topic", msg) —— 这行必须在 tx.Commit() 前调用。
Subscriber 消费卡住或重复消费怎么办？
这不是 Watermill 的 bug，而是默认配置没对齐业务语义。Kafka 和 RabbitMQ 的 offset 提交策略、重试机制、Ack 模式都不同，Subscriber 初始化时漏配参数就会出问题。
- 用 Kafka 时，
ConsumerGroup必须显式设置，否则每次启动都是新消费者，从头开始拉消息 - 用 AMQP（RabbitMQ）时，
AckMode要设为amqp.AutoAck或amqp.ManualAck，别用默认值——默认是ManualAck，但 handler 里忘了调msg.Ack()就会无限重复投递 - 如果 handler 处理慢，Kafka 订阅者可能触发
session.timeout.ms导致 rebalance，表现为“突然停止消费”。这时得调大SessionTimeout配置，并确保 handler 执行时间短于该值
常见错误现象：context deadline exceeded 报错 + 消息反复出现在日志里，基本就是 Ack 或 timeout 配置没对上。
立即学习“go语言免费学习笔记（深入）”；
如何让一个服务同时处理多个 topic 而不写一堆 Subscribe 循环？
别手写多个 for range messages goroutine，Watermill 的 Router 就是干这个的。它能把不同 topic 的消息按规则分发到不同 handler，还能复用中间件、统一错误处理。
- 每个 handler 用
router.AddHandler注册，subscribeTopic设成对应 topic，publishTopic可设为空（不转发）或另一个 topic - handler 函数签名必须是
HandlerFunc类型，返回[]*Message才能触发转发；返回nil, nil表示只消费不转发 - 所有 handler 共享同一个
Subscriber实例，Router 内部用 channel 多路复用，比手动启多个 goroutine 更轻量、更可控
注意：Router 启动后不能动态增删 handler，改配置要重启服务；热更新得靠蓝绿部署+消费组切换，不是代码层能解决的。
HTTP 推送实时事件时，SSE 怎么和 Watermill 对接？
Watermill 本身不处理 HTTP，得靠 watermill-http 扩展包里的 SSERouter。它本质是个“订阅者适配器”：把消息流转成 HTTP 流响应，但容易忽略两个关键点。
-
SSERouter必须传入一个已初始化的Subscriber（比如从 Kafka 来），不能自己 new 一个——它不负责连接 broker，只负责转换 -
StreamAdapter的Validate方法常被跳过，结果导致所有消息推给所有客户端。实际要用它做路由判断，比如解析msg.Metadata.Get("user_id")匹配请求 URL 中的 ID - SSE 连接是长连接，
GetResponse返回的response必须是可序列化结构体，且不能含指针或未导出字段，否则 JSON 编码失败静默丢消息
最容易被忽略的是连接生命周期管理：SSERouter 不管客户端断连，得自己在 GetResponse 里检查 w.(http.CloseNotifier) 或用 context 判断是否超时，否则 goroutine 泄露。
