一、为什么需要观察者#

你写了一个数据服务，当数据更新时需要刷新 UI、写日志、发通知。最直接的做法是在数据更新的函数末尾依次调用这三个函数。但很快需求来了：加一个缓存失效的处理。你得再改数据更新的代码。然后又要加一个埋点上报，再改一次。每次新增功能都要动核心逻辑，违反了开闭原则。

看一段典型的耦合代码：

1
func (s *DataService) Update(id string, data Data) error {
2
    // 核心逻辑：更新数据库
3
    if err := s.db.Save(id, data); err != nil {
4
        return err
5
    }
6

7
    // 下面全是"附带的"副作用，和核心逻辑无关
8
    s.ui.Refresh(id)           // 刷新 UI
9
    s.logger.Log("updated", id) // 写日志
10
    s.notifier.Send(id)        // 发通知
11
    s.cache.Invalidate(id)     // 失效缓存——后加的
12
    s.tracker.Report(id)       // 埋点上报——又后加的
13
    return nil
14
}

每次加新功能都要改 Update 方法，而且 DataService 要引入越来越多不相关的依赖——它根本不需要知道 UI 怎么刷新、日志怎么写。更糟糕的是，这些副作用之间也可能互相影响：如果 notifier.Send 抛了异常，后面的 cache.Invalidate 就不会执行，缓存就脏了。

用观察者模式重构后：

1
func (s *DataService) Update(id string, data Data) error {
2
    if err := s.db.Save(id, data); err != nil {
3
        return err
4
    }
5
    // 只管发射事件，谁关心谁自己处理
6
    s.emitter.Emit("data:updated", DataEvent{ID: id, Data: data})
7
    return nil
8
}
9

10
// 各处独立注册，互不干扰
11
s.emitter.On("data:updated", func(data any) {
12
    s.ui.Refresh(data.(DataEvent).ID)
13
})
14
s.emitter.On("data:updated", func(data any) {
15
    s.logger.Log("updated", data.(DataEvent).ID)
16
})

核心逻辑只做一件事：数据更新后发射事件。谁需要响应、怎么响应，全部在各自模块里注册。新增功能不再动 Update 方法，只需要 On 一下就行。

更深层的问题是耦合。数据服务不应该知道 UI 怎么刷新、日志怎么写。它只需要做一件事：数据变了，告诉关心的人。观察者模式把「一变多应」的关系抽象出来：主题维护订阅者列表，状态变化时遍历列表逐个通知。发送方不知道谁在监听，监听方也不需要轮询检查变化。

这种解耦是观察者模式无处不在的原因：从 DOM 的 addEventListener 到 Redux 的 store.subscribe，从 Node.js 的 EventEmitter 到 React 的 useEffect 清理模式，核心思想完全一致。

不过，解耦也带来了一个容易忽视的问题：内存泄漏。主题持有观察者的引用，如果你注册了观察者却忘了取消订阅，垃圾回收器就无法回收它及闭包捕获的变量。在单页应用中这是最常见的内存泄漏来源——组件销毁了，但事件监听器还在。React 的 useEffect 要求返回清理函数，正是为了解决这个问题：

1
useEffect(() => {
2
    emitter.on("data", handler)
3
    return () => emitter.off("data", handler) // 组件卸载时取消订阅
4
}, [])

一个实用法则：每写一行 on，就立即写一行 off。很多库让 on 返回取消订阅函数，就是为了降低遗忘概率。

二、现实类比#

报纸订阅。你订阅一次，每天早上报纸就送到门口。你不需要跑到报摊去查有没有新报纸——出版商主动推送给所有订户。不想看了？取消订阅，以后就不送了。

类比还可以再进一步：出版商不知道订户拿到报纸之后干什么——有人读头版，有人只看体育。出版商也不关心，它只管印和送。如果某个订户搬家了却忘了取消订阅，报纸还是会送到旧地址，这就对应了前面说的内存泄漏。

三、核心思想#

观察者模式建立一对多的依赖关系：当主题状态改变时，所有注册的观察者自动收到通知。主题不知道观察者会做什么——只是调用它们的回调。

四、变体与对比#

五、多语言实现#

1
type Listener func(data any)
2

3
type EventEmitter struct {
4
    listeners map[string][]Listener
5
}
6

7
func NewEmitter() *EventEmitter {
8
    return &EventEmitter{listeners: make(map[string][]Listener)}
9
}
10

11
// On 注册监听器，返回取消订阅的函数
12
func (e *EventEmitter) On(event string, listener Listener) func() {
13
    e.listeners[event] = append(e.listeners[event], listener)
14
    return func() {
15
        // 从切片中移除该监听器
16
        ls := e.listeners[event]
17
        for i, l := range ls {
18
            if &l == &listener {
19
                e.listeners[event] = append(ls[:i], ls[i+1:]...)
20
                return
21
            }
22
        }
23
    }
24
}
25

26
// Emit 通知所有订阅了该事件的监听器
27
func (e *EventEmitter) Emit(event string, data any) {
28
    for _, listener := range e.listeners[event] {
29
        // 错误隔离：防止单个监听器的 panic 中断整个通知链
30
        func() {
31
            defer func() {
32
                if r := recover(); r != nil {
33
                    log.Printf("监听器异常: %v", r)
34
                }
35
            }()
36
            listener(data)
37
        }()
38
    }
39
}

1
type Listener<T> = (data: T) => void;
2

3
class EventEmitter<Events extends Record<string, unknown>> {
4
  private listeners = new Map<keyof Events, Set<Listener<any>>>();
5

6
  // 订阅事件，返回取消订阅函数
7
  on<K extends keyof Events>(event: K, listener: Listener<Events[K]>): () => void {
8
    if (!this.listeners.has(event)) {
9
      this.listeners.set(event, new Set());
10
    }
11
    this.listeners.get(event)!.add(listener);
12
    return () => this.off(event, listener);
13
  }
14

15
  off<K extends keyof Events>(event: K, listener: Events[K]): void {
16
    this.listeners.get(event)?.delete(listener);
17
  }
18

19
  // 通知所有订阅者，带错误隔离
20
  emit<K extends keyof Events>(event: K, data: Events[K]): void {
21
    this.listeners.get(event)?.forEach((listener) => {
22
      try {
23
        listener(data);
24
      } catch (err) {
25
        console.error(`监听器异常 [${String(event)}]:`, err);
26
      }
27
    });
28
  }
29
}

六、生产验证#

• Node.js —— events.js#L456-L520 EventEmitter.prototype.emit 遍历注册的监听器逐个调用。Node 在 emit 内部用 try-catch 包裹监听器调用，异常转发到 uncaughtException 但不中断后续监听器。Node 还设有 maxListeners 默认阈值（10），超过会打印警告——往往意味着忘了取消订阅导致监听器泄漏。
• Redux —— createStore.ts#L211-L280 subscribe() 添加监听器，dispatch() 执行 reducer 后调用所有监听器。Redux 做了精巧的快照保护：dispatch 前复制监听器数组，通知过程中的 subscribe/unsubscribe 不影响当前轮——新订阅者下一轮才收到通知，取消订阅者当前轮仍会收到。这避免了通知过程中列表被修改导致的跳过或重复调用。Redux 还禁止在监听器内部调用 dispatch，防止递归通知导致状态不一致。
• Chromium —— EventTarget 是 Blink 中 DOM addEventListener 的实现。Chromium 同样做了快照保护：事件分派时先复制监听器列表，增删不影响当前轮次。Chromium 还区分捕获和冒泡阶段——这是观察者模式在 DOM 树上的分层应用，事件从根节点传播到目标节点再传回来，每一层都可以有观察者拦截。

三个项目的共同点值得注意：都做了错误隔离或快照保护。这不是巧合——没有这些保护的观察者实现在生产环境中一定会出问题。

七、小结#

何时使用：

• 事件驱动系统——UI 事件、网络事件、消息队列。浏览器每次点击经过 addEventListener 分发，Node.js 每个连接由 EventEmitter 驱动
• 模块解耦——插件、中间件、扩展点。Webpack 插件系统基于事件钩子（tapable），插件监听编译器生命周期事件来扩展功能
• 状态管理——Redux store、MobX observables、Vue reactivity。Vue 3 响应式系统本质上是精细化的观察者——每个响应式属性维护依赖集合，set 时触发更新
• 日志与监控——发射事件不关心谁在收集。Winston 等日志库内部用事件传递记录，传输层各自监听

何时不用：

• 同步管线——处理顺序和完成很重要时，直接调用函数比回调更可控。比如 parse → validate → transform → save，每步依赖上一步输出，用观察者反而引入不必要的间接性
• 事件风暴——太多事件交织难以调试。高频滚动事件每秒触发几十次，每个观察者再触发更新，性能很快崩溃——应该用 debounce 或 throttle 聚合
• 循环依赖——A 观察 B，B 观察 A，导致无限循环。应重构依赖关系或引入中间状态打破循环
• 强顺序保证——通知顺序不确定，需要严格顺序时用管道或中间件。如果必须保证”A 先于 B”，那 A 和 B 之间有数据依赖，应用显式函数调用

八、参考资料#

• Node.js EventEmitter 文档 - Node.js 事件驱动架构的核心模块
• Redux createStore 源码 - 观察者模式在状态管理中的经典应用
• RxJS 官方文档 - 基于观察者模式的响应式流库
• Observer Pattern - Wikipedia - GoF 设计模式中观察者的经典定义