直接用 channel 做事件分发易丢事件，因消费者阻塞会导致生产者阻塞或缓冲溢出丢数据；应解耦投递与消费，用 sync.Map+chan 实现类型化广播，并通过 select{default:} 避免阻塞。
为什么直接用 channel 做事件分发容易丢事件
Go 中常见误区是把 chan interface{} 当成通用事件总线：所有生产者往里塞，一个消费者从头读。问题在于——如果消费者处理慢，channel 满了就阻塞生产者，或带缓冲但缓冲区溢出时直接 panic 或丢数据。真实业务中，事件来源多样（HTTP、定时器、消息队列），不能因某条事件处理卡住而拖垮整个输入链。
真正可行的路径是「解耦投递」和「独立消费」：事件进队列不阻塞，每个处理器按自己节奏拉取、重试、限流。
- 用
chan struct{}或sync.WaitGroup控制启动/关闭信号，别用select{default:}轮询空 channel - 避免在
select中对同一channel多次读取——Go 不保证顺序，且可能漏收 - 事件结构体必须是值类型或明确管理指针生命周期，否则并发写入字段会引发 data race
用 sync.Map + chan 实现轻量级事件注册与广播
不需要引入第三方库也能支持多监听器。核心思路：用 sync.Map 存 event type → []chan Event 映射，每次 Publish 时遍历对应 channel 列表发送副本。注意不是共享引用，而是深拷贝或只读视图。
下面是一个最小可运行示例，支持按字符串类型订阅：
立即学习“go语言免费学习笔记（深入）”；
type Event struct {
Type string
Data map[string]interface{}
}
<p>type EventBus struct {
mu sync.RWMutex
bus sync.Map // string → []chan Event
}</p><p>func (e *EventBus) Subscribe(topic string, ch chan Event) {
e.mu.Lock()
defer e.mu.Unlock()</p><pre class='brush:php;toolbar:false;'>if v, ok := e.bus.Load(topic); ok {
chs := v.([]chan Event)
e.bus.Store(topic, append(chs, ch))
} else {
e.bus.Store(topic, []chan Event{ch})
}}
func (e *EventBus) Publish(topic string, evt Event) { if v, ok := e.bus.Load(topic); ok { for _, ch := range v.([]chan Event) { select { case ch <- evt: default: // 丢弃或记录 warn，不阻塞发布者 } } } }
- 每个
chan Event应由调用方自行创建并管理生命周期（例如用make(chan Event, 10)） -
Publish中的select{default:}是关键——防止监听器消费太慢拖垮事件源 - 不推荐用
interface{}做 topic，字符串更易调试、序列化、打日志
如何让事件处理器支持失败重试与上下文取消
真实场景下，事件可能依赖外部服务（DB、HTTP），需控制超时、重试、取消。不能把 context.Context 塞进 Event 结构体——它不是事件数据，而是执行约束。
正确做法：在启动处理器 goroutine 时传入 context，并封装为可取消的消费循环：
func (e *EventBus) StartProcessor(ctx context.Context, topic string, handler func(Event)) {
ch := make(chan Event, 10)
e.Subscribe(topic, ch)
<pre class='brush:php;toolbar:false;'>go func() {
defer close(ch)
for {
select {
case evt, ok := <-ch:
if !ok {
return
}
select {
case <-ctx.Done():
return
default:
// 加入重试逻辑（如指数退避）
retry := 0
for retry < 3 {
if err := processWithTimeout(ctx, evt); err != nil {
retry++
time.Sleep(time.Second * time.Duration(1<<uint(retry)))
continue
}
break
}
}
case <-ctx.Done():
return
}
}
}()}
- 每个处理器必须有独立
ctx，避免一个处理器 cancel 影响其他处理器 -
processWithTimeout应使用ctx传递超时，而非硬编码time.Sleep - 不要在 handler 里启动未受控的 goroutine，否则无法被
ctx取消
什么时候该换用消息队列而不是自建 channel 总线
当出现以下任一情况，说明已超出内存级事件总线能力边界：
- 需要跨进程/跨机器分发事件（
channel仅限单进程） - 要求事件持久化、至少一次投递（
channel无 ACK、无存储） - 消费者扩容需动态发现（自建方案需额外服务注册+健康检查）
- 事件积压后需按优先级消费（如紧急告警 > 日志上报）
此时应切换到 Kafka / NATS / Redis Streams 等。Go 生态已有成熟客户端：segmentio/kafka-go、nats-io/nats.go，它们内置了 offset 管理、rebalance、TLS、压缩等，比手撸更稳。
最容易被忽略的一点：本地 channel 总线和远端消息队列不是互斥关系，而是分层——内部模块间用 channel 快速响应，对外投递交由专用 producer goroutine 异步刷到 Kafka。
