排队URL和异步代码

Question

我有以下代码，我使用wappalyzer刮擦网站的技术数据。

const wappalyzer = require('wappalyzer'); 
var fs = require('fs'); 

var myArray =[http://www.url1.com, http://www.url2.com ...] // it's a very 
long array of URLs 

var dataSlice = myArray.slice(0, 1000); 

console.log(dataSlice); 
fs.appendFileSync('webData.json', '[', 'utf8'); 

var done = {}; 
var count = 0; 
for(i = 0; i < dataSlice.length; i++) { 
    (function(i){ 
    wappalyzer.run([dataSlice[i], '--quiet'], function(stdout, stderr) { 
     //Keep track of when all urls are done 

     if(!done[i]){ 
     done[i] = true; 
     count++; 

     if (stdout) { 
      var arr = stdout.split('\n'); 
      stdout = arr.filter(function(elem, pos) { 
       return arr.indexOf(elem) == pos; 
      }); 
      stdout = stdout.join(''); 
      stdout = count >= dataSlice.length ? (stdout + ']') : (stdout + 
','); 
      fs.appendFileSync('WebData.json', stdout, 'utf8'); 
     } 

     } 

     if (stderr) { 
     process.stderr.write(stderr); 
     } 
    }); 
    })(i); 
} 

的问题是我有我通过这个循环运行的URL的一个非常大的列表，以便我切片阵列成较小的部分做〜1000同时然而，当我做任何更多的systemfreezes因为它使用相当多的CPU，我将如何异步运行此代码并排列URL？还有一个简单的NPM软件包可以安装吗？我环顾四周，没有发现任何易于使用的东西。

Answer 1

你可以做一个环形缓冲区。

这个想法是有一个待处理任务的数组，有一个循环（在节点中，循环必须是异步的，不会阻塞引擎），它可以完成工作来重新填充和启动作业，任务结束。

请注意不要在回调中覆盖铃声，因为您会看到我总是将它从呼叫复制到呼叫并同步写入。

如果我正确的话，下面应该做这个工作。

// concurrent/async job processing. 
function main() { 

    var consumer = function(todo, done){ // execute whatever to do. 
    console.log("start ", todo.id, todo.duration); 
    setTimeout(function(){ 
     done(); 
     console.log("done ", todo.id, todo.duration); 
    }, todo.duration); 
    }; 

    var id = 0; 
    function reader(len) { // a dummy function that provides n new jobs 
    var ret = [] 
    for(i=0;i<len;i++){ 
     ret.push({ 
     id: id, 
     duration:getRandomInt(250, 1000), 
     about: "whatever todo", 
     status: "todo" 
     }); 
     id+=1; 
    } 
    return ret 
    }; 

    var ring = []; // the ring contains the element being processed 
    var loop = function(){ // is async 
    ring = processor(ring, 10, reader, consumer); 
    logger(ring); 
    setTimeout(loop, 100); // loop here. 
    };loop(); 

};process.nextTick(main); 

function processor(ring, concurrency, reader, consumer){ 
    ring = ring.filter(notStatus('done')); // remove done jobs 
    var delta = concurrency-ring.length; 
    if(delta>0) { // refill the ring until it is full 
    ring = ring.concat(reader(delta)) 
    } 
    ring.filter(byStatus('todo')).forEach(function(todo){ // select those todo 
    todo.status = 'started'; 
    consumer(todo, function(){ 
     todo.status = "done"; 
    }) 
    }) 
    return ring 
} 

function logger(ring){ 
    console.log(" ring"); 
    ring.forEach(function(todo){ 
    console.log(" ", todo.status, todo.id, todo.duration); 
    }); 
} 

function byStatus(s){ 
    return function(todo){return todo.status==s} 
} 

function notStatus(s){ 
    return function(todo){return todo.status!=s} 
} 

// .... still not std ? 
function getRandomArbitrary(min, max) { 
    return Math.random() * (max - min) + min; 
} 
function getRandomInt(min, max) { 
    return Math.floor(Math.random() * (max - min + 1)) + min; 
}