Nodejs源码解析之events

Nodejs中的events模块是很常见的模块，其实现了事件注册，通知等功能，是观察者模式的实现。其使用很简单，实例代码如下：

// 导入events模块
var events = require('events');
// 创建EventEmitter对象
var eventEmitter = new events.EventEmitter();

//回调函数1
var listener1 = function listener1() {
console.log('listener1 executed.');
}

//回调函数2
var listener2 = function listener2() {
console.log('listener2 executed.');
}

// 注册事件的回调函数， 相当于增加观察者
eventEmitter.addListener('connection', listener1);

// 再次注册同样事件的回调函数， 相当于增加观察者
eventEmitter.on('connection', listener2);

// 触发事件，这个时候，会有函数 listener1 和listener2同时被调用
eventEmitter.emit('connection');

上述代码使用是相当简单的，这个是如何实现的的呢？ 其实这个是一般观察者的设计模式的实现逻辑是类似的，都是有一个类似map的结构，存储监听事件和回调函数的对应关系，当监听事件发生时，调用回调函数，这个也不例外。不多说，来看源代码吧。

初始化

// 对象的构造函数
function EventEmitter() {
//这个就是用于调用Init函数
// 也就是说说构造函数，仅仅是调用的下面的Init
EventEmitter.init.call(this);
}
// 导出函数，这个就是一个events模块的总体导出函数
// 在上述的用法中，我们都是需要创建一个EventEmitter对象的
module.exports = EventEmitter;

// 用于兼容node 0.10.x
EventEmitter.EventEmitter = EventEmitter;
// 是否使用domain，默认用法是不使用。Domain其实是EventEmitter子类，
// 一个单独的模块，这里不进一步分析，有兴趣可以看domain.js
EventEmitter.usingDomains = false;
// 同上，用于domain模块
EventEmitter.prototype.domain = undefined;
// 这个就是用于存储事件和回调的类map对象
EventEmitter.prototype._events = undefined;
//
EventEmitter.prototype._maxListeners = undefined;

// By default EventEmitters will print a warning if more than 10 listeners // are added to it. This is a useful default which helps finding memory leaks.
// 默认的最大的观察者的个数，默认为10， 如果超过，会有警告信息，以避免内存泄漏
EventEmitter.defaultMaxListeners = 10;

// 初始化，构造函数必须调用的部分
EventEmitter.init = function() {

// 下面是对domain的处理，不考虑
this.domain = null;
if (EventEmitter.usingDomains) {
// if there is an active domain, then attach to it.
domain = domain || require('domain');
if (domain.active && !(this instanceof domain.Domain)) {
this.domain = domain.active;
}
}
// 创建了一个_events 的空对象，相当于创建了一个map
if (!this._events || this._events === Object.getPrototypeOf(this)._events)
this._events = {};
// 用于保存当前最大监听数目，后面会用到
this._maxListeners = this._maxListeners || undefined;
};

通过上述的源码分析，可以得出，当我们创建EventEmitter对象时候，其实就是生成了一个_events的空对象，做一些基本的检查。

// 导入events模块
var events = require('events');
// 创建EventEmitter对象
var eventEmitter = new events.EventEmitter();

增/删 监听

首先来查看一下增加监听相关的源代码，这里有三个函数：

addListener: 增加事件监听。

on： addListener的别名，实际上是一样的。

once： 增加事件监听，不同的是，事件被fire一次后，会掉函数不再被执行。

EventEmitter.prototype.addListener = function addListener(type, listener) {
var m;
// 首先查看的是否listener为一个函数，确保是可以被执行的会掉函数
if (!util.isFunction(listener))
throw TypeError('listener must be a function');

//确保_events对象已经被创建
if (!this._events)
this._events = {};

// To avoid recursion in the case that type === "newListener"! Before
// adding it to the listeners, first emit "newListener".
// 从注释上，这个是防止递归调用的，这里只有定义了newListener后，才会发送事件
// newListener，从本函数的代码，可以看出，newListener没有被定义，可以忽视
if (this._events.newListener)
this.emit('newListener', type,
util.isFunction(listener.listener) ?
listener.listener : listener);
// 查看该type（事件）是否存在，
if (!this._events[type])
// Optimize the case of one listener. Don't need the extra array object.
// 如果不存在，直接存入就可以
this._events[type] = listener;
else if (util.isObject(this._events[type]))
// If we've already got an array, just append.
// 如果存在，并且是array，直接push
this._events[type].push(listener);
else
// Adding the second element, need to change to array.
// 如果不是array，生成一个新的array
this._events[type] = [this._events[type], listener];

// Check for listener leak
// 下面的代码就是查看是否监听者超过了最大的数目，这个是关于默认的数目的
if (util.isObject(this._events[type]) && !this._events[type].warned) {
var m;
if (!util.isUndefined(this._maxListeners)) {
m = this._maxListeners;
} else {
m = EventEmitter.defaultMaxListeners;
}
//如果数目过大，直接给出console.error
if (m && m > 0 && this._events[type].length > m) {
this._events[type].warned = true;
console.error('(node) warning: possible EventEmitter memory ' +
'leak detected. %d %s listeners added. ' +
'Use emitter.setMaxListeners() to increase limit.',
this._events[type].length, type);
console.trace();
}
}

return this;
};

// 这里就是说on 和addListener是相互一样的，别名。
EventEmitter.prototype.on = EventEmitter.prototype.addListener;
// 查看once，事件触发一次回调函数，就删除，相当于调用了removeListener
EventEmitter.prototype.once = function once(type, listener) {
// 依然一样是确保参数为函数，可以回调
if (!util.isFunction(listener))
throw TypeError('listener must be a function');
// 回调函数是否被fire了
var fired = false;
// 帮助函数
function g() {
// 删除回调函数，请注意，一旦被fired掉，就删除
this.removeListener(type, g);
// 检查是否被fired过了，我想，这个可能是防止重复增加的case，也就是多次调用了once的  情况
if (!fired) {
fired = true;
// 执行回调函数
listener.apply(this, arguments);
}
}
// 增加一个listener属性为回调函数
g.listener = listener;
// 增加具体的回调函数，该回调函数变成了帮助函数g，而不是listener
this.on(type, g);
// 返回整个对象
return this;
};

下面介绍一下删除回调相关的函数。

removeListener - 删除具体的事件回调

removeAllListeners - 删除具体时间相关的所有的回调

EventEmitter.prototype.removeListener =
function removeListener(type, listener) {
var list, position, length, i;

//依然是检查listener是否为函数
if (!util.isFunction(listener))
throw TypeError('listener must be a function');

// 确保_events是否为空，以及事件存在在对象中
if (!this._events || !this._events[type])
return this;
// 等到回调函数的value
list = this._events[type];
length = list.length;
position = -1;
// 如果当前的value和要删除的回调是相等的，包含once的内容
if (list === listener ||
(util.isFunction(list.listener) && list.listener === listener)) {
// 直接删除
delete this._events[type];
// 发送事件
if (this._events.removeListener)
this.emit('removeListener', type, listener);
} // 如果是队列，直接出来
else if (util.isObject(list)) {
for (i = length; i-- > 0;) {
if (list[i] === listener ||
(list[i].listener && list[i].listener === listener)) {
position = i;
break;
}
}
// 如果查找不到，直接返回
if (position < 0)
return this;
// 如果长度为1，说明可以删除
if (list.length === 1) {
list.length = 0;
delete this._events[type];
} else {
// 直接在数组中删除该回调
list.splice(position, 1);
}
// fire removeListener事件
if (this._events.removeListener)
this.emit('removeListener', type, listener);
}

return this;
};

// 删除与单一事件相关的所有回调函数
EventEmitter.prototype.removeAllListeners =
function removeAllListeners(type) {
var key, listeners;
// 确保_events不为空
if (!this._events)
return this;

// not listening for removeListener, no need to emit
// 查看当前是否有removeListener的监听者，从本模块看，没有赋值，所以一般情况下，都是
// 直接删除回调然后返回
if (!this._events.removeListener) {
if (arguments.length === 0)
this._events = {};
else if (this._events[type])
delete this._events[type];
return this;
}

// emit removeListener for all listeners on all events
// 处理没有参数的情况，在没有参数的情况下，就是删除所有事件的回调
// 相当于清空。
if (arguments.length === 0) {
for (key in this._events) {
// 注意有特殊情况是，不删除removeListener的回调
if (key === 'removeListener') continue;
this.removeAllListeners(key);
}
this.removeAllListeners('removeListener');
this._events = {};
return this;
}
// 处理有具体参数的情况， 找到具体回调函数
listeners = this._events[type];

// 如果回调是单个函数，直接删除就好
if (util.isFunction(listeners)) {
this.removeListener(type, listeners);
} else if (Array.isArray(listeners)) {
// LIFO order
// 处理回调函数是一个数组的情况，从后往前一个一个删除。
while (listeners.length)
this.removeListener(type, listeners[listeners.length - 1]);
}
// 清空事件对于的回调函数对象
delete this._events[type];

return this;
};

事件触发

事件触发是由emit函数来实现的，具体的含义就是发送一个事件，这个会同步的调用回调函数。 具体的源代码解析如下：

EventEmitter.prototype.emit = function emit(type) {
var er, handler, len, args, i, listeners;
// 还是检查当前存储的events队列是否为空
if (!this._events)
this._events = {};

// If there is no 'error' event listener then throw.
// 这里需要对error进行特殊处理，如果没有error事件的监听者，直接会抛出error的错误，
// 所有对error的时间的处理要特别注意。
// 也就是所emit（‘error’）有抛出异常的功能，这个是文档中没有的
if (type === 'error' && !this._events.error) {
er = arguments[1];
// 这个是具体domain的处理
if (this.domain) {
if (!er)
er = new Error('Uncaught, unspecified "error" event.');
er.domainEmitter = this;
er.domain = this.domain;
er.domainThrown = false;
this.domain.emit('error', er);
} else if (er instanceof Error) {
// 直接抛出异常，如果emit一个参数Error的实例
throw er; // Unhandled 'error' event
} else {
抛出生成的异常
throw Error('Uncaught, unspecified "error" event.');
}
return false;
}
// 找到对应事件的回调函数
handler = this._events[type];
// 如果回调函数没有定义，说明不存在，直接返回false
if (util.isUndefined(handler))
return false;
// 如果是domain的使用，直接进入domain内处理
if (this.domain && this !== process)
this.domain.enter();
//处理是单个的函数的情况。
if (util.isFunction(handler)) {
// 检查参数的情况，注意，至少有事件这一个argument。所以先处理1，2，3
switch (arguments.length) {
// fast cases
case 1:
// 当回调函数没有参数时候
handler.call(this);
break;
case 2:
// 当回调函数有一个参数时候
handler.call(this, arguments[1]);
break;
case 3:
// 当回调函数有二个参数时候
handler.call(this, arguments[1], arguments[2]);
break;
// slower
default:
// 当回调函数三个或者以上参数时候，就会做一个copy，然后再调用
// 所以，这里我们可以特别注意的地方是，为了效率考虑，回调函数最好不要用3个或者3个以上的函数参数
len = arguments.length;
args = new Array(len - 1);
for (i = 1; i < len; i++)
args[i - 1] = arguments[i];
handler.apply(this, args);
}
} // 这里处理回调函数为一个数组的情况
else if (util.isObject(handler)) {
// 这里直接生成一个参数的拷贝
len = arguments.length;
args = new Array(len - 1);
for (i = 1; i < len; i++)
args[i - 1] = arguments[i];
// 生成一个回调函数的新队列
listeners = handler.slice();
len = listeners.length;
// 逐个调用回调函数
for (i = 0; i < len; i++)
listeners[i].apply(this, args);
}
// 处理domain 的情况，直接退出
if (this.domain && this !== process)
this.domain.exit();
//返回true，有回调函数处理的情况
return true;
};

监听者状态

源码中还提供了下面两个函数，用于查看当前事件的监听状态：

listeners - 事件的回调函数，单个或者队列

listenerCount - 事件的回调函数的个数

首先看一下listeners的源码

// 具体的实例的原型函数
// 从代码可以看出，listeners返回都是一个数组，可能为空，也可能含有1个或者多个元素
EventEmitter.prototype.listeners = function listeners(type) {
var ret;
// 直接查看是否存在，对应的回调函数存在否
if (!this._events || !this._events[type])
ret = []; // 不存在，直接返回空队列
else if (util.isFunction(this._events[type])) // 如果事件对于的值是一个函数
ret = [this._events[type]]; // 返回只有一个回调函数的数组
else/ 如果事件对于的值是一个队列
ret = this._events[type].slice(); // 直接生成回调函数的数组
return ret; // 返回数组
};

接着看一个listenerCount源代码，

// 这个是一个构造函数的方法，和其他的函数调用有少许的不同。
// 其返回的是个数
EventEmitter.listenerCount = function(emitter, type) {
var ret;
// 和上面的函数一样，也是分三种情况检查

if (!emitter._events || !emitter._events[type])
ret = 0;// 不存在，直接返回个数为0
else if (util.isFunction(emitter._events[type]))
ret = 1;// 如果事件对于的值是一个函数，直接返回1
else
ret = emitter._events[type].length;// 如果事件对于的值是一个数组，直接返回数组长度
return ret;
};

从源码可以看出，使用时候需要注意以下几点：

1.同一事件的监听函数不要太多，默认为10， 过大会报警或者错误

2.同一事件的监听函数最好是1个，否则效率会慢。

3.回调函数的参数最好在2个以内，否则效率会慢。