实战使用 Golang 构建一个简单的发布订阅模式

实战使用 Golang 构建一个简单的发布订阅模式

为什么直接用 sync.Map 而不是自己加锁的 map？

在构建发布订阅系统时，核心挑战之一就是高频并发读写。多个 goroutine 可能同时发布事件，订阅者的注册和取消也随时在发生。如果自己用普通的 map 搭配 sync.RWMutex，一个常见的坑就是在遍历过程中删除键值对，或者因读写竞争导致 map 迭代器失效，进而引发 panic。

免费影视、动漫、音乐、游戏、小说资源长期稳定更新！ 👉 点此立即查看 👈

而 sync.Map 的设计初衷就是解决这类问题。它原生支持并发安全的增删查改，尤其在读多写少的场景下，性能表现更为稳定。直接用它，相当于把并发安全的复杂性的交给了标准库。

这里有个关键细节需要注意：sync.Map 的 Range 方法采用的是快照语义。这意味着，遍历开始后新增的订阅者，不会被本次通知覆盖。对于大多数发布订阅场景来说，这反而是合理的行为——你广播的是“此刻”的快照状态。但如果业务要求强一致性广播，必须通知到“发布瞬间所有活跃的订阅者”，那就得考虑其他方案了，比如用 channel 配合一个中心化的注册管理器。

• 别把 sync.Map 当普通 map 用：必须使用 Load、Store 这类方法，直接写 map[key] = val 会导致编译失败。
• 合理封装值类型：建议将 value 封装为 chan interface{} 或自定义结构体，避免直接存储裸函数指针，这不利于后续的生命周期管理。
• 避免过度设计：如果主题（topic）数量极少，使用 map[string][]func(interface{}) 加一个全局互斥锁的方案可能更直观、更简单。

如何安全地实现 Unsubscribe 并防止 goroutine 泄漏？

实现退订功能时，一个典型的错误是只从 map 里删除回调函数，却忘记关闭对应的接收 channel。尤其是在使用 chan interface{} 作为消息管道时，一个未被关闭的 channel 会让监听它的 goroutine 永远阻塞在接收操作上，最终导致 goroutine 堆积和内存泄漏。

正确的做法是，在 Unsubscribe 逻辑里，不仅要删除 map 中的条目，还必须显式地关闭对应的 channel，并确保消费者 goroutine 能够检测到这个关闭信号后优雅退出。

• 推荐独立消费协程：为每个订阅者启动一个独立的 goroutine，使用 for msg := range ch 循环来接收消息。这样，一旦 channel 被关闭，for-range 循环就会自动结束。
• 管理好生命周期：不要在 Subscribe 返回的函数里直接启动 goroutine 去处理消息（这会使生命周期难以控制）。更好的做法是返回一个可供外部调用的 unsubscribe 函数。
• 精准移除回调：如果采用函数切片来存储处理器（handler），在 Unsubscribe 时需要比较函数地址（通过 unsafe.Pointer）或使用唯一 ID 来标识，否则无法精准移除特定的回调函数。

Publish 时要不要做深度拷贝或同步等待？

先说结论：默认情况下都不建议。发布订阅模式的本质是为了解耦，Publish 方法应当快速返回，避免因为某个订阅者处理过慢而拖垮整个发布流程。因此，典型的实现方式是：遍历当前主题的所有订阅者，对每个 channel 或 handler 进行异步调用（例如 go f(msg)）或非阻塞发送（使用 select 的 default 分支）。

但是，这里有两个数据安全的陷阱需要警惕：第一，如果消息（msg）是一个可变的结构体指针（比如 *User），那么多个订阅者并发修改它，就会引发数据竞争。第二，如果消息是一个大对象，反复进行值传递可能会给垃圾回收（GC）带来压力。

• 优先考虑传值：对于小结构体，或者由 interface{} 包裹的不可变值（如 string、int、struct{ID int; Name string}），直接传递即可。
• 传递指针需谨慎：仅在明确所有订阅者都只读、且发布者之后也不再修改该对象时，才考虑传递指针。否则，应该使用 proto.Clone 或手动进行深度拷贝。
• 切忌同步等待：绝对不要在 Publish 方法里使用 sync.WaitGroup 来等待所有订阅者处理完毕——这完全违背了发布订阅异步和解耦的设计契约。

用 context.Context 控制订阅生命周期是否必要？

很有必要，尤其是在订阅者是临时任务（比如在 HTTP 请求处理期间监听某个日志事件），或者需要超时自动退订的场景下。context.Context 的作用不仅仅是传递取消信号，它更能统一管理 goroutine 的退出、资源清理以及超时逻辑。

举个例子，可以在 Subscribe 函数中接收一个 ctx 参数，在内部启动一个 goroutine 来监听 ctx.Done() 通道。一旦触发取消信号，就自动执行 Unsubscribe 并关闭对应的 channel。这样，上层调用方就无需手动调用退订，也彻底避免了忘记清理资源的问题。

• 慎用默认上下文：不要把 context.Background() 作为默认值传入 Subscribe——它永远不会被取消，等于主动放弃了生命周期管理。
• 自然延伸上下文：如果订阅者内部还需要发起网络请求等操作，将接收到的 ctx 透传给下游的 http.Client 等组件，是自然而然的做法，并非额外负担。
• 选择合适的上下文：context.WithCancel 创建的子上下文最适合控制单次订阅的生命周期；而 context.WithTimeout 则非常适合“监听5秒内首次出现的事件”这类有明确时限的场景。

最后，在实际开发和测试中，有几个细节比模式本身更容易决定成败：主题名称的拼写一致性（大小写、空格、前缀）、订阅者处理函数的 panic 捕获（避免一个 handler 的 panic 导致整个发布循环崩溃）、以及测试时对时间相关逻辑（比如用 time.AfterFunc 做延迟发布）的模拟。把这些边角打磨好，系统才算真正稳健。