精华 探索异步迭代器在 Node.js 中的使用
发布于 6 个月前 作者 Q-Angelo 1400 次浏览 来自 分享

上一节讲解了迭代器的使用,如果对迭代器还不够了解的可以在回顾下《从理解到实现轻松掌握 ES6 中的迭代器》,目前在 JavaScript 中还没有被默认设定 [Symbol.asyncIterator] 属性的内建对象,但是在 Node.js 中已有部分核心模块(Stream、Events)和一些第三方 NPM 模块(mongodb)已支持 Symbol.asyncIterator 属性。本文也是探索异步迭代器在 Node.js 中的都有哪些使用场景,欢迎留言探讨。

目录

  • 在 Events 中使用 asyncIterator
    • events.on() 示例 1
    • events.on() 示例 2
    • events.on() 开启一个 Node.js 服务器
    • 解析 Node.js 源码对 events.on 异步迭代器的实现
  • 在 Stream 中使用 asyncIterator
    • 异步迭代器 与 Readable
    • 从 Node.js 源码看 readable 是如何实现的 asyncIterator
    • 异步迭代器与 Writeable
  • 在 MongoDB 中使用 asyncIterator
    • MongoDB 中的 cursor
    • MongoDB 异步迭代器实现源码分析
    • 使用 for await…of 遍历可迭代对象 cursor
    • 传送 cursor 到可写流

在 Events 中使用 asyncIterator

Node.js v12.16.0 中新增了 events.on(emitter, eventName) 方法,返回一个迭代 eventName 事件的异步迭代器。

events.on() 示例 1

如下例所示, for await...of 循环只会输出 Hello 当触发 error 事件时会被 try catch 所捕获。

const { on, EventEmitter } = require('events');

(async () => {
  const ee = new EventEmitter();
  const ite = on(ee, 'foo');

  process.nextTick(() => {
    ee.emit('foo', 'Hello');
    ee.emit('error', new Error('unknown mistake.'))
    ee.emit('foo', 'Node.js');
  });

  try {
    for await (const event of ite) {
      console.log(event); // prints ['Hello']
    }
  } catch (err) {
    console.log(err.message); // unknown mistake.
  }
})();

上述示例,如果 EventEmitter 对象实例 ee 触发了 error 事件,错误信息会被抛出并且退出循环,该实例注册的所有事件侦听器也会一并移除。

events.on() 示例 2

for await…of 内部块的执行是同步的,每次只能处理一个事件,即使你接下来还有会立即执行的事件,也是如此。如果是需要并发执行的则不建议使用,这个原因会在下面解析 events.on() 源码时给出答案

如下所示,虽然事件是按顺序同时触发了两次,但是在内部块模拟了 2s 的延迟,下一次事件的处理也会得到延迟。

const ite = on(ee, 'foo');

process.nextTick(() => {
  ee.emit('foo', 'Hello');
  ee.emit('foo', 'Node.js');
  // ite.return(); // 调用后可以结束 for await...of 的遍历
  // ite.throw() // 迭代器对象抛出一个错误
});

try {
  for await (const event of ite) {
    console.log(event); // prints ['Hello'] ['Node.js']
    await sleep(2000);
  }
} catch (err) {
  console.log(err.message);
}

// Unreachable here
console.log('这里将不会被执行');

上例中最后一句代码是不会执行的,此时的迭代器会一直处于遍历中,虽然上面两个事件 emit 都触发了,但是迭代器并没有终止,什么时候终止呢?也就是当内部出现一些错误或我们手动调用可迭代对象的 return() 或 throw() 方法时迭代器才会终止。

events.on() 开启一个 Node.js 服务器

之前一篇文章《“Hello Node.js” 这一次是你没见过的写法》写过一段使用 events.on() 开启一个 HTTP 服务器的代码,在留言中当时有小伙伴对此提出疑惑,基于本章对异步迭代器在 events.on() 中使用的学习,可以很好的解释。

相关代码如下所示:

import { createServer as server } from 'http';
import { on } from 'events';
const ee = on(server().listen(3000), 'request');
for await (const [{ url }, res] of ee)
  if (url === '/hello')
    res.end('Hello Node.js!');
  else
    res.end('OK!');

以上代码看似新颖,其核心实现就是使用 events.on() 返回 createServer() 对象 request 事件的异步可迭代对象,之后用 for await...of 语句遍历,客户端每一次请求,就相当于做了一次 ee.emit(‘request’, Req, Res)。

由于内部块的执行是同步的,下一次事件处理需要依赖上次事件完成才可以执行,对于一个 HTTP 服务器需要考虑并发的,请不要使用上面这种方式!

解析 Node.js 源码对 events.on 异步迭代器的实现

events 模块直接导出了 on() 方法,这个 on() 方法主要是将异步迭代器与事件的 EventEmitter 类的实例对象做了结合,实现还是很巧妙的,以下对核心源码做下解释,理解之后你完全也可以自己实现一个 events.on()。

  • 行 {1} ObjectSetPrototypeOf 是为对象设置一个新的原型,这个对象包含了 next()、return()、throw() 三个方法。
  • 行 {2} 根据异步可迭代协议,可迭代对象必须要包含一个 Symbol.asyncIterator 属性,该属性是一个无参数的函数,返回可迭代对象本身,也就是下面代码中 SymbolAsyncIterator。
  • 行 {3} 新的原型就是 ObjectSetPrototypeOf 的第二个参数 AsyncIteratorPrototype。
  • 行 {4} eventTargetAgnosticAddListener 是对事件注册监听器,里面还是用的事件触发器对象的 on() 方法 emitter.on(name, listener) 。
  • 行 {5} addErrorHandlerIfEventEmitter 判断事件名如果不等于 ‘error’ 同时注册一个 error 事件的监听器,具体实现同行 {4}。
  • 行 {6} eventHandler() 函数就是上面注册的监听器函数 listener 当有事件触发时执行该监听器函数,与异步迭代器的结合就在这里,当有新事件触发时会从 unconsumedPromises 数组里取出第一个元素执行,如果理解异步迭代器实现标准你会发现 PromiseResolve(createIterResult(args, false)) 就是异步迭代器对象 next() 方法返回值的标准定义。
  • 下面继续看 unconsumedPromises 从何而来。
module.exports = EventEmitter;
module.exports.on = on;

function on(emitter, event) {
  const unconsumedEvents = [];
  const unconsumedPromises = [];
  const iterator = ObjectSetPrototypeOf({ // {1}
    next() { .... },
    return() { ... },
    throw(err) { ... },
    [SymbolAsyncIterator]() { // {2}
      return this;
    }
  }, AsyncIteratorPrototype); // {3}
  eventTargetAgnosticAddListener(emitter, event, eventHandler); // {4}
  if (event !== 'error') {
    addErrorHandlerIfEventEmitter(emitter, errorHandler); // {5}
  }
  return iterator;
              
  function eventHandler(...args) { // {6}
    const promise =  unconsumedPromises.shift();
    if (promise) {
      // 以下等价于 promise.resolve({ value: args, done: false });
      PromiseResolve(createIterResult(args, false));
    } else {
      // for await...of 遍历器内部块的执行是同步的,所以每次只能处理 1 个事件,如果同时触发多个事件,上次事件未完成剩下的事件会被保存至 unconsumedEvents 中,待上次事件完成后,遍历器会自动调用 iterator 对象的 next() 方法,消费所有未处理的事件。
      unconsumedEvents.push(args);
    }
  }
}

function eventTargetAgnosticAddListener(emitter, name, listener, flags) {
  ...
  emitter.on(name, listener);
}

以下是 iterator 对象的 next() 方法实现:

  • 行 {1} 首先消费未读消息
  • 行 {2} 判断如果是发生错误则抛出错误信息,例如 iterator 对象的 throw() 方法被调用后就会对 error 做赋值待下次遍历器调用 next() 此处代码就会被执行。
  • 行 {3} 如果迭代器对象完成,返回的 Promise 对象 done 属性设置为 true,遍历器也就结束了,变量 finished 是由 iterator 对象的 return() 方法被调用之后设置的。
  • 行 {4} 这个是上面提到的 unconsumedPromises 数据来源处,例如当我们执行 for await...of 语句遍历异步迭代器对象时就会自动触发 iterator 对象的 next() 方法,执行到行 {4} 处会创建一个 Promise 对象但是 resolve 并没有被立即执行,而是先存放在 unconsumedPromises 数组中,所以在上面 #events.on() 示例 2# 提到一个问题,for await...of 遍历事件的异步迭代器对象时后面的代码块并不会被执行, 当我们触发一个事件时才会在监听器函数里执行这个 resolve 函数,此时才会被释放,之后 for await...of 遍历器会自动再次执行 next() 方法,然后 new 一个新的 Promise 反复循环,直到事件对象抛出 error 事件或执行 iterator 对象的 return() 方法。
const iterator = ObjectSetPrototypeOf({
  next() {
    // {1} 首先,我们会消费所有未读消息
    const value = unconsumedEvents.shift();
    if (value) {
      return PromiseResolve(createIterResult(value, false));
    }

    // {2} 如果发生一次 error 就会执行 Promise.reject 抛出一个错误,在这个错误发生后也会停止事件监听。
    if (error) {
      const p = PromiseReject(error);
      // Only the first element errors
      error = null;
      return p;
    }

    // {3} 如果迭代器对象完成,Promise.resolve done 设置为 true
    if (finished) {
      return PromiseResolve(createIterResult(undefined, true));
    }

    // {4} 等待直到一个事件发生
    return new Promise(function(resolve, reject) {
      unconsumedPromises.push({ resolve, reject });
    });
  }
  ...
}

在 Stream 中使用 asyncIterator

Node.js Stream 模块的可读流对象在 v10.0.0 版本试验性的支持了 [Symbol.asyncIterator] 属性,可以使用 for await...of 语句遍历可读流对象,在 v11.14.0 版本以上已 LTS 支持。

异步迭代器 与 Readable

借助 fs 模块创建一个可读流对象 readable。

const fs = require('fs');
const readable = fs.createReadStream('./hello.txt', {
  encoding: 'utf-8',
  highWaterMark: 1
});

以往当我们读取一个文件时,需要监听 data 事件,拼接数据,在 end 事件里判断完成,如下所示:

function readText(readable) {
  let data = '';
  return new Promise((resolve, reject) => {
    readable.on('data', chunk => {
      data += chunk;
    })
    readable.on('end', () => {
      resolve(data);
    });
    readable.on('error', err => {
      reject(err);
    });
  })
}

现在通过异步迭代器能以一种更简单的方式实现,如下所示:

async function readText(readable) {
  let data = '';
  for await (const chunk of readable) {
    data += chunk;
  }
  return data;
}

现在我们可以调用 readText 做测试。

(async () => {
  try {
    const res = await readText(readable);
    console.log(res); // Hello Node.js
  } catch (err) {
    console.log(err.message);
  }
})();

使用 for await...of 语句遍历 readable,如果循环中因为 break 或 throw 一个错误而终止,则这个 Stream 也将被销毁

上述示例中 chunk 每次接收的值是根据创建可读流时 highWaterMark 这个属性决定的,为了能清晰的看到效果,在创建 readable 对象时我们指定了 highWaterMark 属性为 1 每次只会读取一个字符。

从 Node.js 源码看 readable 是如何实现的 asyncIterator

与同步的迭代器遍历语句 for...of 类似,用于 asyncIterator 异步迭代器遍历的 for await...of 语句在循环内部会默认调用可迭代对象 readable 的 Symbol.asyncIterator() 方法得到一个异步迭代器对象,之后调用迭代器对象的 next() 方法获取结果。

本文以 Node.js 源码 v14.x 为例来看看源码是如何实现的。 当我们调用 fs.createReadStream() 创建一个可读流对象时,对应的该方法内部会调用 ReadStream 构造函数

// https://github.com/nodejs/node/blob/v14.x/lib/fs.js#L2001
function createReadStream(path, options) {
  lazyLoadStreams();
  return new ReadStream(path, options);
}

其实在 ReadStream 这个构造函数里没有我们要找的,重点是它通过原型的方式继承了 Stream 模块的 Readable 构造函数。

function ReadStream(path, options) {
  ...
  Readable.call(this, options);
}

那么现在我们重点来看看 Readable 这个构造函数的实现。

Readable 原型上定义了 SymbolAsyncIterator 属性,该方法返回了一个由生成器函数创建的迭代器对象。

// for await...of 循环会调用
Readable.prototype[SymbolAsyncIterator] = function() {
  let stream = this;
  ...
  const iter = createAsyncIterator(stream);
  iter.stream = stream;
  return iter;
};

// 声明一个创建异步迭代器对象的生成器函数
async function* createAsyncIterator(stream) {
  let callback = nop;

  function next(resolve) {
    if (this === stream) {
      callback();
      callback = nop;
    } else {
      callback = resolve;
    }
  }

  const state = stream._readableState;

  let error = state.errored;
  let errorEmitted = state.errorEmitted;
  let endEmitted = state.endEmitted;
  let closeEmitted = state.closeEmitted;
 
  // error、end、close 事件控制了什么时候结束迭代器遍历。
  stream
    .on('readable', next)
    .on('error', function(err) {
      error = err;
      errorEmitted = true;
      next.call(this);
    })
    .on('end', function() {
      endEmitted = true;
      next.call(this);
    })
    .on('close', function() {
      closeEmitted = true;
      next.call(this);
    });

  try {
    while (true) {
      // stream.read() 从内部缓冲拉取并返回数据。如果没有可读的数据,则返回 null
      // readable 的 destroy() 方法被调用后 readable.destroyed 为 true,readable 即为下面的 stream 对象
      const chunk = stream.destroyed ? null : stream.read();
      if (chunk !== null) {
        yield chunk; // 这里是关键,根据迭代器协议定义,迭代器对象要返回一个 next() 方法,使用 yield 返回了每一次的值
      } else if (errorEmitted) {
        throw error;
      } else if (endEmitted) {
        break;
      } else if (closeEmitted) {
        break;
      } else {
        await new Promise(next);
      }
    }
  } catch (err) {
    destroyImpl.destroyer(stream, err);
    throw err;
  } finally {
    if (state.autoDestroy || !endEmitted) {
      // TODO(ronag): ERR_PREMATURE_CLOSE?
      destroyImpl.destroyer(stream, null);
    }
  }
}

通过上面源码可以看到可读流的异步迭代器实现使用了生成器函数 Generator yield,那么对于 readable 对象遍历除了 for await...of 遍历之外,其实也是可以直接使用调用生成器函数的 next() 方法也是可以的。

const ret = readable[Symbol.asyncIterator]()
console.log(await ret.next()); // { value: 'H', done: false }
console.log(await ret.next()); // { value: 'e', done: false }

异步迭代器与 Writeable

通过上面讲解,我们知道了如何遍历异步迭代器从 readable 对象获取数据,但是你有没有想过如何将一个异步迭代器对象传送给可写流?正是此处要讲的。

从迭代器中创建可读流

Node.js 流对象提供了一个实用方法 stream.Readable.from(),对于符合 Symbol.asyncIterator 或 Symbol.iterator 协议的可迭代对象(Iterable)会先创建一个可读流对象 readable 之后从迭代器中构建 Node.js 可读流。

以下是 从理解到实现轻松掌握 ES6 中的迭代器 一文中曾讲解过的例子,r1 就是我们创建的可迭代对象。使用 stream.Readable.from() 方法则可以将可迭代对象构造为一个可读流对象 readable。

function Range(start, end) {
  this.id = start;
  this.end = end;
}
Range.prototype[Symbol.asyncIterator] = async function* () {
  while (this.id <= this.end) {
    yield this.id++;
  }
}
const r1 = new Range(0, 3);
const readable = stream.Readable.from(r1);
readable.on('data', chunk => {
  console.log(chunk); // 0 1 2 3
});

传送异步迭代器到可写流

使用 pipeline 可以将一系列的流和生成器函数通过管道一起传送,并在管道完成时获取通知。

使用 util.promisify 将 pipeline 转化为 promise 形式。

const util = require('util');
const pipeline = util.promisify(stream.pipeline); // 转为 promise 形式

(async () => {
  try {
    const readable = stream.Readable.from(r1);
    const writeable = fs.createWriteStream('range.txt');
    await pipeline(
      readable,
      async function* (source) {
        for await (const chunk of source) {
          yield chunk.toString();
        }
      },
      writeable
    );
    console.log('Pipeline 成功');
  } catch (err) {
    console.log(err.message);
  }
})()

在写入数据时,传入的 chunk 需是 String、Buffer、Uint8Array 类型,否则 writeable 对象在写入数据时会报错。由于我们自定义的可迭代对象 r1 里最终返回的值类型为 Number 在这里需要做次转换,管道中间的生成器函数就是将每次接收到的值转为字符串。

在 MongoDB 中使用 asyncIterator

除了上面我们讲解的 Node.js 官方提供的几个模块之外,在 MongoDB 中也是支持异步迭代的,不过介绍这点的点资料很少,MongoDB 是通过一个游标的概念来实现的。

MongoDB 中的 cursor

本处以 Node.js 驱动 mongodb 模块来介绍,当我们调用 db.collection.find() 这个方法返回的是一个 cursor(游标),如果想要访问文档那么我们需要迭代这个游标对象来完成,但是通常我们会直接使用 toArray() 这个方法来完成。

下面让我们通过一段示例来看,现在我们有一个数据库 example,一个集合 books,表里面有两条记录,如下所示:

image.png

查询 books 集合的所有数据,以下代码中定义的 myCursor 变量就是游标对象,它不会自动进行迭代,可以使用游标对象的 hasNext() 方法检测是否还有下一个,如果有则可以使用 next() 方法访问数据。

通过以下日志记录可以看到在第三次调用 hasNext() 时返回了 false,如果此时在调用 next() 就会报错,游标已关闭,也就是已经没有数据可遍历了。

const MongoClient = require('mongodb').MongoClient;
const dbConnectionUrl = 'mongodb://127.0.0.1:27017/example';

(async () => {
  const client = await MongoClient.connect(dbConnectionUrl, { useUnifiedTopology: true });
  const bookColl = client.db('example').collection('books');
  const myCursor = await bookColl.find();
 
  console.log(await myCursor.hasNext()); // true
  console.log((await myCursor.next()).name); // 深入浅出Node.js
  console.log(await myCursor.hasNext()); // true
  console.log((await myCursor.next()).name); // Node.js实战
  console.log(await myCursor.hasNext()); // false
  console.log((await myCursor.next()).name); // MongoError: Cursor is closed
})()

直接调用 next() 也可检测,如果还有值则返回该条记录,否则 next() 方法返回 null。

console.log((await myCursor.next()).name);
console.log((await myCursor.next()).name);
console.log((await myCursor.next()));

MongoDB 异步迭代器实现源码分析

MongoDB 中游标是以 hasNext() 返回 false 或 next() 返回为 null 来判断是否达到游标尾部,与之不同的是在我们的 JavaScript 可迭代协议定义中是要有一个 Symbol.asyncIterator 属性的迭代器对象,且迭代器对象是 { done, value } 的形式。

幸运的是 MongoDB Node.js 驱动已经帮助我们实现了这一功能,通过一段源码来看在 MongoDB 中的实现。

  • find 方法

find 方法返回的是一个可迭代游标对象。

// https://github.com/mongodb/node-mongodb-native/blob/3.6/lib/collection.js#L470

Collection.prototype.find = deprecateOptions(
  {
    name: 'collection.find',
    deprecatedOptions: DEPRECATED_FIND_OPTIONS,
    optionsIndex: 1
  },
  function(query, options, callback) {
    const cursor = this.s.topology.cursor(
      new FindOperation(this, this.s.namespace, findCommand, newOptions),
      newOptions
    );

    return cursor;
  }
);
  • CoreCursor

核心实现就在这里,这是一个游标的核心类,MongoDB Node.js 驱动程序中所有游标都是基于此,如果当前支持异步迭代器,则在 CoreCursor 的原型上设置 Symbol.asyncIterator 属性,返回基于 Promise 实现的异步迭代器对象,这符合 JavaScript 中关于异步可迭代对象的标准定义。

// https://github.com/mongodb/node-mongodb-native/blob/3.6/lib/core/cursor.js#L610

if (SUPPORTS.ASYNC_ITERATOR) {
  CoreCursor.prototype[Symbol.asyncIterator] = require('../async/async_iterator').asyncIterator;
}
// https://github.com/mongodb/node-mongodb-native/blob/3.6/lib/async/async_iterator.js#L16

// async function* asyncIterator() {
//   while (true) {
//     const value = await this.next();
//     if (!value) {
//       await this.close();
//       return;
//     }

//     yield value;
//   }
// }

// TODO: change this to the async generator function above
function asyncIterator() {
  const cursor = this;

  return {
    next: function() {
      return Promise.resolve()
        .then(() => cursor.next())
        .then(value => {
          if (!value) {
            return cursor.close().then(() => ({ value, done: true }));
          }
          return { value, done: false };
        });
    }
  };
}

目前是默认使用的 Promise 的形式实现的,上面代码中有段 TODO, Node.js 驱动关于异步迭代实现这块可能后期会改为基于生成器函数的实现,这对我们使用是没变化的.

使用 for await…of 遍历可迭代对象 cursor

还是基于我们上面的示例,如果换成 for await...of 语句遍历就简单的多了。

const myCursor = await bookColl.find();
for await (val of myCursor) {
  console.log(val.name);
}

在 MongoDB 中的聚合管道中使用也是如此,就不再做过多分析了,如下所示:

const myCursor = await bookColl.aggregate();
for await (val of myCursor) {
  console.log(val.name);
}

对于遍历庞大的数据集时,使用游标它会批量加载 MongoDB 中的数据,我们也不必担心一次将所有的数据存在于服务器的内存中,造成内存压力过大。

传送 cursor 到可写流

MongoDB 游标对象本身也是一个可迭代对象(Iterable),结合流模块的 Readable.from() 则可转化为可读流对象,是可以通过流的方式进行写入文件。

但是要注意 MongoDB 中的游标每次返回的是单条文档记录,是一个 Object 类型的,如果直接写入,可写流是会报参数类型错误的,因为可写流默认是一个非对象模式(仅接受 String、Buffer、Unit8Array),所以才会看到在 pipeline 传输的中间又使用了生成器函数,将每次接收的数据块处理为可写流 Buffer 类型。

const myCursor = await bookColl.find();
const readable = stream.Readable.from(myCursor);
await pipeline(
  readable,
  async function* (source) {
    for await (const chunk of source) {
      yield Buffer.from(JSON.stringify(chunk));
    }
  },
  fs.createWriteStream('books.txt')
);

Reference

回到顶部