当前位置：首页 > news >正文

极简事件总线：用 Go Channel 实现进程内发布订阅的工程化封装

news 2026/7/10 0:06:58

极简事件总线：用 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[NotificationService<br/>订阅: user.*] S2[AnalyticsService<br/>订阅: user.registered] S3[AuditService<br/>订阅: *] 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[i+1:]...) 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: "hello@example.com"}) // 等待异步处理完成 time.Sleep(100 * time.Millisecond) }