Javascript 在 node.js 中协调并行执行

Coordinating parallel execution in node.js

提问by Thilo

The event-driven programming model of node.js makes it somewhat tricky to coordinate the program flow.

node.js 的事件驱动编程模型使得协调程序流有些棘手。

Simple sequential execution gets turned into nested callbacks, which is easy enough (though a bit convoluted to write down).

简单的顺序执行变成了嵌套的回调，这很容易（虽然写下来有点复杂）。

But how about parallel execution? Say you have three tasks A,B,C that can run in parallel and when they are done, you want to send their results to task D.

但是并行执行呢？假设您有三个可以并行运行的任务 A、B、C，当它们完成时，您想将它们的结果发送到任务 D。

With a fork/join model this would be

使用 fork/join 模型，这将是

• fork A
• fork B
• fork C
• join A,B,C, run D

• 叉子A
• 前叉 B
• 前叉 C
• 加入A，B，C，运行D

How do I write that in node.js ? Are there any best practices or cookbooks? Do I have to hand-roll a solutionevery time, or is there some library with helpers for this?

我如何在 node.js 中编写它？是否有任何最佳实践或食谱？我是否每次都必须手动滚动解决方案，或者是否有一些带有帮助程序的库？

回答by slebetman

Nothing is truly parallel in node.js since it is single threaded. However, multiple events can be scheduled and run in a sequence you can't determine beforehand. And some things like database access are actually "parallel" in that the database queries themselves are run in separate threads but are re-integrated into the event stream when completed.

node.js 中没有真正并行的，因为它是单线程的。但是，可以按您无法事先确定的顺序安排和运行多个事件。像数据库访问这样的一些事情实际上是“并行的”，因为数据库查询本身在单独的线程中运行，但在完成后重新集成到事件流中。

So, how do you schedule a callback on multiple event handlers? Well, this is one common technique used in animations in browser side javascript: use a variable to track the completion.

那么，如何在多个事件处理程序上安排回调？嗯，这是浏览器端 javascript 动画中使用的一种常用技术：使用变量来跟踪完成。

This sounds like a hack and it is, and it sounds potentially messy leaving a bunch of global variables around doing the tracking and in a lesser language it would be. But in javascript we can use closures:

这听起来像是一个黑客，它确实如此，而且在进行跟踪时留下一堆全局变量听起来可能很混乱，并且使用较少的语言。但是在 javascript 中我们可以使用闭包：

function fork (async_calls, shared_callback) {
  var counter = async_calls.length;
  var callback = function () {
    counter --;
    if (counter == 0) {
      shared_callback()
    }
  }

  for (var i=0;i<async_calls.length;i++) {
    async_calls[i](callback);
  }
}

// usage:
fork([A,B,C],D);

In the example above we keep the code simple by assuming the async and callback functions require no arguments. You can of course modify the code to pass arguments to the async functions and have the callback function accumulate results and pass it to the shared_callback function.

在上面的例子中，我们通过假设异步和回调函数不需要参数来保持代码简单。您当然可以修改代码以将参数传递给异步函数，并让回调函数累积结果并将其传递给 shared_callback 函数。

---

---

Additional answer:

补充回答：

Actually, even as is, that fork()function can already pass arguments to the async functions using a closure:

实际上，即使如此，该fork()函数已经可以使用闭包将参数传递给异步函数：

fork([
  function(callback){ A(1,2,callback) },
  function(callback){ B(1,callback) },
  function(callback){ C(1,2,callback) }
],D);

the only thing left to do is to accumulate the results from A,B,C and pass them on to D.

剩下要做的就是将 A、B、C 的结果累加起来，然后将它们传递给 D。

---

---

Even more additional answer:

更多附加答案：

I couldn't resist. Kept thinking about this during breakfast. Here's an implementation of fork()that accumulates results (usually passed as arguments to the callback function):

我无法抗拒。吃早餐的时候一直在想这个。这是一个fork()累积结果的实现（通常作为参数传递给回调函数）：

function fork (async_calls, shared_callback) {
  var counter = async_calls.length;
  var all_results = [];
  function makeCallback (index) {
    return function () {
      counter --;
      var results = [];
      // we use the arguments object here because some callbacks 
      // in Node pass in multiple arguments as result.
      for (var i=0;i<arguments.length;i++) {
        results.push(arguments[i]);
      }
      all_results[index] = results;
      if (counter == 0) {
        shared_callback(all_results);
      }
    }
  }

  for (var i=0;i<async_calls.length;i++) {
    async_calls[i](makeCallback(i));
  }
}

That was easy enough. This makes fork()fairly general purpose and can be used to synchronize multiple non-homogeneous events.

这很容易。这具有fork()相当通用的用途，可用于同步多个非同类事件。

Example usage in Node.js:

Node.js 中的示例用法：

// Read 3 files in parallel and process them together:

function A (c){ fs.readFile('file1',c) };
function B (c){ fs.readFile('file2',c) };
function C (c){ fs.readFile('file3',c) };
function D (result) {
  file1data = result[0][1];
  file2data = result[1][1];
  file3data = result[2][1];

  // process the files together here
}

fork([A,B,C],D);

---

---

Update

更新

This code was written before the existence of libraries like async.js or the various promise based libraries. I'd like to believe that async.js was inspired by this but I don't have any proof of it. Anyway.. if you're thinking of doing this today take a look at async.js or promises. Just consider the answer above a good explanation/illustration of how things like async.parallel work.

这段代码是在像 async.js 这样的库或各种基于 Promise 的库存在之前编写的。我想相信 async.js 受到了这个启发，但我没有任何证据。不管怎样..如果你今天想这样做，看看 async.js 或 promises。只需考虑上面的答案，就可以很好地解释/说明 async.parallel 之类的事情是如何工作的。

For completeness sake the following is how you'd do it with async.parallel:

为了完整起见，以下是您如何使用async.parallel：

var async = require('async');

async.parallel([A,B,C],D);

Note that async.parallelworks exactly the same as the forkfunction we implemented above. The main difference is it passes an error as the first argument to Dand the callback as the second argument as per node.js convention.

请注意，它的async.parallel工作原理与fork我们上面实现的功能完全相同。主要区别在于，它D按照 node.js 约定将错误作为第一个参数传递给，并将回调作为第二个参数传递。

Using promises, we'd write it as follows:

使用 Promise，我们将其编写如下：

// Assuming A, B & C return a promise instead of accepting a callback

Promise.all([A,B,C]).then(D);

回答by Wes Gamble

I believe that now the "async" module provides this parallel functionality and is roughly the same as the fork function above.

我相信现在“async”模块提供了这种并行功能，并且与上面的 fork 功能大致相同。

回答by Randy

The futuresmodule has a submodule called jointhat I have liked to use:

该期货模块有一个名为子模块加入，我喜欢使用：

Joins asynchronous calls together similar to how pthread_joinworks for threads.

将异步调用连接在一起，类似于pthread_join线程的工作方式。

The readme shows some good examples of using it freestyle or using the futuresubmodule using the Promise pattern. Example from the docs:

自述文件展示了一些使用自由式或使用Promise 模式使用future子模块的好例子。文档中的示例：

var Join = require('join')
  , join = Join()
  , callbackA = join.add()
  , callbackB = join.add()
  , callbackC = join.add();

function abcComplete(aArgs, bArgs, cArgs) {
  console.log(aArgs[1] + bArgs[1] + cArgs[1]);
}

setTimeout(function () {
  callbackA(null, 'Hello');
}, 300);

setTimeout(function () {
  callbackB(null, 'World');
}, 500);

setTimeout(function () {
  callbackC(null, '!');
}, 400);

// this must be called after all 
join.when(abcComplete);

回答by Alex

A simple solution might be possible here: http://howtonode.org/control-flow-part-iiscroll to Parallel actions. Another way would be to have A,B, and C all share the same callback function, have that function have an global or at least out-of-the-function incrementor, if all three have called the callback then let it run D, ofcourse you will have to store the results of A,B, and C somewhere as well.

这里可能有一个简单的解决方案：http: //howtonode.org/control-flow-part-ii滚动到并行操作。另一种方法是让 A、B 和 C 都共享相同的回调函数，让该函数具有全局或至少函数外增量器，如果这三个都调用了回调，则让它运行 D，当然，您还必须将 A、B 和 C 的结果存储在某处。

回答by Wilhelm Murdoch

Another option could be the Step module for Node: https://github.com/creationix/step

另一个选项可能是 Node 的 Step 模块：https: //github.com/creationix/step

回答by Konstantin

You may want to try this tiny library: https://www.npmjs.com/package/parallel-io

你可能想试试这个小库：https: //www.npmjs.com/package/parallel-io

回答by Daniel Garmoshka

In addition to popular promises and async-library, there is 3rd elegant way - using "wiring":

除了流行的 promise 和 async-library 之外，还有第三种优雅的方式——使用“布线”：

var l = new Wire();

funcA(l.branch('post'));
funcB(l.branch('comments'));
funcC(l.branch('links'));

l.success(function(results) {
   // result will be object with results:
   // { post: ..., comments: ..., links: ...}
});

https://github.com/garmoshka-mo/mo-wire

https://github.com/garmoshka-mo/mo-wire