必须用watermill对接kafka/rabbitmq实现可靠事件处理;其明确区分publisher/subscriber职责,需配置持久化、预声明队列、正确ack与重试机制,裸用chan会导致并发写崩溃和goroutine泄漏。

用 github.com/ThreeDotsLabs/watermill 对接 Kafka / RabbitMQ,处理跨服务事件
当事件需持久化、保证顺序、支持重试或跨节点分发时,必须脱离内存 chan interface{},接入专业消息系统。Watermill 是 Go 生态中少有的专注事件驱动、明确区分 Publisher 和 Subscriber 职责的库。
容易踩的坑:ConsumerGroup 未设置导致所有实例重复消费;Ack 时机错误造成消息丢失;或误把 router.Publish 当同步调用,实际是异步投递,不等 broker 确认就返回。
-
Publisher必须配置publisher.Confirm(Kafka)或amqp.Publishing.DeliveryMode = amqp.Persistent(RabbitMQ),否则重启即丢 -
Subscriber启动前需确保 topic / queue 已声明,Watermill 不自动建队列,靠router.AddHandler时传入的queueName或topic触发初始化 - 每个 handler 函数签名固定为
func(ctx context.Context, msg *message.Message) error,返回nil才触发Ack,panic 或返回非 nil error 会触发重试(默认 3 次)
裸用 chan interface{} 做全局事件通道,为什么一上线就 panic
这不是“写不写得出来”的问题,而是“一上线就出事”的问题。常见现象包括:fatal error: concurrent map writes、goroutine 泄漏卡死在 select { case ch 、handler panic 后整个接收循环停摆。
根本原因在于:未加锁的 map[string][]chan interface{} 并发写、未 close 的 channel 持续接收、无缓冲 channel 阻塞投递线程。
- 订阅时必须用
sync.RWMutex保护 map,且Subscribe返回的取消函数要显式close(ch),不是 defer - 投递必须用
select { case ch 非阻塞,缓冲区至少 <code>make(chan interface{}, 16) - 每个 handler 必须包裹
recover()+ 日志,并运行在独立 goroutine 中,避免单个 panic 拖垮全局
什么时候该换消息中间件,而不是硬撑内存总线
信号比性能指标更关键:只要出现以下任一情况,说明内存总线已不可靠。
- 需要跨进程通信(比如用户服务在 node-A,风控服务在 node-B)
- 要求事件不丢失——内存总线崩溃即丢,而 Kafka / RabbitMQ 支持副本、ack、重试与死信
- 单机 goroutine 数持续 > 5000,且多数卡在
select { case ch —— 这是背压已满的明确指标 - 业务要求严格按时间顺序处理事件(内存总线无法保证多 subscriber 间的全局时序)
真正难的不是选哪个中间件,而是把“事件契约”提前定义清楚:字段名、类型、版本号、是否允许 null,这些一旦上线就很难改。
golang免费学习笔记(深入):立即使用
在学习笔记中,你将探索golang的核心概念和高级技巧!











