Go (Golang) 中的事件总线与观察者模式：发布/订阅实现

Go (Golang) 中的事件总线与观察者模式：从理论到实践

观察者模式及其以事件总线形式的实现，是构建松耦合、可扩展系统的基础工具。在 Go 中，秉承“通过通道通信，而非共享内存”的哲学，这些模式显得尤为优雅。本文将讲解如何使用 goroutine 和通道在 Go 中创建自己的事件总线，并探讨典型的使用场景。

观察者模式：本质与问题

假设您有一个系统，其中一个对象（发布者）需要通知多个其他对象（订阅者）其状态的变化。没有观察者模式，您就必须将发布者与每个订阅者紧密耦合。观察者模式通过引入订阅接口解决了这个问题：发布者维护一个订阅者列表，并在事件发生时调用它们的 Update() 方法。

事件总线是一个集中式的总线，充当中间人的角色。任何组件都可以发布事件，任何其他组件都可以订阅该事件，而无需知道彼此的存在。

为什么 Go 是事件总线的理想选择？

• Goroutine 和通道：无阻塞的异步事件处理。
• 接口：可以轻松为处理器定义契约。
• 无继承：通过通道和函数进行组合，比经典的面向对象实现更自然。

Go 中事件总线的简单实现

我们从基础版本开始，其中事件是字符串，订阅者是处理函数。

package main

import (    "fmt"    "sync"    "time")

// EventBus — 我们的事件总线type EventBus struct {    subscribers map[string][]func(string)    mu          sync.RWMutex}

// NewEventBus 创建一个新总线func NewEventBus() *EventBus {    return &EventBus{        subscribers: make(map[string][]func(string)),    }}

// Subscribe 将处理函数订阅到事件func (eb *EventBus) Subscribe(event string, handler func(string)) {    eb.mu.Lock()    defer eb.mu.Unlock()    eb.subscribers[event] = append(eb.subscribers[event], handler)}

// Publish 发布事件并同步调用所有订阅者func (eb *EventBus) Publish(event string, data string) {    eb.mu.RLock()    defer eb.mu.RUnlock()    if handlers, ok := eb.subscribers[event]; ok {        for _, handler := range handlers {            handler(data)        }    }}

func main() {    bus := NewEventBus()        bus.Subscribe("user.login", func(data string) {        fmt.Printf("登录: %s", data)    })        bus.Subscribe("user.login", func(data string) {        fmt.Printf("审计: %s 已登录系统", data)    })        bus.Publish("user.login", "alice@example.com")    // 输出:    // 登录: alice@example.com    // 审计: alice@example.com 已登录系统}

这段代码可以工作，但有一个问题：处理函数是同步调用的。如果一个处理函数挂起，它将阻塞整个总线。

使用通道的异步版本

我们通过带缓冲的通道来增加异步性。每个订阅者将获得自己的通道。

type AsyncEventBus struct {    subscribers map[string][]chan string    mu          sync.RWMutex}

func NewAsyncEventBus() *AsyncEventBus {    return &AsyncEventBus{        subscribers: make(map[string][]chan string),    }}

func (eb *AsyncEventBus) Subscribe(event string, buffer int) <-chan string {    eb.mu.Lock()    defer eb.mu.Unlock()    ch := make(chan string, buffer)    eb.subscribers[event] = append(eb.subscribers[event], ch)    return ch}

func (eb *AsyncEventBus) Publish(event string, data string) {    eb.mu.RLock()    defer eb.mu.RUnlock()    if channe