手写发布订阅与观察者模式深度解析：从小型 EventEmitter 到全局事件总线

发表于 2026/07/17

作者 独行的风

32 分钟阅读

一句话概括

发布订阅（Pub/Sub）与观察者（Observer）模式是 JavaScript 中两种实现”一对多依赖关系”的设计模式——观察者模式是”主体直接通知观察者”的紧耦合，发布订阅模式通过”事件通道”实现发布者和订阅者的完全解耦，两者是构建事件驱动架构的基石。

背景与意义

设计模式的现实映射

在构建复杂的单页应用时，组件之间的通信是一个永恒的问题：

场景 1：用户在购物车页面修改商品数量，希望：

购物车组件更新总数
结算组件重新计算价格
顶部导航的购物车图标更新角标

如果通过 props 层层传递，耦合会变得极高。

场景 2：当一个全局状态变化时（如用户登录/登出），希望：

用户信息组件更新显示
权限组件重新校验
路由守卫更新

如果每个组件都通过 if (user) { ... } 来响应，导致大量重复逻辑。

发布订阅模式通过一个中心化的事件通道解决了这个问题——任何组件都可以订阅感兴趣的事件，而事件的发布者不需要知道谁会消费它。

观察者模式 vs 发布订阅模式

这两种模式经常被混淆，但它们有本质区别：

特性	观察者模式	发布订阅模式
耦合度	主体（Subject）直接持有观察者列表	发布者和订阅者完全解耦
通信方式	主体直接调用观察者的方法	通过事件通道（Event Channel）中转
消息格式	无统一格式，主体自由决定	统一的事件名 + 数据载荷
实现层级	通常在一个对象内部	通常是全局或模块级别的
代码追踪	易于追踪（观察者注册在主体上）	增加调试困难（所有事件通过通道传递）

概念与定义

发布订阅模式的核心组件

┌─────────────────────────────────────────┐
│             Event Bus / Event Channel    │
│                                         │
│   ┌──────────┐        ┌──────────┐      │
│   │ 发布者    │──事件──>│   事件中心  │      │
│   │ (Publisher)│        │ (调度分发) │      │
│   └──────────┘        └─────┬────┘      │
│                              │           │
│                    ┌─────────┴─────────┐ │
│                    │        │          │  │
│                    ▼        ▼          ▼  │
│                ┌────┐  ┌────┐  ┌────┐    │
│                │ S1 │  │ S2 │  │ S3 │    │
│                │订阅者│  │订阅者│  │订阅者│  │
│                └────┘  └────┘  └────┘    │
└─────────────────────────────────────────┘

角色	职责
Event Channel	中心化的事件调度器，维护事件名 → 回调函数的映射表
Publisher	触发事件（emit），携带事件名和数据
Subscriber	注册回调（on），在事件触发时被执行
Event	事件名（字符串/Symbol）和数据载荷

核心接口

  
interface EventEmitter {
  on(event: string, listener: Function): this;        // 订阅
  once(event: string, listener: Function): this;      // 一次性订阅
  off(event: string, listener: Function): this;       // 取消订阅
  emit(event: string, ...args: any[]): boolean;       // 发布
  removeAllListeners(event?: string): this;           // 移除所有
  listeners(event: string): Function[];               // 获取监听器
  eventNames(): Array<string | Symbol>;               // 事件名列表
}

最小示例

发布订阅模式（EventEmitter）

  
class EventEmitter {
  constructor() {
    // 事件名 → Set<回调函数>
    this._events = new Map();
  }

  // 订阅事件
  on(event, listener) {
    if (!this._events.has(event)) {
      this._events.set(event, new Set());
    }
    this._events.get(event).add(listener);
    return this; // 支持链式调用
  }

  // 一次性订阅
  once(event, listener) {
    const onceWrapper = (...args) => {
      listener(...args);
      this.off(event, onceWrapper);
    };
    onceWrapper._isOnce = true;
    this.on(event, onceWrapper);
    return this;
  }

  // 移除监听器
  off(event, listener) {
    if (!this._events.has(event)) return this;

    const listeners = this._events.get(event);
    listeners.delete(listener);

    if (listeners.size === 0) {
      this._events.delete(event);
    }
    return this;
  }

  // 触发事件
  emit(event, ...args) {
    if (!this._events.has(event)) return false;

    const listeners = this._events.get(event);
    for (const listener of listeners) {
      listener(...args);
    }
    return true;
  }

  // 获取事件的所有监听器
  listeners(event) {
    if (!this._events.has(event)) return [];
    return [...this._events.get(event)];
  }

  // 获取所有已注册的事件名
  eventNames() {
    return [...this._events.keys()];
  }

  // 获取特定事件的监听器数量
  listenerCount(event) {
    return this._events.has(event) ? this._events.get(event).size : 0;
  }

  // 移除所有监听器
  removeAllListeners(event) {
    if (event) {
      this._events.delete(event);
    } else {
      this._events.clear();
    }
    return this;
  }
}

// 使用示例
const bus = new EventEmitter();

// 订阅
const onData = (data) => console.log('收到数据:', data);
bus.on('data', onData);

// 一次性订阅
bus.once('once', (msg) => console.log('一次性:', msg));

// 发布
bus.emit('data', { id: 1, value: 'hello' }); // 收到数据: {id: 1, value: 'hello'}
bus.emit('once', '第一次');                     // 一次性: 第一次
bus.emit('once', '第二次');                     // 无输出（已被移除）

// 取消订阅
bus.off('data', onData);
bus.emit('data', { ignored: true });           // 无输出

观察者模式

  
// 观察者模式：主体（Subject）直接管理观察者
class Subject {
  constructor() {
    this._observers = new Set();
    this._state = null;
  }

  // 添加观察者
  attach(observer) {
    this._observers.add(observer);
  }

  // 移除观察者
  detach(observer) {
    this._observers.delete(observer);
  }

  // 设置状态并通知
  setState(newState) {
    this._state = newState;
    this.notify();
  }

  getState() {
    return this._state;
  }

  // 通知所有观察者
  notify() {
    for (const observer of this._observers) {
      observer.update(this._state);
    }
  }
}

// 观察者接口
class Observer {
  constructor(name) {
    this.name = name;
  }

  update(state) {
    console.log(`[${this.name}] 状态更新:`, state);
  }
}

// 使用
const store = new Subject();
const logger = new Observer('日志');
const renderer = new Observer('渲染');

store.attach(logger);
store.attach(renderer);

store.setState({ count: 1 });
// [日志] 状态更新: { count: 1 }
// [渲染] 状态更新: { count: 1 }

核心知识点拆解

1. 链式调用与返回值

EventEmitter 的 on/off 等方法返回 this，支持链式调用：

  
const bus = new EventEmitter();

bus
  .on('event1', fn1)
  .on('event2', fn2)
  .on('event3', fn3);

这种设计在 Node.js 的 EventEmitter 中也采用，优点是简洁，但缺点是链式调用中一旦某个方法出错很难定位。

2. 一次性监听器（once）的实现

once 的关键是”在回调执行后自动取消订阅”。上面的实现使用了一个包装函数 onceWrapper，在调用原始监听器后立即调用 this.off 移除自己。

更好的实现：保留对原始监听器的引用，方便通过 off(event, originalListener) 取消尚未触发的 once 监听。

  
once(event, listener) {
  const wrapper = (...args) => {
    listener(...args);
    this.off(event, wrapper);
  };
  wrapper._original = listener;  // 保留原监听器引用
  this.on(event, wrapper);
  return this;
}

off(event, listener) {
  if (!this._events.has(event)) return this;

  const listeners = this._events.get(event);

  // 对于 once 包裹的监听器，需要从 wrapper 中查找原始 listener
  for (const l of listeners) {
    if (l === listener || l._original === listener) {
      listeners.delete(l);
      break;
    }
  }

  if (listeners.size === 0) {
    this._events.delete(event);
  }
  return this;
}

3. 通配符事件

高级 EventEmitter 支持通配符命名空间，如 user:created、user:updated，订阅 user:* 可以监听所有 user 事件：

  
class WildcardEventEmitter extends EventEmitter {
  on(event, listener) {
    if (event.includes('*')) {
      this._wildcardListeners = this._wildcardListeners || [];
      const regex = new RegExp(
        '^' + event.replace(/\*/g, '.*') + '$'
      );
      this._wildcardListeners.push({ regex, listener });
      return this;
    }
    return super.on(event, listener);
  }

  emit(event, ...args) {
    // 普通事件
    super.emit(event, ...args);

    // 通配符事件
    if (this._wildcardListeners) {
      for (const { regex, listener } of this._wildcardListeners) {
        if (regex.test(event)) {
          listener(event, ...args);
        }
      }
    }
    return true;
  }
}

// 使用
const bus = new WildcardEventEmitter();
bus.on('user:*', (eventName, data) => {
  console.log(`[用户事件] ${eventName}:`, data);
});

bus.emit('user:create', { id: 1, name: 'Alice' });
// [用户事件] user:create: { id: 1, name: 'Alice' }
bus.emit('user:update', { id: 1, name: 'Bob' });
// [用户事件] user:update: { id: 1, name: 'Bob' }

4. 异步事件处理

实际场景中，监听器可能是异步的（返回 Promise）。需要决定：是 fire-and-forget（不等待）还是串行执行（等待每个完成再下一个）：

  
class AsyncEventEmitter extends EventEmitter {
  /**
   * 异步 emit：等待所有监听器完成
   * @param {string} event
   * @param  {...any} args
   * @returns {Promise<Array>} 所有监听器的返回值数组
   */
  async emitAsync(event, ...args) {
    if (!this._events.has(event)) return [];

    const listeners = [...this._events.get(event)];
    const results = [];

    for (const listener of listeners) {
      try {
        results.push(await Promise.resolve(listener(...args)));
      } catch (err) {
        results.push(Promise.reject(err));
      }
    }

    return results;
  }
}

// 使用
const asyncBus = new AsyncEventEmitter();

asyncBus.on('fetch', async (url) => {
  const resp = await fetch(url);
  return resp.json();
});

asyncBus.on('fetch', async (url) => {
  console.log(`正在请求: ${url}`);
});

await asyncBus.emitAsync('fetch', '/api/data');
// 两个监听器串行执行

5. 事件优先级

在需要控制监听器执行顺序时，可以引入优先级机制：

  
class PriorityEventEmitter extends EventEmitter {
  constructor() {
    super();
    // 使用 Map 存储优先级排序的监听器列表，而非 Set
    this._events = new Map();
  }

  on(event, listener, priority = 0) {
    if (!this._events.has(event)) {
      this._events.set(event, []);
    }

    const listeners = this._events.get(event);
    listeners.push({ listener, priority });

    // 按优先级降序排序
    listeners.sort((a, b) => b.priority - a.priority);

    return this;
  }

  emit(event, ...args) {
    if (!this._events.has(event)) return false;

    const listeners = this._events.get(event);
    for (const { listener } of listeners) {
      listener(...args);
    }
    return true;
  }

  off(event, listener) {
    if (!this._events.has(event)) return this;
    const listeners = this._events.get(event);

    const filtered = listeners.filter(l => l.listener !== listener);
    if (filtered.length === 0) {
      this._events.delete(event);
    } else {
      this._events.set(event, filtered);
    }
    return this;
  }
}

// 使用
const bus = new PriorityEventEmitter();

bus.on('order', () => console.log('发送推送'), 100);       // 高优先级
bus.on('order', () => console.log('更新订单状态'), 50);     // 中等优先级
bus.on('order', () => console.log('记录日志'), 10);         // 低优先级

bus.emit('order');
// 发送推送（最高优先级先执行）
// 更新订单状态
// 记录日志

实战案例：微前端应用中的全局事件总线

在现代微前端架构（qiankun、Module Federation）中，主子应用之间无法通过 props 直接通信，事件总线是标准解决方案。

  
// global-event-bus.ts
interface Listener {
  id: number;
  callback: Function;
  once: boolean;
  priority: number;
}

type EventMap = Map<string, Listener[]>;

class GlobalEventBus {
  private static instance: GlobalEventBus;
  private events: EventMap = new Map();
  private listenerId = 0;
  private maxListeners = 100; // 每个事件的最大监听数

  private constructor() {} // 单例

  static getInstance(): GlobalEventBus {
    if (!GlobalEventBus.instance) {
      GlobalEventBus.instance = new GlobalEventBus();
    }
    return GlobalEventBus.instance;
  }

  // 订阅事件
  on(event: string, callback: Function, priority = 0): () => void {
    if (!this.events.has(event)) {
      this.events.set(event, []);
    }

    const listeners = this.events.get(event)!;
    const id = ++this.listenerId;

    // 检查最大监听数
    if (listeners.length >= this.maxListeners) {
      console.warn(
        `[EventBus] 事件 "${event}" 的监听器已达上限 ${this.maxListeners}，` +
        `可能发生内存泄漏`
      );
    }

    listeners.push({ id, callback, once: false, priority });
    listeners.sort((a, b) => b.priority - a.priority);

    // 返回取消订阅函数
    return () => this.off(event, id);
  }

  // 一次性订阅
  once(event: string, callback: Function, priority = 0): () => void {
    if (!this.events.has(event)) {
      this.events.set(event, []);
    }

    const listeners = this.events.get(event)!;
    const id = ++this.listenerId;
    listeners.push({ id, callback, once: true, priority });
    listeners.sort((a, b) => b.priority - a.priority);

    return () => this.off(event, id);
  }

  // 发布事件
  emit(event: string, payload?: any): void {
    if (!this.events.has(event)) {
      if (event === 'error') {
        console.error('[EventBus] 未捕获的错误事件:', payload);
      }
      return;
    }

    const listeners = this.events.get(event)!;
    const toRemove: number[] = [];

    for (const listener of listeners) {
      try {
        listener.callback(payload);
      } catch (err) {
        console.error(`[EventBus] 事件 "${event}" 的监听器出错:`, err);
        // 发送 error 事件
        this.emit('error', { event, error: err });
      }

      if (listener.once) {
        toRemove.push(listener.id);
      }
    }

    // 清理一次性监听器
    if (toRemove.length > 0) {
      this.events.set(
        event,
        listeners.filter(l => !toRemove.includes(l.id))
      );
    }
  }

  private off(event: string, id: number): void {
    if (!this.events.has(event)) return;
    const listeners = this.events.get(event)!;
    const filtered = listeners.filter(l => l.id !== id);

    if (filtered.length === 0) {
      this.events.delete(event);
    } else {
      this.events.set(event, filtered);
    }
  }

  // 获取事件信息（用于调试）
  getEventInfo(event: string) {
    if (!this.events.has(event)) return null;
    const listeners = this.events.get(event)!;
    return {
      event,
      listenerCount: listeners.length,
      priorities: listeners.map(l => l.priority),
    };
  }

  // 清空所有事件
  clearAll(): void {
    this.events.clear();
  }

  // 获取所有已注册事件
  getRegisteredEvents(): string[] {
    return Array.from(this.events.keys());
  }
}

// 在微前端场景中的使用
// 子应用 A（发布登录事件）
const bus = GlobalEventBus.getInstance();

function login(token: string) {
  // 登录逻辑...
  bus.emit('auth:login', { token, userId: 'u_123', timestamp: Date.now() });
}

// 子应用 B（订阅登录事件）
const unsubscribe = bus.on('auth:login', (payload) => {
  // 更新子应用 B 的用户状态
  updateLocalUserState(payload);
  console.log('[子应用 B] 用户已登录:', payload.userId);
});

// 子应用 C（订阅登录事件，高优先级）
bus.on('auth:login', (payload) => {
  // 权限模块优先执行
  initPermissions(payload.userId);
}, 100); // high priority

// 组件卸载时取消订阅
function componentWillUnmount() {
  unsubscribe(); // 通过返回的取消函数
}

实际应用场景：全局键盘快捷键系统

  
class KeyboardShortcutBus {
  constructor() {
    this.emitter = new EventEmitter();
    this.boundKeydown = (e) => this.handleKeydown(e);
    document.addEventListener('keydown', this.boundKeydown);
  }

  // 注册快捷键
  register(shortcut, handler, description = '') {
    this.emitter.on(`key:${shortcut.toLowerCase()}`, handler);
    return this;
  }

  handleKeydown(event) {
    const parts = [];
    if (event.ctrlKey) parts.push('ctrl');
    if (event.metaKey) parts.push('cmd');
    if (event.shiftKey) parts.push('shift');
    if (event.altKey) parts.push('alt');

    const key = event.key.toLowerCase();
    if (key === 'control' || key === 'shift' || key === 'alt' || key === 'meta') {
      return; // 修饰键单独按下不触发
    }

    parts.push(key);
    const shortcut = parts.join('+');

    const emitted = this.emitter.emit(`key:${shortcut}`, event);

    if (emitted) {
      event.preventDefault(); // 快捷键被消费时阻止默认行为
    }
  }

  destroy() {
    document.removeEventListener('keydown', this.boundKeydown);
    this.emitter.removeAllListeners();
  }
}

// 使用
const shortcuts = new KeyboardShortcutBus();

shortcuts
  .register('ctrl+s', () => saveDocument(), '保存文档')
  .register('ctrl+z', () => undo(), '撤销')
  .register('ctrl+shift+z', () => redo(), '重做')
  .register('ctrl+k', () => openCommandPalette(), '打开命令面板');

底层原理

1. Node.js EventEmitter 源码分析（C++ 层）

Node.js 的 EventEmitter 核心实现在 JavaScript 层的 lib/events.js，但性能关键部分调用 C++ 层。

  
// Node.js lib/events.js 简化版
class EventEmitter {
  constructor() {
    EventEmitter.init.call(this);
  }

  static init() {
    // _events 最初不存在，只有在注册事件时才创建
    // 这样才能最大化性能（大部分事件不会同时使用）
    if (this._events === undefined || this._events === Object.getPrototypeOf(this)._events) {
      this._events = Object.create(null);
      this._eventsCount = 0;
    }
  }

  _addListener(event, listener, prepend) {
    let events;
    let existing;

    events = this._events;
    if (events === undefined) {
      events = this._events = Object.create(null);
      this._eventsCount = 0;
    }

    existing = events[event];

    if (existing === undefined) {
      // 首次注册：直接存储为单个函数（而不是数组）
      // 这是微优化：90% 的事件只有一个监听器
      events[event] = listener;
      ++this._eventsCount;
    } else {
      if (typeof existing === 'function') {
        // 从单个函数升级为数组
        events[event] = prepend
          ? [listener, existing]
          : [existing, listener];
      } else {
        // 已经是数组
        if (prepend) {
          existing.unshift(listener);
        } else {
          existing.push(listener);
        }
      }
    }

    return this;
  }

  emit(type, ...args) {
    const handler = this._events[type];
    if (handler === undefined) return false;

    if (typeof handler === 'function') {
      // 单监听器优化
      ReflectApply(handler, this, args);
    } else {
      const len = handler.length;
      const listeners = arrayClone(handler, len);
      for (let i = 0; i < len; ++i) {
        ReflectApply(listeners[i], this, args);
      }
    }

    return true;
  }
}

关键微优化：

单个监听器不创建数组（减少内存分配和 GC 压力）
使用 Object.create(null) 代替 {}（避免原型链查找）
emit 时 clone 监听器列表（防止 listener 在迭代过程中被修改导致的”同时修改”错误）

2. 内存泄漏与弱引用

EventEmitter 最大的问题是内存泄漏：当一个对象订阅了事件，但没有及时取消订阅，这个对象就不会被 GC 回收。

  
// 典型的内存泄漏场景
class HeavyComponent {
  constructor() {
    // ❌ 订阅全局事件
    globalBus.on('data', (data) => {
      // 这个箭头函数捕获了 `this`（HeavyComponent 实例）
      this.processData(data);
    });
    // 注意：这里没有保存取消订阅函数！
    // 当 HeavyComponent 实例被销毁时，globalBus 仍然持有
    // 对这个箭头函数的引用，而箭头函数又持有 this 引用
    // → HeavyComponent 实例无法被 GC 回收
  }
}

// ✅ 改进方案 1：在销毁时取消订阅
class SafeComponent {
  constructor() {
    this.unsubscribe = globalBus.on('data', this.handleData);
  }

  handleData = (data) => {
    this.processData(data);
  };

  destroy() {
    this.unsubscribe(); // 取消订阅
  }
}

// ✅ 改进方案 2：使用 WeakRef（ES2021+）
// 但 WeakRef 不保证立即回收，谨慎使用
class WeakComponent {
  constructor() {
    const ref = new WeakRef(this);
    globalBus.on('data', (data) => {
      const self = ref.deref();
      if (self) {
        self.processData(data);
      } else {
        // 对象已被 GC，可以安全移除监听器
        // 但需要知道要移除哪个函数...
      }
    });
  }
}

3. 发布订阅在 Vue 响应式系统中的应用

Vue 3 的响应式系统（reactive/ref）本质上就是观察者 + 发布订阅的混合：

  
// Vue 3 响应式简化
class Dep {
  constructor() {
    this.subs = new Set(); // 观察者集合
  }

  depend() {
    if (activeEffect) {
      this.subs.add(activeEffect); // 注册当前正在运行的 Effect
    }
  }

  notify() {
    for (const sub of this.subs) {
      sub(); // 通知所有观察者执行
    }
  }
}

// Vue 的 watch API 就是建立在 EventEmitter 之上的
// watch(source, callback) → 内部创建一个 Effect
//   在 source 变化时 → dep.notify() → 触发 Effect → 执行 callback

高频面试题解析

面试题 1：EventEmitter 中，如果在 emit 的回调中又调用了 on/off，会发生什么？如何解决？

解答：

这是一个”同时修改”问题。当迭代监听器列表时，如果在回调中增加了新监听器或移除了当前监听器，会导致：

移除的监听器应该不再执行，但循环中可能已经被跳过
新添加的监听器可能不应该在当前 emit 中执行

解决方案：emit 时复制监听器列表（如 Node.js 的做法）：

  
emit(event, ...args) {
  const listeners = this._events.get(event);
  if (!listeners) return false;

  // 复制列表，防止迭代时被修改
  const snapshot = [...listeners];

  for (const listener of snapshot) {
    listener(...args);
  }

  // 注意：这仍然不能解决"回调中订阅新事件"的问题
  // 新订阅的事件会在后续 emit 中执行
  return true;
}

在 Node.js 源码中，emit 使用 arrayClone(listeners, len) 创建快照。

面试题 2：观察者模式和发布订阅模式，在实际生产环境中应如何选择？

解答：

场景判断矩阵：

场景特征	推荐模式	理由
单个状态变化影响多种视图	观察者模式	简单直接，主体控制生命周期
跨模块/跨应用的解耦通信	发布订阅	解耦彻底，模块互不知晓
需要事件命名空间/过滤	发布订阅	事件名 + 通配符天然支持
关注点在于”数据变更”	观察者模式	更贴合 MVC/MVVM 的数据流
需要发布-订阅历史回放	发布订阅	可以包装为 ReplaySubject
微服务/SOA 架构	发布订阅	消息队列天然是 Pub/Sub

混合使用：在实际项目中，两者经常混合使用：

Vue 的响应式系统是观察者模式（dep.subs 观察数据变化）
Vue 的 eventBus 和组件 $emit/$on 是发布订阅模式
Redux 中 store.subscribe(listener) 是观察者模式，但 dispatch(action) 是类似发布的行为

面试题 3：实现一个 EventEmitter，要求：

支持 on、once、off、emit
支持通配符 * 订阅所有事件
事件订阅数不能超过 10（超过时打印警告）
emit 时如果某个监听器抛出异常，不能影响其他监听器执行

完整解答：

  
class RobustEventEmitter {
  constructor(options = {}) {
    this._events = new Map();
    this._maxListeners = options.maxListeners || 10;
  }

  on(event, listener) {
    if (!this._events.has(event)) {
      this._events.set(event, []);
    }

    const listeners = this._events.get(event);
    if (listeners.length >= this._maxListeners) {
      console.warn(
        `[EventEmitter] 事件 "${event}" 的监听器数量 (${listeners.length + 1})` +
        `超过最大限制 (${this._maxListeners})，可能存在内存泄漏`
      );
    }

    listeners.push({ listener, once: false });
    return this;
  }

  once(event, listener) {
    if (!this._events.has(event)) {
      this._events.set(event, []);
    }

    this._events.get(event).push({ listener, once: true });
    return this;
  }

  off(event, listener) {
    if (!this._events.has(event)) return this;

    const listeners = this._events.get(event);
    // 支持通过原始 listener 移除 once 包裹的监听器
    const remaining = listeners.filter(
      // 注意：此处假设 once 使用了 wrapper._original = listener
      l => l.listener !== listener &&
           (l.listener._original === undefined || l.listener._original !== listener)
    );

    if (remaining.length === 0) {
      this._events.delete(event);
    } else {
      this._events.set(event, remaining);
    }
    return this;
  }

  emit(event, ...args) {
    // 特殊事件：* 通配符
    this._emitToListeners('*', event, args);

    // 普通事件
    this._emitToListeners(event, event, args);

    return true;
  }

  _emitToListeners(eventKey, originalEvent, args) {
    if (!this._events.has(eventKey)) return;

    const listeners = this._events.get(eventKey);
    // 快照复制，防止迭代时修改
    const snapshot = [...listeners];

    for (const entry of snapshot) {
      try {
        entry.listener(originalEvent, ...args);
      } catch (err) {
        console.error(
          `[EventEmitter] 事件 "${originalEvent}" 的监听器出错:`, err
        );
      }
    }

    // 清理一次性监听器
    const remaining = listeners.filter(l => !l.once);
    if (remaining.length === 0) {
      this._events.delete(eventKey);
    } else {
      this._events.set(eventKey, remaining);
    }
  }

  // 修改 once 支持 off 通过原始 listener 移除
  once(event, listener) {
    const wrapper = (...args) => {
      listener(...args);
      this.off(event, wrapper);
    };
    wrapper._original = listener;
    return this.on(event, wrapper);
  }

  removeAllListeners(event) {
    if (event) {
      this._events.delete(event);
    } else {
      this._events.clear();
    }
    return this;
  }

  listenerCount(event) {
    return this._events.has(event) ? this._events.get(event).length : 0;
  }

  eventNames() {
    return Array.from(this._events.keys());
  }
}

总结与扩展

发布订阅和观察者模式是 JavaScript 事件驱动生态的两个支点。理解它们的区别和实现细节，不仅仅是应付面试，更是在设计松耦合、可维护的系统时的基础能力。

值得进一步探索的方向：

RxJS 中的 Subject：RxJS 将 Observable 和 Observer 统一为 Subject，同时具备发布和订阅能力，并支持运算符链
WebSocket 中的事件分发：WebSocket 收到消息时用 EventEmitter 分发到不同处理器（message type dispatch）
Vue 3 的 mitt 库：一个轻量级的 EventEmitter（仅 200 字节），用于替代 Vue 2 中的 $on/$off
自定义事件的性能：原生 CustomEvent + dispatchEvent 也是一种发布订阅，但开销比纯 JS 实现大

  
// 使用原生 CustomEvent 实现事件通信
const bus = document.createElement('div');

// 发布
function emit(event, detail) {
  bus.dispatchEvent(new CustomEvent(event, { detail }));
}

// 订阅
function on(event, handler) {
  bus.addEventListener(event, (e) => handler(e.detail));
}

掌握这些模式，就是掌握了”在适当层级做解耦”的设计思想——这在大型前端应用中比任何框架知识都更持久。

前端核心, JavaScript

发布订阅 EventEmitter 观察者模式设计模式手写题

本文由作者按照 CC BY 4.0 进行授权