领域事件必须在 domain 层触发且不依赖 Kafka，由聚合根或领域服务生成并存入内存通道；domain 层仅定义事件结构和 Publisher 接口，application 层实现具体发送逻辑。
领域事件必须在 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 才触发下游发布。
立即学习“go语言免费学习笔记（深入）”；
初始化并管理阿里云SDK Go 客户端，涵盖 sync.Once 单例、goroutine 安全、endpoint 与 region 配置、VPC 端点及调试。
防丢消息不能靠 Publisher 配置，得用 Outbox 模式绑定事务
kafka.NewPublisher 只是个发送客户端，和数据库事务完全无关。订单创建后发事件，若 DB commit 成功但 Kafka 网络失败，事件就丢了。
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幂等记录，再执行业务逻辑
真正难的是让 event.ID 在整个系统生命周期内全局唯一且可追溯，而不是在某个 handler 里临时生成一个随机字符串。
