跳到主要内容

观察者模式

问题​

Go 中如何实现观察者/发布订阅模式？

答案​

接口实现​

// 观察者接口
type Observer interface {
    OnEvent(event Event)
}

// 事件
type Event struct {
    Type string
    Data interface{}
}

// 事件总线
type EventBus struct {
    mu        sync.RWMutex
    observers map[string][]Observer
}

func NewEventBus() *EventBus {
    return &EventBus{observers: make(map[string][]Observer)}
}

func (eb *EventBus) Subscribe(eventType string, obs Observer) {
    eb.mu.Lock()
    defer eb.mu.Unlock()
    eb.observers[eventType] = append(eb.observers[eventType], obs)
}

func (eb *EventBus) Publish(event Event) {
    eb.mu.RLock()
    defer eb.mu.RUnlock()
    for _, obs := range eb.observers[event.Type] {
        obs.OnEvent(event)
    }
}

函数式 + Channel 实现​

Go 更地道的方式是用 channel：

type EventBus struct {
    mu   sync.RWMutex
    subs map[string][]chan Event
}

func (eb *EventBus) Subscribe(topic string) <-chan Event {
    eb.mu.Lock()
    defer eb.mu.Unlock()
    ch := make(chan Event, 16) // 带缓冲
    eb.subs[topic] = append(eb.subs[topic], ch)
    return ch
}

func (eb *EventBus) Publish(topic string, data interface{}) {
    eb.mu.RLock()
    defer eb.mu.RUnlock()
    event := Event{Type: topic, Data: data}
    for _, ch := range eb.subs[topic] {
        // 非阻塞发送，防止慢消费者阻塞发布者
        select {
        case ch <- event:
        default:
            log.Println("subscriber channel full, dropping event")
        }
    }
}

// 使用
bus := NewEventBus()

// 订阅
ch := bus.Subscribe("user.created")
go func() {
    for event := range ch {
        fmt.Println("收到事件:", event.Data)
    }
}()

// 发布
bus.Publish("user.created", User{Name: "Alice"})

常见面试问题​

Q1: 用接口还是 channel 实现观察者模式？​

答案：

  • 接口方式：同步执行，适合简单场景
  • channel 方式：天然异步，解耦更彻底，Go 惯用方式
  • 函数回调：最轻量，func(Event) 作为订阅者

实际项目中，小规模用函数回调，需要解耦用 channel，跨服务用消息队列。

Q2: channel 实现需要注意什么？​

答案：

  • 缓冲：用带缓冲的 channel 避免慢消费者阻塞发布者
  • 关闭：取消订阅时 close channel，消费者 range 自动退出
  • goroutine 泄漏：确保消费者 goroutine 最终退出

相关链接​