Node.js 知名框架 Express Koa 都在使用的 Events 模块你了解吗？

时间 2019-11-05

标签 node.js node 知名框架 express koa 都在使用 events 模块了解栏目 Node.js 繁體版

原文原文链接

人生如同故事。重要的并不在有多长，而是在有多好。——塞涅卡html

在 Node.js 中一个很重要的模块 Events（EventEmitter 事件触发器），也称为发布/订阅模式，为何说它重要，由于在 Node.js 中绝大多数模块都依赖于此，例如 Net、HTTP、FS、Stream 等，除了这些系统模块比较知名的 Express、Koa 框架中也能看到 EventEmitter 的踪影。前端

谈起事件前端的同窗可能会联想到浏览器中的事件，与浏览器中的事件不一样的是它不存在事件冒泡、preventDefault()、stopPropagation() 等方法，EventEmitter 提供了 on()、once()、removeListener() 等方法来对事件进行监听移除。node

做者简介：五月君，Nodejs Developer，慕课网认证做者，热爱技术、喜欢分享的 90 后青年，欢迎关注 Nodejs技术栈和 Github 开源项目 www.nodejs.redgit

经过本文你能学到什么

了解 EventEmitter 是什么？一些基础 API 的使用
在 Node.js 的一些核心模块（Stream、Net）中是如何使用 EventEmitter 的?
主流的 Express/Koa 框架也是基于此实现，咱们如何实现一个基于 EventEmitter 的自定义对象？
高并发场景下雪崩问题如何利用 EventEmitter 特性解决？
事件是否等价于异步？

先从一个简单的例子开始

事件驱动是 Node.js 的核心，怎么体现事件驱动呢？一般一种最多见的形式就是回调，触发一次事件，而后经过回调来接收一些处理，关于这种形式在 JavaScript 编程中家常便饭，例如 fs.readFile(path, callback)、TCP 中的 server.on('data', callback) 等。github

一个简单的实现

主要用到如下两个 API，触发、注册一个监听函数。数据库

emit：触发一个监听函数
on：注册一个监听函数

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

emitter.on("起床", function(time) {
    console.log(`早上 ${time} 开始起床，新的一天加油！`)
    //console.log(`关注公众号Nodejs技术栈，早上 ${time} 点开始起床阅读，从 Node.js 技术栈`);
});

emitter.emit("起床", "6:00");
复制代码

运行程序以后效果以下所示：编程

早上 6:00 开始起床，新的一天加油！
复制代码

除了上面使用 emit、on 方法外还有一些颇有用的 API，你也许须要先去 Node.js 官网（nodejs.cn/api/events.…）作一个了解，那里介绍的很全，在接来的学习中，我会在一些示例中演示一部分的核心 API 如何应用。api

自定义 EventEmitter 类

当你了解了 EventEmitter，你会发现它在 Node.js 中无所不在，Node.js 的核心模块、Express/Koa 等知名框架中，你都会发现它的踪影，例如，下面在 Koa 中 new 一个 app 对象，经过 app.emit() 触发一个事件，实如今整个系统中进行传递。浏览器

const Koa = require('koa');
const app = new Koa();

app.on("koa", function() {
    console.log("在 Koa 中使用 EventEmitter");
});

app.emit("koa");
复制代码

系统模块自定义 EventEmitter 类的实现

在这开始以前让咱们先看下 Node.js 中的 Stream、Net 模块是怎么实现的？缓存

在 Stream 模块中的实现

// https://github.com/nodejs/node/blob/v10.x/lib/internal/streams/legacy.js#L6

const EE = require('events');
const util = require('util');

function Stream() {
  EE.call(this);
}
util.inherits(Stream, EE);
复制代码

在 Net 模块中的实现

// https://github.com/nodejs/node/blob/v10.x/lib/net.js#L1121
const EventEmitter = require('events');
const util = require('util');

function Server(options, connectionListener) {
  if (!(this instanceof Server))
    return new Server(options, connectionListener);

  EventEmitter.call(this);

  ...
}
util.inherits(Server, EventEmitter);
复制代码

观察上面两个 Node.js 模块的自定义 EventEmitter 实现，都有一个共同点使用了 util.inherits(constructor, superConstructor) 方法，这个是 Node.js 中的工具类，这让我想起来了以前在看 JavaScript 权威指南（第 6 章 122 页）中的一个方法 function inherit(p)，意思为经过原型继承建立一个新对象，而 util.inherits 是经过原型复制来实现的对象间的继承。

例如上面的 util.inherits(Server, EventEmitter) 函数，也就是 Server 对象继承了 EventEmitter 在原型中定义的函数，也就拥有了 EventEmitter 事件触发器中的 on、emit 等方法。可是如今 Node.js 官网不建议使用 util.inherits() 方法，而是使用 ES6 中的 class 和 extends 关键词得到语言层面的继承支持，那么在原声 JS 中仍是使用 Object.setPrototypeOf() 来实现的继承，所以在 Node.js 12x 版本中你会看到以下代码实现。

// https://github.com/nodejs/node/blob/v12.x/lib/net.js#L1142
function Server(options, connectionListener) {
  if (!(this instanceof Server))
    return new Server(options, connectionListener);

  EventEmitter.call(this);

  ...
}

// https://github.com/nodejs/node/blob/v12.x/lib/net.js#L1188
Object.setPrototypeOf(Server.prototype, EventEmitter.prototype);
Object.setPrototypeOf(Server, EventEmitter);
复制代码

实现一个基于 EventEmitter 的自定义类

这里用一个例子一天的计划来展现如何基于 EventEmitter 自定义类，在不一样的时间触发相应的事件，经过监听事件来作一些事情。

下面展现了咱们自定义的 OneDayPlan 是如何继承于 EventEmitter

const EventEmitter = require('events');
const oneDayPlanRun = {
    "6:00": function() {
        console.log(`如今是早上 6:00，起床，开始新的一天加油！`);
    },
    "7:00": function() {
        console.log(`如今是早上 7:00，吃早饭！`);
    }
}

function OneDayPlan() {
    EventEmitter.call(this);
}

Object.setPrototypeOf(OneDayPlan.prototype, EventEmitter.prototype);
Object.setPrototypeOf(OneDayPlan, EventEmitter);
复制代码

如今让咱们实例化上面自定义的 OneDayPlan 类，实现事件的触发/监听

const oneDayPlan = new OneDayPlan();

oneDayPlan.on("6:00", function() {
    oneDayPlanRun["6:00"]();
});

oneDayPlan.on("7:00", function() {
    oneDayPlanRun["7:00"]();
});

async function doMain() {
    oneDayPlan.emit("6:00");

    await sleep(2000); // 间隔 2 秒钟输出

    oneDayPlan.emit("7:00");
}

doMain();

async function sleep(s) {
    return new Promise(function(reslve) {
        setTimeout(function() {
            reslve(1);
        }, s);
    });
}
复制代码

EventEmitter 解决高并发下雪崩问题

对于须要查询 DB 的数据，咱们通常称之为热点数据，这类数据一般是要在 DB 之上增长一层缓存，可是在高并发场景下，若是这个缓存正好失效，此时就会有大量的请求直接涌入数据库，对数据库形成必定的压力，对于缓存雪崩的解决方案，网上也不乏有更好的解决方案，可是在 Node.js 中咱们能够利用 events 模块提供的 once() 方法来解决。

once 方法介绍

当触发屡次相同名称事件，经过 once 添加的侦听器只会执行一次，而且在执行以后会接触与它关联的事件，至关于 on 方法和 removeListener 方法的组合，

proxy.once('我很帅', function() {
    console.log('once: 我很帅！');
});

proxy.on('我很帅', function() {
    console.log('on: 我很帅！');
});


proxy.emit('我很帅');
proxy.emit('我很帅');
proxy.emit('我很帅');
复制代码

上面触发了三次 “我很帅” 事件，on 方法乖乖的重复了三次，可是 once 方法说我知道我很帅我只说一次就够了。

once: 我很帅！
on: 我很帅！
on: 我很帅！
on: 我很帅！
复制代码

上面说的 once 方法是 on 和 removeListener 的结合体，在源码中也可看到 github.com/nodejs/node… once 方法接收到信息以后使用 on 方法监听，在 onceWrapper 方法中经过 removeListener 删掉监听函数自身。

function onceWrapper(...args) {
  if (!this.fired) {
    this.target.removeListener(this.type, this.wrapFn);
    this.fired = true;
    return Reflect.apply(this.listener, this.target, args);
  }
}

function _onceWrap(target, type, listener) {
  var state = { fired: false, wrapFn: undefined, target, type, listener };
  var wrapped = onceWrapper.bind(state);
  wrapped.listener = listener;
  state.wrapFn = wrapped;
  return wrapped;
}

EventEmitter.prototype.once = function once(type, listener) {
  checkListener(listener);

  this.on(type, _onceWrap(this, type, listener));
  return this;
};
复制代码

编码实现

利用 once 方法将全部请求的回调都压入事件队列中，对于相同的文件名称查询保证在同一个查询开始到结束的过程当中永远只有一次，若是是 DB 查询也避免了重复数据带来的数据库查询开销。代码编写参考了深刻浅出 Nodejs Events 模块一书，这里使用 fs 进行文件查询，若是是 DB 也同理，另外注意使用 status 键值对形式保存了触发/监听的事件名称和状态，最后建议进行清除，避免引发大对象致使内存泄露问题。

const events = require('events');
const emitter = new events.EventEmitter();
const fs = require('fs');
const status = {};

const select = function(file, filename, cb) {
    emitter.once(file, cb);
    
    if (status[file] === undefined) {
        status[file] = 'ready'; // 不存在设置默认值
    }
    if (status[file] === 'ready') {
        status[file] = 'pending';
        fs.readFile(file, function(err, result) {
            console.log(filename);
            emitter.emit(file, err, result.toString());
            status[file] = 'ready';
            
            setTimeout(function() {
                delete status[file];
            }, 1000);
        });
    }
}

for (let i=1; i<=11; i++) {
    if (i % 2 === 0) {
        select(`/tmp/a.txt`, 'a 文件', function(err, result) {
            console.log('err: ', err, 'result: ', result);
        });
    } else {
        select(`/tmp/b.txt`, 'b 文件', function(err, result) {
            console.log('err: ', err, 'result: ', result);
        });
    }
}
复制代码

控制台运行以上代码进行测试，虽然发起了屡次文件查询请求，fs 模块真正只执行了两次，分别查询了 a、b 两个文件，对于相同的请求，经过利用事件监听器 once 的特性避免了相同条件重复查询。

b 文件
err:  null result:  b
err:  null result:  b
err:  null result:  b
err:  null result:  b
err:  null result:  b
err:  null result:  b
err:  null result:  b
a 文件
err:  null result:  a
err:  null result:  a
err:  null result:  a
err:  null result:  a
err:  null result:  a
复制代码

默认状况下，若是为特定事件添加了超过 10 个监听器，则 EventEmitter 会打印一个警告。可是，并非全部的事件都要限制 10 个监听器。 emitter.setMaxListeners() 方法能够为指定的 EventEmitter 实例修改限制。

(node:88835) Warning: Possible EventEmitter memory leak detected. 11 /tmp/b.txt listeners added. Use emitter.setMaxListeners() to increase limit
(node:88835) Warning: Possible EventEmitter memory leak detected. 11 /tmp/a.txt listeners added. Use emitter.setMaxListeners() to increase limit
复制代码

EventEmitter 循环调用问题

以下代码所示，尝试分析如下两种状况的输出结果

const events = require('events');
const emitter = new events.EventEmitter();
const test = () => console.log('test');

/** 例一 */
emitter.on('test', function() {
    test();
    emitter.emit('test');
})

emitter.emit('test');

/** 例二 */
emitter.on('test', function() {
    test();
    emitter.on('test', test);
})

emitter.emit('test');
复制代码

例一由于在监听函数 on 里执行了 emit 事件触发，会陷入死循环致使栈溢出。

例二结果为只输出一次 test，emitter.on('test', test); 这行代码只是在当前的事件回调中添加了一个事件监听器。

例一：RangeError: Maximum call stack size exceeded
例二：test
复制代码

同步仍是异步

换一个问题事件是否等于异步？答案是不等的，看如下代码示例执行顺序，先输出 111 再输出 222，为何这样？摘自官方 API 的一段话 “EventEmitter 会按照监听器注册的顺序同步地调用全部监听器。因此必须确保事件的排序正确，且避免竞态条件。”

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

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

// 输出
// 111
// 222
复制代码

也可使用 setImmediate() 或 process.nextTick() 切换到异步模式，代码以下所示：

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

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

// 输出
// 222
// 111
复制代码

错误处理

最后一个最重要的错误处理，在 Node.js 中错误处理是一个须要重视的事情，一旦抛出一个错误没有人为处理，可能形成的结果是进程自动退出，以下代码由于事件触发器带有错误信息，而没有相应的错误监听在，会致使进程退出。

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

emitter.emit('error', new Error('This is a error'));
console.log('test');
复制代码

调用后程序崩溃致使 Node 进程自动退出，因受上一行的影响，以后的 console.log('test'); 也不会获得执行。

events.js:167
      throw er; // Unhandled 'error' event
      ^

Error: This is a error
复制代码

做为最佳实践，应该始终为 'error' 事件注册监听器

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

emitter.on('error', function(err) {
    console.error(err);
})

emitter.emit('error', new Error('This is a error'));

console.log('test');
复制代码

Error: This is a error
    at Object.<anonymous> ...
test
复制代码

如上代码所示，第一次调用后错误 error 事件会被监听，Node 进程也不会像以前的程序同样会自动退出，console.log('test'); 也获得了正常运行。

总结

许多 Node.js 成功的模块和框架都是基于 EventEmitter 的，学会 EventEmitter 的使用，而且知道该在何时去使用是很是有用的。

EventEmitter 本质上就是观察者模式的实现，一个相似的模式是发布/订阅，生产者将消息发布以后无需关心订阅者的实现，关注过Nodejs技术栈公众号的同窗，也许你会收到过我以前发布的 RabbitMQ 系列文章，RabbitMQ 自己也是基于 AMQP 协议，这在一个分布式集群环境中使用也是很是好的一种方案。