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

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()is
时进行循环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此处进行语法支持可迭代的语法，该语法支持正确的接口。并且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();
  }

})();