排隊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; 
}