node中的Event模块(下）

Question

node中的Event模块(下）

SunShinewyf opened this issue 7 years ago · 0 comments

EventEmitter是node中比较核心的模块，除此之外，net.Server、fs.ReadStram、stream也是EventEmitter的实例，可见EventEmitter的核心重要性了

介绍

node中的事件模块是发布/订阅的一种模式，这个模块比前端中的大量DOM事件简单一些，不存在事件冒泡，也不存在preventDefault()、stopPropagation() stopImmediatePropagation()这些控制事件传递的方法。它包含了emit,on,once,addListener等方法。具体的用法可以移步官网

源码解析

node中涉及EventEmitter的代码位于lib/events.js，其中的代码也很简单，主要是构造了一个EventEmitter对象，并且暴露了一些原型方法。源码比较简单，这里只解析一些自己觉得有必要记录的地方。

emit原型方法
emit方法中做了一些参数的初始化以及容错处理，核心部分是根据所传参数个数不同而做的不同处理，代码如下：

  switch (len) {
    // fast cases
    case 1:
      emitNone(handler, isFn, this);
      break;
    case 2:
      emitOne(handler, isFn, this, arguments[1]);
      break;
    case 3:
      emitTwo(handler, isFn, this, arguments[1], arguments[2]);
      break;
    case 4:
      emitThree(handler, isFn, this, arguments[1], arguments[2], arguments[3]);
      break;
    // slower
    default:
      args = new Array(len - 1);
      for (i = 1; i < len; i++)
        args[i - 1] = arguments[i];
      emitMany(handler, isFn, this, args);
  }

代码注释只是说，根据所传不同参数个数有相应的处理，处理起来会使速度变快，但是我个人觉得这种处理方式很傻（闭嘴）。且不论对错，先追踪到emitMany函数看看(emitNone,emitOne,emitTwo,emitThree都长得一样）：

function emitMany(handler, isFn, self, args) {
  if (isFn)
    handler.apply(self, args);
  else {
    var len = handler.length;
    var listeners = arrayClone(handler, len);
    for (var i = 0; i < len; ++i)
      listeners[i].apply(self, args);
  }
}

这个函数是触发on函数中对不同事件类型定义的回调函数，并将emit中传入的参数传入回调函数。这里有一个逻辑分支，就是当listener是函数类型的话，则直接执行，也就是对应下面的简单情形:

 const EventEmitter = require('events');
 let emitter = new EventEmitter();
 
 emitter.on('test',function(){
 	 console.log('aaaa');
 });
 
 emitter.emit('test');

第二个分支刚开始一直想不到是什么情形下触发的，因为在定义on,也就是为事件类型定义监听事件的时候，传入的listener必须是函数类型的，也就是必然会符合isFn为true从而执行第一种逻辑分支。但是当同事对一种事件类型声明多个监听事件时，此时的isFn就是false，这种情形代码如下：

const EventEmitter = require('events');

let emitter = new EventEmitter();

emitter.on('test', () => {
   console.log(1111);
})

emitter.on('test', function test(){
   console.log(2222);
})

emitter.emit('test');

此时的handler如下:

源码中有一行代码比较关键：

var listeners = arrayClone(handler, len);

之所以将handle进行拷贝并且执行，主要是为了防止在触发监听器的时候，原始注册的监听器发生了修改，如下面的情形：

const EventEmitter = require('events');

let emitter = new EventEmitter();

function fun1(){
  emitter.removeListener('test',fun2);
}

function fun2(){
  console.log('uuuu');
}

emitter.on('test',fun1);
emitter.on('test',fun2);

emitter.emit('test');

执行上面这段代码的时候，并不会因为提前删除了fun2而报错。
对于这篇博文里面提到的arrayClone的作用不太认同，里面提出的示例是:

let emitter = new eventEmitter;
emitter.on('message1', function test () {
    // some codes here
    // ...
    emitter.on('message1', test}
});
emitter.emit('message1');

这段代码根本不会执行到arrayClone中去。（在这一块纠结了好久，断点调试发现根本不符合执行条件）

`on`和`addListener`

这两个函数是相同的，用于添加新的监听器。两者都是直接调用的_addListener，源码如下：

function _addListener(target, type, listener, prepend) {
  var m;
  var events;
  var existing;

  if (typeof listener !== 'function')
    throw new TypeError('"listener" argument must be a function');

  events = target._events;
  if (!events) {
    //先判断EventEmitter对象是否存在_events成员函数
    events = target._events = Object.create(null);
    target._eventsCount = 0;
  } else {
  
    if (events.newListener) {
      target.emit('newListener', type,
                  listener.listener ? listener.listener : listener);
      events = target._events;
    }
    existing = events[type];
  }

  if (!existing) {
    // Optimize the case of one listener. Don't need the extra array object.
    existing = events[type] = 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);
    } else {
      existing.push(listener);
    }

    // Check for listener leak
    if (!existing.warned) {
      m = $getMaxListeners(target);
      if (m && m > 0 && existing.length > m) {
        existing.warned = true;
        const w = new Error('Possible EventEmitter memory leak detected. ' +
                            `${existing.length} ${String(type)} listeners ` +
                            'added. Use emitter.setMaxListeners() to ' +
                            'increase limit');
        w.name = 'MaxListenersExceededWarning';
        w.emitter = target;
        w.type = type;
        w.count = existing.length;
        process.emitWarning(w);
      }
    }
  }

  return target;
}

这个函数代码比较简单,在每次添加监听器的时候，都触发newListener，所以如果需要在某个事件类型之前执行一些东西，例如：

const EventEmitter = require('events');

let emitter = new EventEmitter();
emitter.on('newListener', (event, listener) => {
    if (event === 'test') {
        console.log('before test');
    }
});

emitter.on('test', () => {
    console.log('test!');
});

emitter.emit('test');

打印出来的就是:

before test
test!

除此之外，就是对maxListener的一个限定的判断，比较简单，在此不赘述。

once

once用来限制事件监听器只被执行一次，其源码如下:

EventEmitter.prototype.once = function once(type, listener) {
  if (typeof listener !== 'function')
    throw new TypeError('"listener" argument must be a function');
  this.on(type, _onceWrap(this, type, listener));
  return this;
};

通过代码可以看出once调用的on方法，并把_onceWrap作为listener传过去，最后执行的是onceWrapper。源码如下：

function onceWrapper() {
  if (!this.fired) {
    this.target.removeListener(this.type, this.wrapFn);
    this.fired = true;
    switch (arguments.length) {
      case 0:
        return this.listener.call(this.target);
      case 1:
        return this.listener.call(this.target, arguments[0]);
      case 2:
        return this.listener.call(this.target, arguments[0], arguments[1]);
      case 3:
        return this.listener.call(this.target, arguments[0], arguments[1],
                                  arguments[2]);
      default:
        const args = new Array(arguments.length);
        for (var i = 0; i < args.length; ++i)
          args[i] = arguments[i];
        this.listener.apply(this.target, args);
    }
  }
}

这个函数和emit的核心部分是一样的，只是设置了一个fired字段来标记是否是第一次执行，如果是，则对当前事件进行移除并设置fired为true。

值得注意的点

`Eventemitter` 的 `emit` 是同步的

这是为了保证正确的事件排序以及避免资源抢夺和逻辑错误。
执行下面这段代码就可以看出来：

const EventEmitter = require('events');
let emitter = new EventEmitter();

emitter.on('test',function(){
  console.log(222);
});
console.log(111)
emitter.emit('test');
console.log(333)

打印的分别是:111 222 333

同时也可以通过使用setImmediate()或者process.nextTick()方法来实现异步。例如:

const EventEmitter = require('events');

let emitter = new EventEmitter();

emitter.on('async',function(){
  setImmediate(function(){
  		console.log('执行异步方法');
  });
})

emitter.emit('async');

防止死循环调用

如下面代码：

const EventEmitter = require('events');

let emitter = new EventEmitter();

emitter.on('test',function(){
  console.log(222);
  emitter.emit('test');
});
emitter.emit('test');

这个例子会触发死循环调用，不断打印出222。因为在监听回调里面不断执行了emit进行事件的触发，导致不断循环调用。

但是下面这段代码就不会死循环:

const EventEmitter = require('events');

let emitter = new EventEmitter();

emitter.on('test',function test(){
  console.log(222);
  emitter.on('test',test)
});

emitter.emit('test');

因为在emit触发事件回调的时候，此时执行 emitter.on('test',test)这行代码的时候，只是在当前的test这个事件类型中多加了一个事件监听器而已，通过打印test的监听器数量时:

emitter.listenerCount('test')

会打印出2

如何继承 eventEmitter

fs模块继承了eventEmitter模块，具体调用方式如下:

  function FSWatcher(){
  		EventEmitter.call(this);
  }
  util.inherits(FSWatcher, EventEmitter);

调用比较简单

介绍

源码解析

on和addListener

once

值得注意的点

Eventemitter 的 emit 是同步的

防止死循环调用

如何继承 eventEmitter

总结： node中的event模块的源码比较简单，但是一些实现的细节还是值得去深究的，会有很多借鉴的地方

`on`和`addListener`

`Eventemitter` 的 `emit` 是同步的

总结： `node`中的`event`模块的源码比较简单，但是一些实现的细节还是值得去深究的，会有很多借鉴的地方