nodejs 并行回调设计模式答案

【问题标题】：nodejs parallel callback design patternnodejs 并行回调设计模式
【发布时间】：2011-08-24 18:37:32
【问题描述】：

我正在努力寻找一种好的模式来执行一堆并行任务。

让我定义一些任务来举例说明。任务a, b, c, d, e, f, g 执行为a(function(er, ra){//task a returned, ra is result})，b 到g 也是如此

还有一些任务需要在完成某个任务后执行，我们称它们为ab, bc, abc, bd, bcd, af, fg，意思是当a和b返回时ab(ra, rb)应该立即执行，而当b和c返回，bc(rb, rc)应该立即执行，如果a，b，c全部返回，abc(ra, rb, rc)应该被执行。

对于最简单的情况，如果只有a 和b，我可以这样做：

(function(cb){
    var count = 2, _ra, _rb;
    function update(){if(--count == 0) cb(null, _ra, _rb)}
    a(function(er, ra){_ra = ra; update()});
    b(function(er, ra){_rb = rb; update()});
})(function(er, ra, rb){
    ab(ra, rb);
});

如您所见，a 和 b 并行执行，当两者都完成后，ab(ra, rb) 执行。

但是我怎样才能为大量并行任务做更多的事情呢？

【问题讨论】：

标签： javascript design-patterns concurrency node.js parallel-processing

【解决方案1】：

你真正想要的是一个延迟模式，虽然像futures。

function defer(f) {
    // create a promise.
    var promise = Futures.promise();
    f(function(err, data) {
        if (err) {
            // break it
            promise.smash(err);
        } else {
            // fulfill it
            promise.fulfill(data);
        }
    });
    return promise;
}
var da = defer(a), db = defer(b), dc = defer(c), dd = defer(d), de = defer(e), df = defer(f), dg = defer(g);

// when a and b are fulfilled then call ab
// ab takes one parameter [ra, rb]
Futures.join(da, db).when(ab);
Futures.join(db, dc).when(bc);
// abc takes one parameter [ra, rb, rc]
Futures.join(da, db, dc).when(abc);
Futures.join(db, dd).when(bd);
Futures.join(db, dc, dd).when(bcd);
Futures.join(da, df).when(af);
// where's e ?
Futures.join(df,dg).when(fg);
Futures.join(da,db,dc,dd,de,df,dg).fail(function() {
    console.log(":(");
});

【讨论】：

就像音乐一样美妙
当 node.js 是单线程在单个进程中运行时，这是否真正并行处理这些任务？还是将任务推到事件队列的末尾，待事件队列无事可做时再处理？
@AntKutschera 没有多线程。它只是说并行运行 a-g（由不同线程中的不同进程处理异步），并且当任何对完成时运行此函数
谁启动那个子进程？期货图书馆？怎么样？
@AntKutschera 我假设a,b,c,d, etc 要么启动子进程，要么通过套接字或流进行异步 IO。 Node 做了最少的提升。期货没有。期货只是说当 da 和 db 都返回时运行 ab。

【解决方案2】：

您应该查看 Step (https://github.com/creationix/step)。它只有一百多行代码，因此您可以根据需要阅读整个内容。

我的首选模式如下所示：


function doABunchOfCrazyAsyncStuff() {
  Step (
    function stepA() {
      a(arg1, arg2, arg3, this); // this is the callback, defined by Step
    }
    ,function stepB(err, data) {
      if(err) throw err; // causes error to percolate to the next step, all the way to the end.  same as calling "this(err, null); return;"
      b(data, arg2, arg3, this);
    }
    ,function stepC(err, data) {
      if(err) throw err;
      c(data, arg2, arg3, this);
    }
    ,function stepDEF(err, data) {
      if(err) throw err;
      d(data, this.parallel());
      e(data, this.parallel());
      f(data, this.parallel());
    }
    ,function stepGGG(err, dataD, dataE, dataF) {
      if(err) throw err;
      var combined = magick(dataD, dataE, dataF);
      var group = this.group();  // group() is how you get Step to merge multiple results into an array
      _.map(combined, function (element) {
        g(element, group()); 
      });
    }
    ,function stepPostprocess(err, results) {
      if(err) throw err;
      var processed = _.map(results, magick);
      return processed; // return is a convenient alternative to calling "this(null, result)"
    }
    ,cb // finally, the callback gets (err, result) from the previous function, and we are done
  );
}

注意事项

我的例子也使用了下划线库，“the tie to match to JQuery's tux”：http://documentcloud.github.com/underscore/
将每个步进函数命名为 stepXXXX 是一个好习惯，这样堆栈跟踪清晰易读。
Step 可让您对串行和并行执行进行强大而优雅的组合。这些模式简单易懂。如果您需要更复杂的内容，例如“当 5 种方法中的 3 种完成时，请转到下一步”，请认真重新思考您的设计。你真的需要这么复杂的模式吗？（也许您正在等待法定人数集）。如此复杂的模式值得拥有自己的功能。

【讨论】：

是组任务是并行的还是串行的
这种异步方法有几个缺点，即如果你想调用一个实用方法，比如使用瀑布流，你需要一个包装器来处理错误。例如：collection.find.bind(...)，您可以在其中对链中的方法使用函数绑定器...这非常有用，但不适用于 step。

【解决方案3】：

尝试查看step 模块和this 文章。

【讨论】：

这应该是一种恭维。解释如何使用 step 是一个答案。
@Raynos 我读了几次才意识到我认为你的意思是“评论”，但不要改变它，因为这很有趣

【解决方案4】：

是的，看看流控制模块，比如 step、chain 或 flow ~ 我认为 underscore.js 中也有类似的东西

【讨论】：

【解决方案5】：

为此设置一个非常简单的障碍：https://github.com/berb/node-barrierpoints

【讨论】：

【解决方案6】：

nimble 是另一个不错的选择。

【讨论】：