javascript - 在 node.js 中协调并行执行

Question

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

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

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

使用 fork/join 模型，这将是

• fork A
• fork B
• fork C
• 加入 A、B、C，运行 D

如何在 node.js 中编写它？有没有最佳实践或食谱？我必须要hand-roll a solution每次，还是有一些图书馆为此提供帮助？

Answer 1

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

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

这听起来像是一个 hack，而且确实如此，而且它可能会留下一堆全局变量来进行跟踪，并且用一种较少的语言来进行跟踪。但是在 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);

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

---

补充答案:

实际上，即使是这样，fork() 函数已经可以使用闭包将参数传递给异步函数:

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

剩下要做的就是将 A、B、C 的结果累加并传递给 D。

---

更多附加答案:

我无法抗拒。早餐的时候一直在想这个。下面是 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));
  }
}

这很容易。这使得 fork() 相当通用，可用于同步多个非同类事件。

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);

---

更新

此代码是在 async.js 等库或各种基于 Promise 的库出现之前编写的。我想相信 async.js 是受此启发的，但我没有任何证据。无论如何..如果你今天打算这样做，请查看 async.js 或 promises。只需考虑上面的答案即可很好地解释/说明诸如 async.parallel 之类的工作原理。

为了完整起见，以下是使用 async.parallel 的方法:

var async = require('async');

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

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

使用 Promise，我们可以这样写:

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

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