[译]理解 Node.js 事件驱动架构

原文地址： Understanding Node.js Event-Driven Architecture

大部分 Node 模块，例如 http 和 stream，都是基于 EventEmitter 模块实现的，所以它们拥有 触发 和 监听 事件的能力。

const EventEmitter = require('events');
复制代码

事件驱动的世界中，对于大部分 Node.js 函数，通过回调的形式就是最简单的，例如 fs.readFile 。在这个例子中，事件会被触发一次（当 Node 已经准备好去调用回调函数时），并且回调函数将作为事件处理函数。

首先让我们看一下基本形式。

Node，当你准备好的时候 call 我

Node 控制异步事件最初的形式是通过回调函数。那是在很久以前，那时候 Javascript 还没有支持原生的 Promise 和 async/await 特性。

回调函数最初只是你传递到其他函数的函数。因为 Javascript 中，函数是第一类对象，所以才让这种行为成为可能。

回调函数不代表代码就是异步调用的，理解这一点是非常重要的。一个函数调用回调函数时，既可以通过同步，也可以通过异步。

例如，下面的 fileSize 函数接受 cb 作为回调函数，并且可以根据条件，通过异步或同步触发回调。

function fileSize(fileName, cb) {
  if (typeof fileName !== 'string') {
    return cb(new TypeError('argument should be string')); // 同步
  }
  fs.stat(fileName, (err, stats) => {
    if (err) {
      return cb(err); // 异步
    }
    cb(null, stats.size); // 异步
  });
}
复制代码

注意：这是一个可能会导致意料之外错误的坏实践。设计函数时，回调函数调用最好只通过异步，或者只通过同步。

让我们看一个用回调形式编写，典型异步 Node 函数的简单例子：

const readFileAsArray = function(file, cb) {
  fs.readFile(file, function(err, data) {
    if (err) {
      return cb(err);
    }
    const lines = data
      .toString()
      .trim()
      .split('\n');
    cb(null, lines);
  });
};
复制代码

readFileAsArray 参数包括一个路径和一个回调函数。该宿主函数读取文件内容，并将它们分离到 lines 数组中，并将 lines 传入回调函数中。

下面是一个使用案例。假如在同一目录下，我们有一个文件 numbers.txt ，内容如下：

10
11
12
13
14
15
复制代码

如果需要找出文件内奇数的数量，我们可以使用 readFileAsArray 简化代码：

readFileAsArray('./numbers.txt', (err, lines) => {
  if (err) throw err;
  const numbers = lines.map(Number);
  const oddNumbers = numbers.filter(n => n % 2 === 1);
  console.log('奇数的数量为：', oddNumbers.length);
});
复制代码

上面的代码会读取数字内容并转化为字符串数组，将它们解析为数字，并找出奇数。

这里只用了 Node 的回调函数形式。回调函数第一个参数是 err 错误对象，没有错误时，返回 null 。宿主函数中，回调函数作为最后一个参数传入其中。在你的函数中，你应该总是这么做。也就是将宿主函数的最后一个参数设置为回调函数，并且将回调函数第一个参数设置为错误对象。

现代 Javascript 对于回调函数的替代方式

现代的 Javascript 中，我们拥有 Promise 对象。Promise 成为异步 API 中回调函数的替代方案。在 Promise 中，是通过一个 Promise 对象来单独处理成功和失败的情况，并且允许我们异步链式调用它们。而不是通过传入回调函数作为参数，并且错误处理也不会在同一个地方。

如果函数 readFileAsArray 支持 Promise，我们就可以这样使用：

readFileAsArray('./numbers.txt')
  .then(lines => {
    const numbers = lines.map(Number);
    const oddNumbers = numbers.filter(n => n % 2 === 1);
    console.log('Odd numbers count:', oddNumbers.length);
  })
  .catch(console.error);
复制代码

我们通过在宿主函数的返回值上，调用函数 .then ，而不是传入回调函数。函数 .then 会给我们获取相同行数的数组的途径，就像回调函数版本的一样，并且我们可以像之前一样进行处理。如果想要进行错误处理，我们需要在返回值上调用 .catch 函数，这让我们在错误发生的时候可以进行处理。

因为在现代 Javascript 中有 Promise 对象，所以让宿主函数支持 Promise 接口变得非常容易。下面是 readFileAsArray 函数，在已经拥有回调函数接口的情况下，修改成支持 Promise 接口的例子：

const readFileAsArray = function(file, cb = () => {}) {
  return new Promise((resolve, reject) => {
    fs.readFile(file, function(err, data) {
      if (err) {
        reject(err);
        return cb(err);
      }
      const lines = data
        .toString()
        .trim()
        .split('\n');
      resolve(lines);
      cb(null, lines);
    });
  });
};
复制代码

我们让函数返回了一个包裹 fs.readFile 异步调用的 Promise 对象。这个 Promise 对象暴露了两个参数，分别是 resolve 函数和 reject 函数。

我们可以使用Promise的 reject 方法处理错误时的调用。也可以通过 resolve 函数，处理正常获取数据的调用。

在 Promise 已经被使用的情况下，我们需要做的事情只有为回调函数添加一个默认值。我们可以在参数中使用一个简单，默认的空函数： () => {} 。

通过 async/await 使用 Promise

当需要循环一个异步函数时，添加 Promise 接口让你的代码运行起来更简单。如果使用回调函数，会变得很杂乱。

Promise 让事情变得简单，而 Generator（生成器）让事情变得更简单了。也就是说，更近代的运行异步代码的方式，是通过使用 async 函数，这让我们可以使用同步的方式书写异步代码，也让代码可读性更强。

下面是通过 async/await 的方式，告诉了我们该如何使用 readFileAsArray 函数的例子：

async function countOdd() {
  try {
    const lines = await readFileAsArray('./numbers');
    const numbers = lines.map(Number);
    const oddCount = numbers.filter(n => n % 2 === 1).length;
    console.log('Odd numbers count:', oddCount);
  } catch (err) {
    console.error(err);
  }
}
countOdd();
复制代码

首先，我们创建了一个异步函数，只是比正常函数的前面多了一个 async 字段。在这个异步函数中，我们通过 await 关键字，调用 readFileAsArray 函数，就像这个函数直接返回了行数一样。然后，调用 readFileAsArray 的代码就像同步一样。

我们执行异步函数，让它可以运作。这非常简单并且更具可读性。如果想要进行错误处理，我们需要把异步调用包裹在 try / catch 语句中。

通过 async/await 特性，我们不需要使用一些特殊的 API（例如.then 和.catch）。我们只需要标记函数，并使用原生的 Javascript 代码就可以了。

只要函数支持 Promise 接口，我们就可以使用 async/await 特性。但是，在 async 函数中，我们不能使用回调函数形式的代码（例如 setTimeout）。

EventEmitter 模块

在 Node 中，EventEmitter 是一个可以加快对象之间通信的模块，也是 Node 异步事件驱动架构的核心。许多 Node 内建模块也是继承于 EventEmitter 的。

核心概念非常简单：Emitter 对象触发具名事件，这会导致事先注册了监听器的具名事件被调用。所以，一个 Emitter 对象拥有两个基本特性：

• 触发事件
• 注册和取消注册监听函数

我们只需要创建一个继承于 EventEmitter 的类，就可以让 EventEmitter 起作用了。

class MyEmitter extends EventEmitter {
  //
}
复制代码

Emitter 对象是基于 EventEmitter 类的实例化对象：

const myEmitter = new MyEmitter();
复制代码

在 Emitter 对象生命周期的任何时刻，我们都可以通过使用 emit 函数去触发我们想要的具名事件。

myEmitter.emit('something-happened');
复制代码

触发事件是某些条件发生了的标志。这个条件通常是 Emitter 对象中状态的变化产生的。

我们通过使用方法 on 添加监听函数。每当 Emitter 对象触发相关联的事件时，这些函数将会被调用。

事件 !== 异步

让我们看一个例子：

const EventEmitter = require('events');

class WithLog extends EventEmitter {
  execute(taskFunc) {
    console.log('Before executing');
    this.emit('begin');
    taskFunc();
    this.emit('end');
    console.log('After executing');
  }
}

const withLog = new WithLog();

withLog.on('begin', () => console.log('About to execute'));
withLog.on('end', () => console.log('Done with execute'));

withLog.execute(() => console.log('*** Executing task ***'));
复制代码

WithLog 类是一个事件 Emitter。它定义了实例属性 execute 。这个 excute 函数接收一个参数，也就是一个任务函数，并把这个函数包裹在 log 语句中。它在执行前后触发了事件。

为了能够看到执行的先后顺序，我们注册了两个事件，并通过一个任务去触发它们。

下面代码的输出结果：

Before executing
About to execute
*** Executing task ***
Done with execute
After executing
复制代码

关于上面这个输出信息，我想让你注意的就是所有代码是 同步 进行的，而不是通过异步。

begin
*** Executing task ***
end

就像老式的回调函数一样，所以千万不要认为事件就意味着代码是同步的或者是异步的。

这个概念很重要，因为如果我们传入一个异步 taskFunc 来进行 execute ，事件触发顺序就不再精确。

我们可以通过 setImmediate 模拟这种情况：

// ...

withLog.execute(() => {
  setImmediate(() => {
    console.log('*** Executing task ***');
  });
});
复制代码

下面是输出结果：

Before executing
About to execute
Done with execute
After executing
*** Executing task ***
复制代码

这样是错误的。如果使用了异步调用，将会在调用了 Done with execute 和 After executing 之后，才会执行这一行代码，这样将不再精确。

为了在异步函数调用完成之后触发事件，我们需要通过基于事件的通信，绑定回调函数（或者是 Promise）。下面这个例子做了示范。

使用事件，而不使用回调的一个好处就是我们可以通过注册多个监听器，对相同信号的事件进行多次响应。如果通过回调完成相同的事情，我们必须在单个回调中写更多的逻辑代码。对于应用程序，事件系统是一个在应用顶级构建功能的极好方式，这也允许我们扩展多个插件。你也可以认为是一个状态变化后，允许我们自定义任务的钩子点。

异步事件

让我们把刚才同步的例子转化为异步，这样可以让代码更实用一些。

const fs = require('fs');
const EventEmitter = require('events');

class WithTime extends EventEmitter {
  execute(asyncFunc, ...args) {
    this.emit('begin');
    console.time('execute');
    asyncFunc(...args, (err, data) => {
      if (err) {
        return this.emit('error', err);
      }

      this.emit('data', data);
      console.timeEnd('execute');
      this.emit('end');
    });
  }
}

const withTime = new WithTime();

withTime.on('begin', () => console.log('About to execute'));
withTime.on('end', () => console.log('Done with execute'));

withTime.execute(fs.readFile, __filename);
复制代码

WithTime 类执行一个 asyncFunc 函数，并通过使用 console.time 和 console.timeEnd 打印 asyncFunc 运行的时间。它触发了事件执行前后，正确的顺序。并且也使用异步调用常规的标志，去触发 error/data 事件。

我们通过调用异步函数 fs.readFile 测试 withTime 。我们现在可以通过监听 data 事件，而不必使用回调来处理文件数据。

当执行这些代码时，我们如期地获取到了正确顺序，并且获取了代码执行所用的事件，这非常有用：

About to execute
execute: 4.507ms
Done with execute
复制代码

那我们该如何做才能将回调函数和事件触发器结合起来呢？如果 asyncFunc 也支持 Promise，我们可以使用 async/await 特性完成同样的事情：

class WithTime extends EventEmitter {
  async execute(asyncFunc, ...args) {
    this.emit('begin');
    try {
      console.time('execute');
      const data = await asyncFunc(...args);
      this.emit('data', data);
      console.timeEnd('execute');
      this.emit('end');
    } catch (err) {
      this.emit('error', err);
    }
  }
}
复制代码

总之，这种方式的代码对我来说比回调函数和.then/.catch 的方式更具可读性。async/await 特性让我们更贴近 Javascript 语言本身，这无疑是一大成功。

事件参数和错误

在上面的例子中，两个事件被触发的时候，都附带了额外的参数。

error 事件触发时，附带了错误对象。

this.emit('error', err);
复制代码

data 事件触发时，附带了 data 数据。

this.emit('data', data);
复制代码

我们可以在具名事件中附带很多参数，所有的这些参数可以在之前注册的监听器函数中访问到。

例如，data 事件可用时，我们注册的监听函数就可以获取到事件触发时传递的参数。这个 data 对象就是 asyncFunc 暴露的。

withTime.on('data', data => {
  // do something with data
});
复制代码

通常 error 事件是比较特殊的一个。在基于回调函数的例子中，如果我们没有设置错误事件的监听器，node 进程将会自动退出。

为了示范，添加了另外一个执行错误参数方法的回调：

class WithTime extends EventEmitter {
  execute(asyncFunc, ...args) {
    console.time('execute');
    asyncFunc(...args, (err, data) => {
      if (err) {
        return this.emit('error', err); // Not Handled
      }

      console.timeEnd('execute');
    });
  }
}

const withTime = new WithTime();

withTime.execute(fs.readFile, ''); // BAD CALL
withTime.execute(fs.readFile, __filename);
复制代码

上面的第一个 execute 调用将会引发错误。node 进程将会崩溃并退出：

events.js:163
      throw er; // Unhandled 'error' event
      ^
Error: ENOENT: no such file or directory, open ''
复制代码

而第二个 execute 调用会因为程序崩溃受到影响，并且永远不会执行。

如果我们注册了一个特殊的 error 事件，node 进程的行为将会改变。例如：

withTime.on('error', err => {
  // do something with err, for example log it somewhere
  console.log(err);
});
复制代码

如果我们像上面这样做，来自第一个 execute 调用的错误将会被报告给事件，从而 node 进程就不会崩溃和退出了。另外一个 execute 调用将会正常执行：

{ Error: ENOENT: no such file or directory, open '' errno: -2, code: 'ENOENT', syscall: 'open', path: '' }
execute: 4.276ms
复制代码

注意现在基于 promise 的 Node 的行为将有所不同，只是会输出一个警告，但是最终将会改变。

UnhandledPromiseRejectionWarning: Unhandled promise rejection (rejection id: 1): Error: ENOENT: no such file or directory, open ''

DeprecationWarning: Unhandled promise rejections are deprecated. In the future, promise rejections that are not handled will terminate the Node.js process with a non-zero exit code.
复制代码

另外一个捕获错误事件的方式是通过注册一个全局 uncaughtException 事件。然而，通过这个事件全局捕获错误不是一个好主意。

避免使用 uncaughtException ，但是如果你必须使用它（比如报告发生了什么或者做清除），你应该让你的程序无论如何都要退出：

process.on('uncaughtException', err => {
  // something went unhandled.
  // Do any cleanup and exit anyway!

  console.error(err); // don't do just that.

  // FORCE exit the process too.
  process.exit(1);
});
复制代码

然而，想象一下，如果许多错误事件在同一个时间触发。这意味着上面的 uncaughtException 监听函数将会触发很多次，这对于一些清除代码可能会发生问题。比如当许多数据库调用发生时，就停止操作。

EventEmitter 模块暴露了一个 once 方法。这个方法意味着只会调用监听器一次，而不是每一次事件触发都调用。所以，这是一个 uncaughtException 的实际用例，因为发生了第一个未捕获的异常时，我们将开始清除，而且无论如何进程都将会退出。

监听器的顺序

如果我们在同一个事件上，注册了多个监听器，这些监听器的调用将按照顺序进行。也就是说，第一个注册的监听函数，将会被第一个调用。

// 第一个
withTime.on('data', data => {
  console.log(`Length: ${data.length}`);
});

// 另一个
withTime.on('data', data => {
  console.log(`Characters: ${data.toString().length}`);
});

withTime.execute(fs.readFile, __filename);
复制代码

上面的代码将会先执行 Length 这一行，后执行 Characters 这一行，因为这是按照我们定义监听器的顺序执行的。

如果你需要定义一个新的监听器，但是如果需要将这个监听器设置为第一个被调用，你需要使用 prependListener 方法：

// 第一个
withTime.on('data', data => {
  console.log(`Length: ${data.length}`);
});

// 另一个
withTime.prependListener('data', data => {
  console.log(`Characters: ${data.toString().length}`);
});

withTime.execute(fs.readFile, __filename);
复制代码

这面的代码将会让 Characters 先被打印。

最后，如果你需要删除某一个监听器，你可以使用 removeListener 方法。

这就是本次话题的所有内容。感谢你的阅读！期待下一次！