nodeJS/asyncJS 并行处理动态队列答案

【问题标题】：nodeJS/asyncJS parallel processing a dynamic queuenodeJS/asyncJS 并行处理动态队列
【发布时间】：2016-05-15 02:02:18
【问题描述】：

我有一个进程在文件夹中生成数据文件，大约每 10 秒一个新文件。

我有另一个 nodeJS 观察者，在新文件进入时监视目录。

        const watcher = chokidar.watch(['data_folder']);

        watcher.on('add', (path, stats)=>{
            if (stats && stats.size > 0){
                console.log(path);
                //spawn child_process to do further processing
                spawn_child_process_to_run(path);
            }
        });

新文件随后由 child_process 进一步处理，这可能需要相当长的时间才能完成。

问题是如何对文件进行排队，以便它们可以并行处理，而不会达到nodeJS子进程的数量限制。

【问题讨论】：

标签： node.js

【解决方案1】：

您可以使用async 库。

async.cargo 会很有用，更多信息here & here

使用指定的有效负载创建一个货物对象。添加到货物的任务将被完全处理（直到有效载荷限制）。如果工作程序正在进行中，则任务将排队，直到它变得可用。一旦工作人员完成了一些任务，就会调用这些任务的每个回调。查看这些动画，了解货物和队列的工作原理。

虽然队列一次只将一个任务传递给一组工作人员中的一个，但货物将一组任务传递给单个工作人员，当工作人员完成时重复。

var chokidar = require('chokidar');
var async = require('async')

var cargo = async.cargo(function (tasks, callback) {
    async.map(tasks,function(task,cb){
        console.log('spawn_child_process_to_run(path);',task);
        cb();
    },callback);
}, 2);// Number of tasks in parallel 

const watcher = chokidar.watch(['data_folder']);

watcher.on('add', (path, stats)=>{
    if (stats && stats.size > 0){
        cargo.push(path);//Push payload
    }
});

【讨论】：

【解决方案2】：

在async.queue的帮助下

var async = require('async');    
var exec = require('child_process').exec;


var q = async.queue(function (path, callback) {
    console.log('hello ' + path);
    exec('ping 127.0.0.1 -n 6 >nul ', (err, stdout, stderr)=>{console.log(stdout);callback()});    //simulate 6sec processing time
   }, 4);


  // assign a callback
  q.drain = function() {
     console.log('all items have been processed');
 }



 q.push([1,2,3,4,5,6,7,8],function(){console.log("done");})

【讨论】：