NodeJs源码学习之events
-
基础
- 重要性:许多核心模块都是基于Events来实现的。例如Stream是基于Events实现,而net和http等是基于Stream来实现的
- 注意: 与on相比,once只触发一次回调。移除监听器用removeListener,移除所有的监听器用removeAllListeners。 也可以监听一个事件,从而来追踪新的监听器。通过监听error事件,从而管理异常。
- 例子:
const util = require('util'); const EventEmitter = require('events').EventEmitter; const myEmitter = new EventEmitter(); const connection = (id) => { console.log('client id: ' + id); }; myEmitter.once('connection', connection); // 移除监听器removeListener // myEmitter.removeListener('connection', connection); myEmitter.emit('connection', 6); // 6 myEmitter.emit('connection', 7); // 无效 myEmitter.once('newListener', (event, listener) => { if (event === 'event') { myEmitter.on('event', () => { console.log('B'); }); } }); myEmitter.on('event', () => { console.log('A'); }); myEmitter.emit('event'); // B A myEmitter.on('error', (err) => { console.log('有错误'); }); myEmitter.emit('error', new Error('whoops!')); myEmitter.emit('HelloWorld');
-
源码
- 最大监听器数量
- 默认值:10
- 源码:
- 文件路径: /lib/events.js
- 代码:
上面的代码很清晰,也很简单。当用户调用getMaxListeners获取最大监听数时,然后getMaxListeners内部调用_getMaxListeners方法,获取最大监听数。如果用户没有设置最大监听数,那么_maxListeners就为undefined,这时就返回默认的defaultMaxListeners,也就是10.如果用户调用setMaxListeners设置最大监听数,那么内部就将设置的值赋值给_maxListeners,此时调用getMaxListeners就返回用户设置的值。 在获取最大监听数时,使用了Object.defineProperty的数据劫持。let defaultMaxListeners = 10; EventEmitter.prototype._maxListeners = undefined; // Obviously not all Emitters should be limited to 10. This function allows // that to be increased. Set to zero for unlimited. EventEmitter.prototype.setMaxListeners = function setMaxListeners(n) { if (typeof n !== 'number' || n < 0 || NumberIsNaN(n)) { throw new ERR_OUT_OF_RANGE('n', 'a non-negative number', n); } this._maxListeners = n; return this; }; function _getMaxListeners(that) { if (that._maxListeners === undefined) return EventEmitter.defaultMaxListeners; return that._maxListeners; } EventEmitter.prototype.getMaxListeners = function getMaxListeners() { return _getMaxListeners(this); }; ObjectDefineProperty(EventEmitter, 'defaultMaxListeners', { enumerable: true, get: function() { return defaultMaxListeners; }, set: function(arg) { if (typeof arg !== 'number' || arg < 0 || NumberIsNaN(arg)) { throw new ERR_OUT_OF_RANGE('defaultMaxListeners', 'a non-negative number', arg); } defaultMaxListeners = arg; } });
- 实例:
const myEmitter = new EventEmitter(); const connection = (id) => { console.log('client id: ' + id); }; console.log(myEmitter.getMaxListeners()); // 输出10 myEmitter.setMaxListeners(20); console.log(myEmitter.getMaxListeners()); // 输出20
- 事件监听
- addListener和on
function _addListener(target, type, listener, prepend) { // _addListener供addListener、on和prependListener调用 let m; let events; let existing; // 判断是不是一个函数,不是一个回调函数就报错 checkListener(listener); // 获取当前作用域的_events。如果没有init,那么_events就不存在。于是就创建一个空对象并初始化_eventsCount events = target._events; if (events === undefined) { events = target._events = ObjectCreate(null); target._eventsCount = 0; } else { // To avoid recursion in the case that type === "newListener"! Before // adding it to the listeners, first emit "newListener". if (events.newListener !== undefined) { // 触发了newListener事件后 target.emit('newListener', type, listener.listener ? listener.listener : listener); // Re-assign `events` because a newListener handler could have caused the // this._events to be assigned to a new object events = target._events; } // 获取传入的事件 existing = events[type]; } if (existing === undefined) { // 事件不存在 // Optimize the case of one listener. Don't need the extra array object. events[type] = listener; // 不存在,就直接使用listener ++target._eventsCount; } else { if (typeof existing === 'function') { // Adding the second element, need to change to array. existing = events[type] = prepend ? [listener, existing] : [existing, listener]; // If we've already got an array, just append. } else if (prepend) { existing.unshift(listener); // 将listener添加到数组的最前面 } else { existing.push(listener); // 将listener添加到数组的末尾 } // Check for listener leak m = _getMaxListeners(target); // 判断监听数是否超过最大的监听数目。如果超过,就报错并且标记已警告 if (m > 0 && existing.length > m && !existing.warned) { existing.warned = true; // No error code for this since it is a Warning // eslint-disable-next-line no-restricted-syntax const w = new Error('Possible EventEmitter memory leak detected. ' + `${existing.length} ${String(type)} listeners ` + `added to ${inspect(target, { depth: -1 })}. Use ` + 'emitter.setMaxListeners() to increase limit'); w.name = 'MaxListenersExceededWarning'; w.emitter = target; w.type = type; w.count = existing.length; process.emitWarning(w); } } return target; } function checkListener(listener) { if (typeof listener !== 'function') { throw new ERR_INVALID_ARG_TYPE('listener', 'Function', listener); } } EventEmitter.prototype.addListener = function addListener(type, listener) { return _addListener(this, type, listener, false); }; EventEmitter.prototype.on = EventEmitter.prototype.addListener; EventEmitter.prototype.prependListener = function prependListener(type, listener) { return _addListener(this, type, listener, true); };
- 例子:
const EventEmitter = require('events').EventEmitter; const emitter = new EventEmitter(); const OpenTheDoor = (key) => { console.log('Got the key: ' + key); }; const Move = (name) => { console.log(name + ' Moveed yet!'); }; emitter.on('Office', Move); emitter.emit('Office', 'Gold'); // 输出Gold Moveed yet! emitter.addListener('Open', OpenTheDoor); emitter.emit('Open', 'Im Frank,please the door now'); // 输出 Got the key: Im Frank,please the door now emitter.once('newListener', (event, listener) => { if (event === 'HelloWorld') { emitter.on('HelloWorld', () => { console.log('HelloWorld'); }); } }); emitter.on('HelloWorld', () => { console.log('HelloWorld'); }); emitter.emit('HelloWorld');
- 例子:
- once
function _onceWrap(target, type, listener) { const state = { fired: false, wrapFn: undefined, target, type, listener }; // 将state绑定到onceWrapper中 const wrapped = onceWrapper.bind(state); wrapped.listener = listener; state.wrapFn = wrapped; return wrapped; } function onceWrapper() { if (!this.fired) { this.target.removeListener(this.type, this.wrapFn); this.fired = true; if (arguments.length === 0) return this.listener.call(this.target); return this.listener.apply(this.target, arguments); } } function on(emitter, event) { const unconsumedEvents = []; const unconsumedPromises = []; let error = null; let finished = false; const iterator = ObjectSetPrototypeOf({ next() { // First, we consume all unread events const value = unconsumedEvents.shift(); if (value) { return PromiseResolve(createIterResult(value, false)); } // Then we error, if an error happened // This happens one time if at all, because after 'error' // we stop listening if (error) { const p = PromiseReject(error); // Only the first element errors error = null; return p; } // If the iterator is finished, resolve to done if (finished) { return PromiseResolve(createIterResult(undefined, true)); } // Wait until an event happens return new Promise(function(resolve, reject) { unconsumedPromises.push({ resolve, reject }); }); }, return() { emitter.removeListener(event, eventHandler); emitter.removeListener('error', errorHandler); finished = true; for (const promise of unconsumedPromises) { promise.resolve(createIterResult(undefined, true)); } return PromiseResolve(createIterResult(undefined, true)); }, throw(err) { if (!err || !(err instanceof Error)) { throw new ERR_INVALID_ARG_TYPE('EventEmitter.AsyncIterator', 'Error', err); } error = err; emitter.removeListener(event, eventHandler); emitter.removeListener('error', errorHandler); }, [SymbolAsyncIterator]() { return this; } }, AsyncIteratorPrototype); emitter.on(event, eventHandler); emitter.on('error', errorHandler); return iterator; function eventHandler(...args) { const promise = unconsumedPromises.shift(); if (promise) { promise.resolve(createIterResult(args, false)); } else { unconsumedEvents.push(args); } } // 针对error事件,移除响应的事件并返回对应的错误原因 function errorHandler(err) { finished = true; const toError = unconsumedPromises.shift(); if (toError) { toError.reject(err); } else { // The next time we call next() error = err; } iterator.return(); } } EventEmitter.prototype.once = function once(type, listener) { // 首先检查listener是不是一个回调函数。不是的话,就报错 checkListener(listener); this.on(type, _onceWrap(this, type, listener)); return this; }; function createIterResult(value, done) { return { value, done }; }
- addListener和on
- 删除一个回调方法: removeListener
// Emits a 'removeListener' event if and only if the listener was removed. EventEmitter.prototype.removeListener = function removeListener(type, listener) { let originalListener; checkListener(listener); // 当前作用域没有任何事件时,直接退出 const events = this._events; if (events === undefined) return this; const list = events[type]; // 没有type类型的事件,直接退出 if (list === undefined) return this; // 存在listener事件,删除type类型的事件并更新事件的总数 if (list === listener || list.listener === listener) { if (--this._eventsCount === 0) this._events = ObjectCreate(null); else { delete events[type]; // 存在删除事件,直接删除该事件 if (events.removeListener) this.emit('removeListener', type, list.listener || listener); } } else if (typeof list !== 'function') { // 对应事件不是一个回调函数 let position = -1; // 遍历获取当前事件所在的位置并保存当前事件 for (let i = list.length - 1; i >= 0; i--) { if (list[i] === listener || list[i].listener === listener) { originalListener = list[i].listener; position = i; break; } } if (position < 0) return this; if (position === 0) list.shift(); else { if (spliceOne === undefined) spliceOne = require('internal/util').spliceOne; spliceOne(list, position); } if (list.length === 1) events[type] = list[0]; // 删除当前事件 if (events.removeListener !== undefined) this.emit('removeListener', type, originalListener || listener); } return this; };
- 删除多个回调方法: removeAllListeners
EventEmitter.prototype.removeAllListeners = function removeAllListeners(type) { const events = this._events; if (events === undefined) return this; // Not listening for removeListener, no need to emit if (events.removeListener === undefined) { // 如果没有传key去删除事件 if (arguments.length === 0) { this._events = ObjectCreate(null); this._eventsCount = 0; } else if (events[type] !== undefined) { if (--this._eventsCount === 0) this._events = ObjectCreate(null); else // 删除当前事件 delete events[type]; } return this; } // Emit removeListener for all listeners on all events if (arguments.length === 0) { for (const key of ObjectKeys(events)) { if (key === 'removeListener') continue; // 如果是removeListener,不做处理,不删除 this.removeAllListeners(key); } this.removeAllListeners('removeListener'); this._events = ObjectCreate(null); this._eventsCount = 0; return this; } // 如果指定某个事件,就删除该事件下所有的回调函数 const listeners = events[type]; if (typeof listeners === 'function') { // 递归调用 this.removeListener(type, listeners); } else if (listeners !== undefined) { // LIFO order // 如果有多个相同的事件,遍历依次删除 for (let i = listeners.length - 1; i >= 0; i--) { this.removeListener(type, listeners[i]); } } return this; };
- 事件触发: emit
const kErrorMonitor = Symbol('events.errorMonitor'); EventEmitter.prototype.emit = function emit(type, ...args) { let doError = (type === 'error'); const events = this._events; if (events !== undefined) { if (doError && events[kErrorMonitor] !== undefined) this.emit(kErrorMonitor, ...args); doError = (doError && events.error === undefined); } else if (!doError) // 什么事件都没有,即使error事件也没有。直接退出。 return false; // If there is no 'error' event listener then throw. // 当前事件是一个Error事件 if (doError) { let er; if (args.length > 0) er = args[0]; if (er instanceof Error) { try { const capture = {}; // eslint-disable-next-line no-restricted-syntax Error.captureStackTrace(capture, EventEmitter.prototype.emit); ObjectDefineProperty(er, kEnhanceStackBeforeInspector, { value: enhanceStackTrace.bind(this, er, capture), configurable: true }); } catch {} // Note: The comments on the `throw` lines are intentional, they show // up in Node's output if this results in an unhandled exception. throw er; // Unhandled 'error' event } let stringifiedEr; const { inspect } = require('internal/util/inspect'); try { stringifiedEr = inspect(er); } catch { stringifiedEr = er; } // At least give some kind of context to the user const err = new ERR_UNHANDLED_ERROR(stringifiedEr); err.context = er; throw err; // Unhandled 'error' event } // 某个指定的事件 const handler = events[type]; if (handler === undefined) return false; if (typeof handler === 'function') { const result = ReflectApply(handler, this, args); // We check if result is undefined first because that // is the most common case so we do not pay any perf // penalty if (result !== undefined && result !== null) { addCatch(this, result, type, args); } } else { const len = handler.length; // 赋值定值事件下的所有回调函数 const listeners = arrayClone(handler, len); for (let i = 0; i < len; ++i) { const result = ReflectApply(listeners[i], this, args); // We check if result is undefined first because that // is the most common case so we do not pay any perf // penalty. // This code is duplicated because extracting it away // would make it non-inlineable. if (result !== undefined && result !== null) { addCatch(this, result, type, args); } } } return true; }; function addCatch(that, promise, type, args) { if (!that[kCapture]) { return; } // Handle Promises/A+ spec, then could be a getter // that throws on second use. try { const then = promise.then; if (typeof then === 'function') { then.call(promise, undefined, function(err) { // The callback is called with nextTick to avoid a follow-up // rejection from this promise. process.nextTick(emitUnhandledRejectionOrErr, that, err, type, args); }); } } catch (err) { that.emit('error', err); } } function arrayClone(arr, n) { const copy = new Array(n); for (let i = 0; i < n; ++i) copy[i] = arr[i]; return copy; }
- 最大监听器数量