不能直接用chan做全局通知总线,因其单向、无缓冲或固定缓冲导致发送阻塞;多goroutine并发写入panic;缺乏topic过滤、动态订阅与失败重试能力;裸chan无法可靠处理事件分发,必须封装带路由、非阻塞投递和生命周期管理的总线。

为什么不能直接用 chan 做全局通知总线
因为 chan 是单向、无缓冲或固定缓冲的,一旦消费者没及时读,就会阻塞发送方;多个 goroutine 并发写入同一 chan 会 panic;也没法按 topic 过滤、动态订阅、失败重试。真实业务里,你得处理「谁关心什么事件」「丢了怎么补」「并发写是否安全」这些事,裸 chan 压根撑不住。
所以得封装一层:统一注册中心 + topic 路由 + 非阻塞投递 + 可选持久化兜底。核心不是“怎么发”,而是“怎么让发和收解耦且可靠”。
-
chan只适合模块内短生命周期、确定协程数的通信,别当总线用 - 如果用
chan做总线,必须自己加锁 + map + select default + goroutine 池,等于重造轮子 - Go 标准库没有内置 pub/sub,别指望
sync.Map或context能替代
用 github.com/ThreeDotsLabs/watermill 快速搭一个可扩展总线
它不是最轻量的,但解决了绝大多数痛点:支持多种后端(内存、Kafka、RabbitMQ)、自动序列化、中间件链(日志、重试、限流)、topic 级别订阅隔离。关键是——它不强制你改业务逻辑,只在发/收两端加几行胶水代码。
示例:定义一个用户注册成功事件,广播给邮件、短信、风控三个服务:
type UserRegistered struct {
UserID string `json:"user_id"`
Email string `json:"email"`
At time.Time `json:"at"`
}
<p>// 发送
msg := watermill.NewMessage(watermill.NewUUID(), []byte(<code>{"user_id":"u123","email":"a@b.c","at":"2024-06-01T12:00:00Z"}</code>))
msg.Metadata.Set("event_type", "user_registered")
publisher.Publish("events.user_registered", msg)</p><p>// 订阅(每个 handler 独立 goroutine)
sub, _ := subscriber.Subscribe(context.Background(), "events.user_registered")
for msg := range sub {
var event UserRegistered
json.Unmarshal(msg.Payload, &event)
// 处理邮件...
msg.Ack()
}</p>
- 内存模式下,
publisher和subscriber可共用一个watermill.NewPublisher实例,无需额外启动服务 - topic 名(如
"events.user_registered")是字符串,建议按domain.action命名,避免硬编码 - 别忽略
msg.Ack()—— 内存模式下不调用会导致消息堆积在 channel 中,最终 OOM
自己封装轻量总线时,sync.Map + chan 组合的关键细节
如果你坚持手写(比如资源受限或只想用标准库),sync.Map 存 topic → chan 映射,每个 topic 对应一个带缓冲的 chan,再配一个 dispatcher goroutine 拉取并分发。但必须处理三件事:goroutine 泄漏、chan 满溢出、订阅者退出。
使用 uber-go/dig 在 Golang 中实现依赖注入——基于反射容器,提供 Provide/Invoke、dig.In/dig.Out 参数与结果对象,支持命名依赖等
关键代码片段:
type Bus struct {
topics sync.Map // string → *topicChan
}
<p>type topicChan struct {
ch chan Message
closed chan struct{}
}</p><p>func (b <em>Bus) Publish(topic string, msg Message) {
if v, ok := b.topics.Load(topic); ok {
t := v.(</em>topicChan)
select {
case t.ch </p><p>func (b <em>Bus) Subscribe(topic string) topicChan).ch
}</em></p>
- 缓冲大小不是越大越好:
make(chan Message, 100)占用 100×结构体大小内存,大 payload 时极易爆 heap - 没有取消机制:
Subscribe返回的chan无法关闭,订阅者退出后 goroutine 仍在往ch写,必须靠外部信号或 context 控制生命周期 -
sync.Map的LoadOrStore在高并发下比mutex+map更快,但别在里面存复杂对象(比如带 mutex 的 struct)
通知失败时,要不要加本地持久化兜底
要看场景。内部服务间通知(如订单创建后发库存扣减),网络抖动导致一次失败可以接受重试;但用户注册后发欢迎邮件,丢了就是客诉。这时候就得落盘再投。
推荐方案:用 SQLite 或 BadgerDB 存失败消息(含 topic、payload、retry_count、next_retry_at),搭配一个后台 goroutine 定时扫描重投。不要用文件写,容易丢、难并发、没事务。
- SQLite 的
PRAGMA journal_mode = WAL能扛住每秒几百次写入,够中小业务用 - 失败消息的
next_retry_at别用固定间隔(如 5s),要用指数退避:time.Now().Add(time.Second * time.Duration(math.Pow(2, float64(retryCount)))) - 持久化层和通知总线要解耦——总线只管“尽力投递”,失败回调交给单独的
onDeliveryFailedhandler 处理
真正麻烦的不是怎么存,是怎么判断“到底算失败”。HTTP 超时?5xx?还是下游返回 {"code":500,"msg":"busy"}?这个边界必须由业务方定义清楚,总线层只提供钩子,不替你决策。
golang免费学习笔记(深入):立即使用
在学习笔记中,你将探索golang的核心概念和高级技巧!










