带有异步子任务的异步游标迭代

Question

我想在没有数字键（_id）的 mongoDB 集合上执行迭代。集合只有随机字符串作为_id，并且集合的大小很大，因此使用RAM加载整个文档.toArray()不是一个可行的选择。加上我想对每个元素执行异步任务。的使用.map()或者.each()，.forEach()是因为任务的异步性质的限制。我尝试使用上述方法运行任务，但它当然与异步任务冲突，返回未决的承诺而不是正确的结果。

例子

async function dbanalyze(){

  let cursor = db.collection('randomcollection').find()
  for(;;){
    const el = cursor.hasNext() ? loaded.next() : null;
    if(!cursor) break
    await performAnalyze(cursor) // <---- this doesn't return a document but just a cursor object
  }

}

如何仅使用 mongoDB 集合迭代for()？

Answer 1

该Cursor.hasNext()方法也是“异步的”，因此您也需await要这样做。也一样Cursor.next()。因此，实际的“循环”用法应该是while：

async function dbanalyze(){

  let cursor = db.collection('randomcollection').find()
  while ( await cursor.hasNext() ) {  // will return false when there are no more results
    let doc = await cursor.next();    // actually gets the document
    // do something, possibly async with the current document
  }

}

如评论中所述，最终Cursor.hasNext()将false在游标实际耗尽时返回，并且Cursor.next()实际上是从游标中检索每个值。你可以做其他的结构和break循环的时候hasNext()是false，但它更自然而然地到while。

这些仍然是“异步的”，因此您需要await对每个进行承诺解析，这是您缺少的主要事实。

至于Cursor.map()，那么您可能错过了可以async在提供的函数上用标志标记的点：

 cursor.map( async doc => {                   // We can mark as async
    let newDoc = await someAsyncMethod(doc);  // so you can then await inside
    return newDoc;
 })

但是您实际上仍然希望在某个地方“迭代”，除非您可以使用.pipe()其他输出目的地。

此外，async/await标志还使“再次变得更实用”，因为它的一个常见缺陷是无法简单地处理“内部”异步调用，但是有了这些标志，您现在可以轻松地做到这一点，尽管无可否认，因为您必须使用回调，您可能想将其包装在 Promise 中：Cursor.forEach()

await new Promise((resolve, reject) => 
  cursor.forEach(
    async doc => {                              // marked as async
      let newDoc = await someAsyncMethod(doc);  // so you can then await inside
      // do other things
    },
    err => {
      // await was respected, so we get here when done.
      if (err) reject(err);
      resolve();
    }
  )
);

当然，总是有办法通过回调或普通的 Promise 实现来应用它，但它的“糖”async/await实际上使这看起来更干净。

NodeJS v10.x 和 MongoDB Node 驱动程序 3.1.x 及更高版本

最喜欢的版本AsyncIterator现在在 NodeJS v10 及更高版本中启用。这是一种更简洁的迭代方式

async function dbanalyze(){

  let cursor = db.collection('randomcollection').find()
  for await ( let doc of cursor ) {
    // do something with the current document
  }    
}

其中“在某种程度上”又回到了最初提出的关于使用for循环的问题，因为我们可以在for-await-of此处执行语法，以支持支持正确接口的 iterable。并且Cursor确实支持这个接口。

如果您是好奇心，这里是我前段时间制作的一个清单，用于演示各种游标迭代技术。它甚至包括一个来自生成器函数的异步迭代器的案例：

const Async = require('async'),
      { MongoClient, Cursor } = require('mongodb');

const testLen = 3;
(async function() {

  let db;

  try {
    let client = await MongoClient.connect('mongodb://localhost/');

    let db = client.db('test');
    let collection = db.collection('cursortest');

    await collection.remove();

    await collection.insertMany(
      Array(testLen).fill(1).map((e,i) => ({ i }))
    );

    // Cursor.forEach
    console.log('Cursor.forEach');
    await new Promise((resolve,reject) => {
      collection.find().forEach(
        console.log,
        err => {
          if (err) reject(err);
          resolve();
        }
      );
    });

    // Async.during awaits cursor.hasNext()
    console.log('Async.during');
    await new Promise((resolve,reject) => {

      let cursor = collection.find();

      Async.during(
        (callback) => Async.nextTick(() => cursor.hasNext(callback)),
        (callback) => {
          cursor.next((err,doc) => {
            if (err) callback(err);
            console.log(doc);
            callback();
          })
        },
        (err) => {
          if (err) reject(err);
          resolve();
        }
      );

    });

    // async/await allows while loop
    console.log('async/await while');
    await (async function() {

      let cursor = collection.find();

      while( await cursor.hasNext() ) {
        let doc = await cursor.next();
        console.log(doc);
      }

    })();

    // await event stream
    console.log('Event Stream');
    await new Promise((end,error) => {
      let cursor = collection.find();

      for ( let [k,v] of Object.entries({ end, error, data: console.log }) )
        cursor.on(k,v);
    });

    // Promise recursion
    console.log('Promise recursion');
    await (async function() {

      let cursor = collection.find();

      function iterate(cursor) {
        return cursor.hasNext().then( bool =>
          (bool) ? cursor.next().then( doc => {
            console.log(doc);
            return iterate(cursor);
          }) : Promise.resolve()
        )
      }

      await iterate(cursor);

    })();

    // Uncomment if node is run with async iteration enabled
    // --harmony_async_iteration


    console.log('Generator Async Iterator');
    await (async function() {

      async function* cursorAsyncIterator() {
        let cursor = collection.find();

        while (await cursor.hasNext() ) {
          yield cursor.next();
        }

      }

      for await (let doc of cursorAsyncIterator()) {
        console.log(doc);
      }

    })();


    // This is supported with Node v10.x and the 3.1 Series Driver
    await (async function() {

      for await (let doc of collection.find()) {
        console.log(doc);
      }

    })();

    client.close();

  } catch(e) {
    console.error(e);
  } finally {
    process.exit();
  }

})();