带有异步队列和瀑布的猫鼬

Posted 2023-03-10

【中文标题】带有异步队列和瀑布的猫鼬

【英文标题】：Mongoose with async queue & waterfall

【发布时间】：2013-10-24 02:18:36

【问题描述】：

我的目标是通过 Mongoose 导入大量数据。作为一个新手，我无法通过异步使用各种机制正确设置流控制。很高兴有人能指出适当的解决方案。谢谢。

var async = require('async'),
    mongoose = require('mongoose');

mongoose.connect('mongodb://localhost/test');
var Cat = mongoose.model('Cat',  name: String );

// Imagine this is a huge array with a million items.
var content = ['aaa', 'bbb', 'ccc'];
var queries = [];
content.forEach(function(name) 
  queries.push(function(cb) 
    var obj = new Cat( name: name );
    obj.save(function(err) 
      console.log("SAVED: " + name);
      console.log(err);
    );
    return true;
  );
);

// FAILED: async.parallel adds all content to db, 
// but it would exhaust the resource with too many parallel tasks.
async.parallel(queries, function(err, result) 
  if (err)
    return console.log(err);
  console.log(result);
);

// FAILED: save the first item but not the rest
async.waterfall(queries, function(err, result) 
  if (err)
    return console.log(err);
  console.log(result);
);

// FAILED: same as async.waterfall, async.queue saves the first item only
var q = async.queue(function(name, cb) 
  var obj = new Cat( name: name );
  obj.save(function(err) 
    console.log("SAVED: " + name);
    console.log(err);
  );
)
q.push(content, function (err) 
  console.log('finished processing queue');
);

【参考方案1】：

我认为eachLimit 或eachSeries 最适合您的情况：

var content = ['aaa', 'bbb', 'ccc'];
async.eachLimit(content, 10, function(name, done) 
  var obj = new Cat( name : name );
  obj.save(done);
  // if you want to print some status info, use this instead:
  //
  // obj.save(function(err) 
  //   console.log("SAVED: " + name);
  //   console.log(err);
  //   done(err);
  // );
  //
, function(err) 
  // handle any errors;
);

使用eachLimit，您可以“并行”运行 X 数量的查询（上例中为 10 个）以加快处理速度而不会耗尽资源。 eachSeries 将等待上一次保存，然后再继续下一次，因此一次有效地保存一个对象。

请注意，使用each*，您将不会得到包含（已保存）对象的列表（这有点像一种即发即弃的机制，您对结果不感兴趣，除非出现任何错误）。如果您确实想要最后保存对象的列表，您可以使用等效的map* 函数：mapLimit 和mapSeries。

【讨论】：

感谢@robertklep 的回答。我尝试了您的代码（eachLimit with limit=1），但它仅保存数组的第一项（与队列和瀑布相同）。 eachLimit 是否完成了数组的其余部分？我对queue 和waterfall 的使用有什么问题？

我可能已经想通了。如果done(err) 被跳过，它不会运行下一次迭代。为什么需要它？

@Unmeow 这就是async 可以调节异步调用的方式。它需要通知迭代完成，这就是回调函数的用途。