协调node.js中的并行执行

Question

协调node.js中的并行执行

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

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

但并行执行怎么样？假设您有三个可以并行运行的任务A，B，C，当它们完成时，您希望将结果发送到任务D.

使用fork / join模型，这将是

• 叉子A.

• 叉子B.

• 叉子C.

• 加入A，B，C，运行D.

我如何在node.js中编写它？有没有最佳做法或烹饪书？我是否每次都必须手动滚动解决方案，或者是否有一些带帮助程序的库？

Answer 1

在node.js中没有什么是真正的并行，因为它是单线程的。但是，可以按照您无法事先确定的顺序安排和运行多个事件。而数据库访问等实际上是“并行”的，因为数据库查询本身在不同的线程中运行，但在完成时会重新集成到事件流中。那么，如何在多个事件处理程序上安排回调？嗯，这是浏览器端javascript动画中使用的一种常用技术：使用变量来跟踪完成情况。这听起来像是一个黑客，它听起来可能是混乱的，留下一堆全局变量围绕着跟踪，而用较少的语言。但在javascript中我们可以使用闭包：function&nbsp;fork&nbsp;(async_calls,&nbsp;shared_callback)&nbsp;{ &nbsp;&nbsp;var&nbsp;counter&nbsp;=&nbsp;async_calls.length; &nbsp;&nbsp;var&nbsp;callback&nbsp;=&nbsp;function&nbsp;()&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;counter&nbsp;--; &nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(counter&nbsp;==&nbsp;0)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;shared_callback() &nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;} &nbsp;&nbsp;for&nbsp;(var&nbsp;i=0;i<async_calls.length;i++)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;async_calls[i](callback); &nbsp;&nbsp;}}//&nbsp;usage:fork([A,B,C],D);在上面的例子中，我们通过假设async和callback函数不需要参数来保持代码简单。您当然可以修改代码以将参数传递给异步函数，并让回调函数累积结果并将其传递给shared_callback函数。补充答案：实际上，即使是这样，该fork()函数也可以使用闭包将参数传递给异步函数：fork([ &nbsp;&nbsp;function(callback){&nbsp;A(1,2,callback)&nbsp;}, &nbsp;&nbsp;function(callback){&nbsp;B(1,callback)&nbsp;}, &nbsp;&nbsp;function(callback){&nbsp;C(1,2,callback)&nbsp;}],D);剩下要做的唯一事情是累积A，B，C的结果并将它们传递给D.更多补充答案：我无法抗拒。在早餐期间一直想着这个。这fork()是累积结果的实现（通常作为参数传递给回调函数）：function&nbsp;fork&nbsp;(async_calls,&nbsp;shared_callback)&nbsp;{ &nbsp;&nbsp;var&nbsp;counter&nbsp;=&nbsp;async_calls.length; &nbsp;&nbsp;var&nbsp;all_results&nbsp;=&nbsp;[]; &nbsp;&nbsp;function&nbsp;makeCallback&nbsp;(index)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;return&nbsp;function&nbsp;()&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;counter&nbsp;--; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;var&nbsp;results&nbsp;=&nbsp;[]; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;we&nbsp;use&nbsp;the&nbsp;arguments&nbsp;object&nbsp;here&nbsp;because&nbsp;some&nbsp;callbacks&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;in&nbsp;Node&nbsp;pass&nbsp;in&nbsp;multiple&nbsp;arguments&nbsp;as&nbsp;result. &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;for&nbsp;(var&nbsp;i=0;i<arguments.length;i++)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;results.push(arguments[i]); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;all_results[index]&nbsp;=&nbsp;results; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(counter&nbsp;==&nbsp;0)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;shared_callback(all_results); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;} &nbsp;&nbsp;for&nbsp;(var&nbsp;i=0;i<async_calls.length;i++)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;async_calls[i](makeCallback(i)); &nbsp;&nbsp;}}这很容易。这fork()非常通用，可用于同步多个非同类事件。Node.js中的示例用法：//&nbsp;Read&nbsp;3&nbsp;files&nbsp;in&nbsp;parallel&nbsp;and&nbsp;process&nbsp;them&nbsp;together:function&nbsp;A&nbsp;(c){&nbsp;fs.readFile('file1',c)&nbsp;};function&nbsp;B&nbsp;(c){&nbsp;fs.readFile('file2',c)&nbsp;};function&nbsp;C&nbsp;(c){&nbsp;fs.readFile('file3',c)&nbsp;};function&nbsp;D&nbsp;(result)&nbsp;{ &nbsp;&nbsp;file1data&nbsp;=&nbsp;result[0][1]; &nbsp;&nbsp;file2data&nbsp;=&nbsp;result[1][1]; &nbsp;&nbsp;file3data&nbsp;=&nbsp;result[2][1]; &nbsp;&nbsp;//&nbsp;process&nbsp;the&nbsp;files&nbsp;together&nbsp;here}fork([A,B,C],D);更新此代码是在存在像async.js或各种基于promise的库之类的库之前编写的。我想相信async.js的灵感来自于此，但我没有任何证据。无论如何..如果你今天想要这样做，请看看async.js或promises。只需考虑上面的答案，就async.parallel之类的工作方式做了很好的解释/说明。为了完整起见，以下是您如何做到这一点async.parallel：var&nbsp;async&nbsp;=&nbsp;require('async');async.parallel([A,B,C],D);请注意，async.parallel它与fork我们上面实现的功能完全相同。主要区别在于它将第一个参数传递给错误，D并根据node.js约定将回调传递为第二个参数。使用promises，我们将其编写如下：//&nbsp;Assuming&nbsp;A,&nbsp;B&nbsp;&&nbsp;C&nbsp;return&nbsp;a&nbsp;promise&nbsp;instead&nbsp;of&nbsp;accepting&nbsp;a&nbsp;callbackPromise.all([A,B,C]).then(D);

Answer 2

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

Answer 3

该期货模块有一个名为子模块加入，我喜欢用：将异步调用连接在一起类似于pthread_join线程的工作方式。自述文件显示了使用自由式或使用Promise模式使用未来子模块的一些很好的例子。来自文档的示例：var&nbsp;Join&nbsp;=&nbsp;require('join') &nbsp;&nbsp;,&nbsp;join&nbsp;=&nbsp;Join() &nbsp;&nbsp;,&nbsp;callbackA&nbsp;=&nbsp;join.add() &nbsp;&nbsp;,&nbsp;callbackB&nbsp;=&nbsp;join.add() &nbsp;&nbsp;,&nbsp;callbackC&nbsp;=&nbsp;join.add();function&nbsp;abcComplete(aArgs,&nbsp;bArgs,&nbsp;cArgs)&nbsp;{ &nbsp;&nbsp;console.log(aArgs[1]&nbsp;+&nbsp;bArgs[1]&nbsp;+&nbsp;cArgs[1]);}setTimeout(function&nbsp;()&nbsp;{ &nbsp;&nbsp;callbackA(null,&nbsp;'Hello');},&nbsp;300);setTimeout(function&nbsp;()&nbsp;{ &nbsp;&nbsp;callbackB(null,&nbsp;'World');},&nbsp;500);setTimeout(function&nbsp;()&nbsp;{ &nbsp;&nbsp;callbackC(null,&nbsp;'!');},&nbsp;400);//&nbsp;this&nbsp;must&nbsp;be&nbsp;called&nbsp;after&nbsp;all&nbsp;join.when(abcComplete);