手写 EventEmitter：深入理解发布订阅模式

引言

在 JavaScript 开发中，事件驱动编程是构建可维护、可扩展应用的核心范式之一。从浏览器 DOM 事件到 Node.js 的异步 I/O，从 Vue 的组件通信到 React 的状态管理，发布订阅模式无处不在。

通过手写一个符合 Node.js EventEmitter 标准的实现，我们不仅能深入理解事件驱动架构的设计原理，还能掌握 JavaScript 中闭包、内存管理、设计模式等核心概念。更重要的是，这是面试中常见的高级题目，能体现你对JavaScript设计模式的理解深度。

本文将带你从零实现一个功能完整的EventEmitter，并探讨其在实际项目中的应用和优化策略。

一、发布订阅模式的核心概念

1.1 什么是发布订阅模式

发布订阅模式（Pub/Sub）是一种消息传递范式，消息的发送者（发布者）不会将消息直接发送给特定的接收者（订阅者），而是通过一个中间件（事件中心）进行通信。

1// 类比：报纸订阅系统
2class NewspaperSystem {
3    constructor() {
4        this.subscribers = new Map(); // 事件中心
5    }
6    
7    // 订阅（读者订阅报纸）
8    subscribe(topic, reader) {
9        if (!this.subscribers.has(topic)) {
10            this.subscribers.set(topic, []);
11        }
12        this.subscribers.get(topic).push(reader);
13    }
14    
15    // 发布（报社发布新闻）
16    publish(topic, news) {
17        if (this.subscribers.has(topic)) {
18            this.subscribers.get(topic).forEach(reader => reader(news));
19        }
20    }
21    
22    // 取消订阅（读者退订）
23    unsubscribe(topic, reader) {
24        if (this.subscribers.has(topic)) {
25            const readers = this.subscribers.get(topic);
26            const index = readers.indexOf(reader);
27            if (index > -1) {
28                readers.splice(index, 1);
29            }
30        }
31    }
32}
33

1.2 核心组件

1. 事件中心(EventEmitter): 存储事件和回调的对应关系
2. 发布者(Publisher): 触发事件，传递数据
3. 订阅者(Subscriber): 监听事件，处理数据

1.3 与观察者模式的对比

二、基础 EventEmitter 实现

2.1 最小可行实现

1class EventEmitter {
2  constructor() {
3    // 存储事件和对应的回调函数
4    this.events = new Map();
5  }
6
7  /**
8   * 订阅事件
9   * @param {string} eventName 事件名称
10   * @param {Function} callback 回调函数
11   * @returns {Function} 取消订阅的函数
12   */
13  on(eventName, callback) {
14    if (!this.events.has(eventName)) {
15      this.events.set(eventName, []);
16    }
17    this.events.get(eventName).push(callback);
18
19    // 返回取消订阅的函数
20    return () => this.off(eventName, callback);
21  }
22
23  /**
24   * 触发事件
25   * @param {string} eventName 事件名称
26   * @param {...any} args 传递给回调函数的参数
27   */
28  emit(eventName, ...args) {
29    if (!this.events.has(eventName)) {
30      return;
31    }
32
33    const callbacks = this.events.get(eventName);
34    callbacks.forEach((callback) => {
35      try {
36        callback.apply(null, args);
37      } catch (error) {
38        console.error(`Error in event listener for ${eventName}:`, error);
39      }
40    });
41  }
42
43  /**
44   * 取消订阅
45   * @param {string} eventName 事件名称
46   * @param {Function} callback 要移除的回调函数
47   */
48  off(eventName, callback) {
49    if (!this.events.has(eventName)) {
50      return;
51    }
52
53    const callbacks = this.events.get(eventName);
54    const index = callbacks.indexOf(callback);
55
56    if (index > -1) {
57      callbacks.splice(index, 1);
58
59      // 如果没有回调函数了, 删除这个事件
60      if (callbacks.length === 0) {
61        this.events.delete(eventName);
62      }
63    }
64  }
65
66  /**
67   * 一次性订阅
68   * @param {string} eventName 事件名称
69   * @param {Function} callback 回调函数
70   */
71  once(eventName, callback) {
72    const onceCallback = (...args) => {
73      callback.apply(null, args);
74      this.off(eventName, callback);
75    };
76
77    this.on(eventName, onceCallback);
78    return () => this.off(eventName, onceCallback);
79  }
80
81  /**
82   * 移除所有事件监听器, 或指定事件的所有监听器
83   * @param {string} eventName 可选, 事件名称
84   */
85  removeAllListeners(eventName) {
86    if (eventName) {
87      this.events.delete(eventName);
88    } else {
89      this.events.clear();
90    }
91  }
92}
93
94// 基础使用示例
95const emitter = new EventEmitter();
96
97// 订阅事件
98const unsubscribe = emitter.on("message", (data) => {
99  console.log("收到消息:", data);
100});
101
102// 触发事件
103emitter.emit("message", "Hello World!"); // 输出: 收到消息: Hello World!
104
105// 取消订阅
106unsubscribe();
107
108// 再次触发，不会有输出
109emitter.emit("message", "Hello Again!");
110

三、完整的 EventEmitter 实现

3.1 支持更多特性的完整实现

1class EventEmitter {
2  constructor() {
3    // 使用 Object.create(null) 避免原型污染
4    this._events = Object.create(null);
5    this._maxListeners = 10;
6  }
7
8  // ================ 核心方法 ================
9
10  /**
11   * 添加事件监听器
12   * @param {string} eventName 事件名称
13   * @param {Function} listener 监听器函数
14   * @returns {EventEmitter} this
15   */
16  on(eventName, listener) {
17    return this._addListener(eventName, listener, false);
18  }
19
20  /**
21   * 添加事件监听器（别名）
22   */
23  addListener(eventName, listener) {
24    return this.on(eventName, listener);
25  }
26
27  /**
28   * 添加一次性事件监听器
29   * @param {string} eventName 事件名称
30   * @param {Function} listener 监听器函数
31   * @returns {EventEmitter} this
32   */
33  once(eventName, listener) {
34    return this._addListener(eventName, listener, true);
35  }
36
37  /**
38   * 触发事件
39   * @param {string} eventName 事件名称
40   * @param {...any} args 传递给监听器的参数
41   * @returns {boolean} 是否有监听器被调用
42   */
43  emit(eventName, ...args) {
44    if (!this._events[eventName]) {
45      // 如果没有 error 事件的监听器，抛出错误
46      if (eventName === "error") {
47        const error = args[0];
48        if (error instanceof Error) {
49          throw error;
50        } else {
51          throw new Error("Unhandled error event");
52        }
53      }
54      return false;
55    }
56
57    const listeners = this._events[eventName];
58    const listenersCopy = listeners.slice(); // 创建副本避免迭代时修改
59
60    let called = false;
61
62    for (const listener of listenersCopy) {
63      try {
64        // 检查是否为 once 包装函数
65        if (listener._once) {
66          // 移除原始监听器
67          this._removeListener(eventName, listener);
68        }
69
70        listener.apply(this, args);
71        called = true;
72      } catch (error) {
73        // 触发错误事件
74        if (eventName !== "error") {
75          this.emit("error", error);
76        }
77      }
78    }
79
80    return called;
81  }
82
83  /**
84   * 移除事件监听器
85   * @param {string} eventName 事件名称
86   * @param {Function} listener 要移除的监听器
87   * @returns {EventEmitter} this
88   */
89  off(eventName, listener) {
90    return this.removeListener(eventName, listener);
91  }
92
93  /**
94   * 移除事件监听器
95   * @param {string} eventName 事件名称
96   * @param {Function} listener 要移除的监听器
97   * @returns {EventEmitter} this
98   */
99  removeListener(eventName, listener) {
100    return this._removeListener(eventName, listener);
101  }
102
103  /**
104   * 移除所有事件监听器
105   * @param {string} [eventName] 可选，事件名称
106   * @returns {EventEmitter} this
107   */
108  removeAllListeners(eventName) {
109    if (eventName) {
110      delete this._events[eventName];
111    } else {
112      this._events = Object.create(null);
113    }
114    return this;
115  }
116
117  // ================ 辅助方法 ================
118
119  /**
120   * 设置最大监听器数量
121   * @param {number} n 最大监听器数量
122   * @returns {EventEmitter} this
123   */
124  setMaxListeners(n) {
125    if (typeof n !== "number" || n < 0) {
126      throw new TypeError("n must be a non-negative number");
127    }
128    this._maxListeners = n;
129    return this;
130  }
131
132  /**
133   * 获取最大监听器数量
134   * @returns {number} 最大监听器数量
135   */
136  getMaxListeners() {
137    return this._maxListeners;
138  }
139
140  /**
141   * 获取指定事件的监听器数量
142   * @param {string} eventName 事件名称
143   * @returns {number} 监听器数量
144   */
145  listenerCount(eventName) {
146    if (!this._events[eventName]) {
147      return 0;
148    }
149    return this._events[eventName].length;
150  }
151
152  /**
153   * 获取所有事件名称
154   * @returns {string[]} 事件名称数组
155   */
156  eventNames() {
157    return Object.keys(this._events);
158  }
159
160  /**
161   * 获取指定事件的所有监听器
162   * @param {string} eventName 事件名称
163   * @returns {Function[]} 监听器数组
164   */
165  listeners(eventName) {
166    if (!this._events[eventName]) {
167      return [];
168    }
169    // 返回副本，避免外部修改内部数组
170    return this._events[eventName].slice();
171  }
172
173  /**
174   * 添加监听器到数组开头
175   * @param {string} eventName 事件名称
176   * @param {Function} listener 监听器函数
177   * @returns {EventEmitter} this
178   */
179  prependListener(eventName, listener) {
180    return this._addListener(eventName, listener, false, true);
181  }
182
183  /**
184   * 添加一次性监听器到数组开头
185   * @param {string} eventName 事件名称
186   * @param {Function} listener 监听器函数
187   * @returns {EventEmitter} this
188   */
189  prependOnceListener(eventName, listener) {
190    return this._addListener(eventName, listener, true, true);
191  }
192
193  /**
194   * 内部方法：添加监听器
195   * @private
196   */
197  _addListener(eventName, listener, once = false, prepend = false) {
198    if (typeof listener !== "function") {
199      throw new TypeError("listener must be a function");
200    }
201
202    // 初始化事件数组
203    if (!this._events[eventName]) {
204      this._events[eventName] = [];
205    }
206
207    const listeners = this._events[eventName];
208
209    // 检查最大监听器限制
210    if (listeners.length >= this._maxListeners && this._maxListeners !== 0) {
211      console.warn(
212        `MaxListenersExceededWarning: Possible EventEmitter memory leak detected. ` +
213          `${listeners.length} ${eventName} listeners added. ` +
214          `Use emitter.setMaxListeners() to increase limit`
215      );
216    }
217
218    // 如果是 once，创建包装函数
219    let listenerToAdd = listener;
220    if (once) {
221      const onceWrapper = (...args) => {
222        listener.apply(this, args);
223        // 标记为 once 包装函数
224        onceWrapper._once = true;
225      };
226      // 保存原始监听器引用，用于移除
227      onceWrapper._originalListener = listener;
228      listenerToAdd = onceWrapper;
229    }
230
231    // 添加到数组开头或结尾
232    if (prepend) {
233      listeners.unshift(listenerToAdd);
234    } else {
235      listeners.push(listenerToAdd);
236    }
237
238    return this;
239  }
240
241  /**
242   * 内部方法：移除监听器
243   * @private
244   */
245  _removeListener(eventName, listener) {
246    if (!this._events[eventName]) {
247      return this;
248    }
249
250    const listeners = this._events[eventName];
251
252    // 查找要移除的监听器
253    // 需要考虑两种情况：
254    // 1. 直接传入监听器
255    // 2. 传入 once 包装函数的原始监听器
256    let index = -1;
257
258    // 尝试直接查找
259    index = listeners.indexOf(listener);
260
261    // 如果没找到，尝试查找原始监听器
262    if (index === -1) {
263      for (let i = 0; i < listeners.length; i++) {
264        const current = listeners[i];
265        if (current._originalListener === listener) {
266          index = i;
267          break;
268        }
269      }
270    }
271
272    if (index > -1) {
273      listeners.splice(index, 1);
274
275      // 如果数组为空，删除事件
276      if (listeners.length === 0) {
277        delete this._events[eventName];
278      }
279    }
280
281    return this;
282  }
283}
284

3.2 类型安全的TypeScript版本

1type Listener = (...args: any[]) => void;
2type OnceWrapper = Listener & { _originalListener?: Listener; _once?: boolean };
3
4class EventEmitter {
5    private _events: Record<string, Listener[]> = Object.create(null);
6    private _maxListeners: number = 10;
7    
8    // 核心方法
9    on(eventName: string, listener: Listener): this {
10        return this._addListener(eventName, listener, false);
11    }
12    
13    addListener(eventName: string, listener: Listener): this {
14        return this.on(eventName, listener);
15    }
16    
17    once(eventName: string, listener: Listener): this {
18        return this._addListener(eventName, listener, true);
19    }
20    
21    emit(eventName: string, ...args: any[]): boolean {
22        const listeners = this._events[eventName];
23        if (!listeners) {
24            if (eventName === 'error') {
25                const error = args[0];
26                throw error instanceof Error ? error : new Error('Unhandled error event');
27            }
28            return false;
29        }
30        
31        const listenersCopy = listeners.slice();
32        let called = false;
33        
34        for (const listener of listenersCopy) {
35            try {
36                // 检查是否为 once 包装函数
37                const onceWrapper = listener as OnceWrapper;
38                if (onceWrapper._once) {
39                    this._removeListener(eventName, listener);
40                }
41                
42                listener.apply(this, args);
43                called = true;
44            } catch (error) {
45                if (eventName !== 'error') {
46                    this.emit('error', error);
47                }
48            }
49        }
50        
51        return called;
52    }
53    
54    off(eventName: string, listener: Listener): this {
55        return this.removeListener(eventName, listener);
56    }
57    
58    removeListener(eventName: string, listener: Listener): this {
59        return this._removeListener(eventName, listener);
60    }
61    
62    removeAllListeners(eventName?: string): this {
63        if (eventName) {
64            delete this._events[eventName];
65        } else {
66            this._events = Object.create(null);
67        }
68        return this;
69    }
70    
71    // 辅助方法
72    setMaxListeners(n: number): this {
73        if (typeof n !== 'number' || n < 0) {
74            throw new TypeError('n must be a non-negative number');
75        }
76        this._maxListeners = n;
77        return this;
78    }
79    
80    getMaxListeners(): number {
81        return this._maxListeners;
82    }
83    
84    listenerCount(eventName: string): number {
85        const listeners = this._events[eventName];
86        return listeners ? listeners.length : 0;
87    }
88    
89    eventNames(): string[] {
90        return Object.keys(this._events);
91    }
92    
93    listeners(eventName: string): Listener[] {
94        const listeners = this._events[eventName];
95        return listeners ? listeners.slice() : [];
96    }
97    
98    prependListener(eventName: string, listener: Listener): this {
99        return this._addListener(eventName, listener, false, true);
100    }
101    
102    prependOnceListener(eventName: string, listener: Listener): this {
103        return this._addListener(eventName, listener, true, true);
104    }
105    
106    // 私有方法
107    private _addListener(eventName: string, listener: Listener, once: boolean, prepend: boolean = false): this {
108        if (typeof listener !== 'function') {
109            throw new TypeError('listener must be a function');
110        }
111        
112        if (!this._events[eventName]) {
113            this._events[eventName] = [];
114        }
115        
116        const listeners = this._events[eventName];
117        
118        // 检查最大监听器限制
119        if (listeners.length >= this._maxListeners && this._maxListeners !== 0) {
120            console.warn(`MaxListenersExceededWarning for event ${eventName}`);
121        }
122        
123        let listenerToAdd: Listener = listener;
124        
125        if (once) {
126            const onceWrapper: OnceWrapper = (...args: any[]) => {
127                listener.apply(this, args);
128                onceWrapper._once = true;
129            };
130            onceWrapper._originalListener = listener;
131            listenerToAdd = onceWrapper;
132        }
133        
134        if (prepend) {
135            listeners.unshift(listenerToAdd);
136        } else {
137            listeners.push(listenerToAdd);
138        }
139        
140        return this;
141    }
142    
143    private _removeListener(eventName: string, listener: Listener): this {
144        const listeners = this._events[eventName];
145        if (!listeners) return this;
146        
147        let index = listeners.indexOf(listener);
148        
149        // 如果没找到，尝试查找原始监听器
150        if (index === -1) {
151            for (let i = 0; i < listeners.length; i++) {
152                const current = listeners[i] as OnceWrapper;
153                if (current._originalListener === listener) {
154                    index = i;
155                    break;
156                }
157            }
158        }
159        
160        if (index > -1) {
161            listeners.splice(index, 1);
162            if (listeners.length === 0) {
163                delete this._events[eventName];
164            }
165        }
166        
167        return this;
168    }
169}
170