极简事件总线用 Go Channel 实现进程内发布订阅的工程化封装一、事件总线的迷思为什么进程内通信不需要 MQ发送一条消息通知其他模块——这个需求在微服务架构中的标准答案是消息队列Kafka/RabbitMQ/NATS。但在单体应用或单个进程中引入一个外部 MQ 会让架构肥胖。启动一个 MQ 服务只是为了在UserService和NotificationService之间传递一条用户已注册的消息——这就像在两个人面对面聊天时非要用微信发文字。进程内事件总线的核心价值是用最少的代码实现模块间的松耦合。它不需要网络、不需要序列化、不需要运维——只是一个发消息、收消息的机制。Go 的 Channel 天然适合做进程内事件总线。但直接使用 Channel 有几个问题需要手动管理订阅关系、不支持通配符匹配、无法处理订阅者 panic 导致发布者崩溃。graph TB subgraph Pub[发布者] P1[UserService] P2[OrderService] end subgraph Bus[EventBus] B[Channel Topic 路由] end subgraph Sub[订阅者] S1[NotificationServicebr/订阅: user.*] S2[AnalyticsServicebr/订阅: user.registered] S3[AuditServicebr/订阅: *] end P1 --|user.registered| Bus P2 --|order.created| Bus Bus --|user.registered| S1 Bus --|user.registered| S2 Bus --|user.registered| S3 Bus --|order.created| S3 style Bus fill:#4dabf7,color:#fff二、事件总线的核心设计Topic 匹配与并发安全一个进程内事件总线需要解决三个核心问题Topic 路由发布者指定 Topic如user.registered订阅者声明感兴趣的 Topic 模式如user.*表示所有用户相关事件。路由逻辑在发布时执行——找到所有匹配的订阅者并发通知。并发安全多个 goroutine 可能同时发布和订阅。需要使用读写锁保护订阅列表。但锁的粒度要控制——发布和订阅是低频操作读写锁足够不需要无锁数据结构。错误隔离一个订阅者的 panic 不能影响其他订阅者更不能影响发布者。每个订阅者的调用需要独立的 recover。三、EventBus 的完整实现package eventbus import ( context fmt log path/filepath sync time ) // Event 是总线中传递的消息 type Event struct { Topic string Data interface{} Timestamp time.Time } // Handler 是事件处理函数 type Handler func(ctx context.Context, event Event) error // Subscription 表示一个订阅关系 type Subscription struct { id string pattern string // 支持通配符: user.*, *.created, handler Handler } // EventBus 进程内事件总线 type EventBus struct { mu sync.RWMutex subs []*Subscription nextID int timeout time.Duration closed bool } func New(timeout ...time.Duration) *EventBus { t : 5 * time.Second if len(timeout) 0 { t timeout[0] } return EventBus{ subs: make([]*Subscription, 0), timeout: t, } } // Subscribe 订阅匹配 pattern 的事件返回取消订阅函数 func (b *EventBus) Subscribe(pattern string, handler Handler) func() { b.mu.Lock() defer b.mu.Unlock() id : fmt.Sprintf(sub-%d, b.nextID) b.nextID sub : Subscription{id: id, pattern: pattern, handler: handler} b.subs append(b.subs, sub) // 返回取消订阅的函数 return func() { b.unsubscribe(id) } } // SubscribeSync 同步订阅用于测试 func (b *EventBus) SubscribeSync(pattern string, handler Handler) func() { return b.Subscribe(pattern, func(ctx context.Context, e Event) error { handler(ctx, e) return nil }) } func (b *EventBus) unsubscribe(id string) { b.mu.Lock() defer b.mu.Unlock() for i, sub : range b.subs { if sub.id id { b.subs append(b.subs[:i], b.subs[i1:]...) return } } } // Publish 发布事件。默认异步执行不等待订阅者完成 func (b *EventBus) Publish(topic string, data interface{}) { b.publishAsync(topic, data) } // PublishSync 同步发布事件等待所有订阅者完成 func (b *EventBus) PublishSync(topic string, data interface{}) []error { return b.publishInternal(topic, data, true) } // publishAsync 异步发布事件 func (b *EventBus) publishAsync(topic string, data interface{}) { event : Event{Topic: topic, Data: data, Timestamp: time.Now()} b.mu.RLock() matched : b.matchSubscribers(topic) b.mu.RUnlock() // 每个订阅者在独立的 goroutine 中执行 for _, sub : range matched { go func(s *Subscription) { ctx, cancel : context.WithTimeout(context.Background(), b.timeout) defer cancel() b.safeCall(ctx, s, event) }(sub) } } // publishInternal 内部发布逻辑控制同步/异步 func (b *EventBus) publishInternal(topic string, data interface{}, sync bool) []error { event : Event{Topic: topic, Data: data, Timestamp: time.Now()} b.mu.RLock() matched : b.matchSubscribers(topic) b.mu.RUnlock() if sync { var errs []error for _, sub : range matched { ctx, cancel : context.WithTimeout(context.Background(), b.timeout) defer cancel() if err : b.safeCall(ctx, sub, event); err ! nil { errs append(errs, err) } } return errs } for _, sub : range matched { go func(s *Subscription) { ctx, cancel : context.WithTimeout(context.Background(), b.timeout) defer cancel() b.safeCall(ctx, s, event) }(sub) } return nil } // safeCall 安全调用订阅者捕获 panic func (b *EventBus) safeCall(ctx context.Context, sub *Subscription, event Event) (err error) { defer func() { if r : recover(); r ! nil { err fmt.Errorf(handler panic for %s: %v, sub.pattern, r) log.Printf(EventBus: %v, err) } }() return sub.handler(ctx, event) } // matchSubscribers 找到所有匹配的订阅者 func (b *EventBus) matchSubscribers(topic string) []*Subscription { var matched []*Subscription for _, sub : range b.subs { if matchPattern(sub.pattern, topic) { matched append(matched, sub) } } return matched } // matchPattern 简单的通配符匹配: user.* 匹配 user.registered, 匹配所有 func matchPattern(pattern, topic string) bool { if pattern { return true } matched, _ : filepath.Match(pattern, topic) return matched } // Close 关闭事件总线清理资源 func (b *EventBus) Close() { b.mu.Lock() defer b.mu.Unlock() b.closed true b.subs nil }使用示例func main() { bus : eventbus.New(3 * time.Second) defer bus.Close() // 订阅用户注册事件 cancelFn : bus.Subscribe(user.registered, func(ctx context.Context, e Event) error { user : e.Data.(User) fmt.Printf(发送欢迎邮件给 %s\n, user.Email) return nil }) defer cancelFn() // 订阅所有用户相关事件 bus.Subscribe(user.*, func(ctx context.Context, e Event) error { fmt.Printf([审计] 用户事件: %s\n, e.Topic) return nil }) // 发布事件 bus.Publish(user.registered, User{ID: 123, Email: helloexample.com}) // 等待异步处理完成 time.Sleep(100 * time.Millisecond) }四、这个 EventBus 的边界条件不支持消息持久化进程重启后所有未处理的事件丢失。如果业务需要事件不丢失应该用外部 MQ。但对于通知类场景发邮件、清理缓存丢失一条通知是可以接受的。不支持远程调用所有订阅者必须在同一进程内。如果需要跨进程通信这是正确的架构选择——进程内事件总线不应该试图做它不擅长的事。内存压力异步模式下如果订阅者的处理速度跟不上发布速度goroutine 会堆积。使用context.WithTimeout可以限制单个订阅者的执行时间但无法限制并发 goroutine 的总数。不适用场景需要事件持久化的关键业务如订单状态变更跨进程/跨服务的事件通信需要严格顺序处理的场景异步模式下无法保证顺序五、总结EventBus 用不到 150 行 Go 代码实现了进程内的发布订阅模式。它让模块间通信从直接调用变成事件驱动降低了耦合度。落地路径先在项目中识别一个模块做了一件事后需要通知其他模块的场景如用户注册后发邮件、订单创建后清理缓存然后用 EventBus 替代直接的函数调用最后在集成测试中用PublishSync同步模式验证事件的正确性。少即是多。进程内的事件通知不需要引入 MQ——100 行代码的 EventBus 足矣。