节点js中的限制q承诺并发性

Question

有没有办法限制在节点js中同时执行的并发q承诺的数量？

我正在构建一个web 刮板，它必须请求和解析更多的3000+页面，并且不需要节流，我提出的一些请求没有及时响应，因此连接停止，所需的响应(html代码)变得不可用。

为了应对这种情况，我发现限制我的问题的请求数量就会消失。

我曾尝试过以下方法，但均徒劳无功：

• Concurrency limit in Q promises - node
• How can I limit Q promise concurrency?
• https://gist.github.com/gaearon/7930162
• https://github.com/ForbesLindesay/throat

我需要请求一个urls数组，每次只执行一个请求，并且当数组中的所有urls都完成时，然后返回数组中的结果。

function processWebsite() {
  //computed by this stage
  urls = [u1,u2,u3,u4,l5,u6,u7,u8,u9];

  var promises = throttle(urls,1,myfunction);

  // myfunction returns a Q promise and takes a considerable 
  // amount of time to resolve (approximately 2-5 minutes)
  
  Q.all(promises).then(function(results){
      //work with the results of the promises array
  });
}

Answer 1

您可以在then()块中请求一个新的url

myFunction(urls[0]).then(function(result) {
  myFunction(urls[1]).then(function(result) {
    myFunction(urls[2]).then(function(result) {
      ...
    });
  });
});

当然，这将是它的动态行为。一旦承诺达成，我就会排长队并排出一个url。那就提出另一个请求。并且可能有一个哈希对象将urls与结果关联起来。

第二次拍摄：

var urls = ...;
var limit = ...;
var dequeue = function() {
  return an array containing up to limit
};

var myFunction = function(dequeue) {
  var urls = dequeue();

  $q.all(process urls);
};

myFunction(dequeue).then(function(result) {
  myFunction(dequeue).then(function(result) {
    myFunction(dequeue).then(function(result) {
      ...
    });
  });
});

Answer 2

我会这样做，这将迭代每个URL，构建一个承诺链，当上一个承诺完成时运行，并使用请求结果数组进行解析。

return urls.reduce(function(acc, url){
    return acc.then(function(results)
        return myfunction(url).then(function(requestResult){
             return results.concat(requestResult)
        });
    });
}, Q.resolve([]));

你也可以把它变成帮手：

var results = map(urls, myfunction);

function map(items, fn){
    return items.reduce(function(acc, item){
        return acc.then(function(results)
            return fn(item).then(function(result){
                 return results.concat(result)
            });
        });
    }, Q.resolve([])
}

注意，bluebird承诺库有一个助手来简化这类事情。

return Bluebird.map(urls, myfunction, {concurrency: 1});

Answer 3

下面是我为Q制作一个节流map函数的尝试。

function qMap(items, worker, concurrent) {
    var result = Q.defer();
    var work = [];
    var working = 0;
    var done = 0;

    concurrent = parseInt(concurrent, 10) || 1;

    function getNextIndex() {
        var i;
        for (i = 0; i < items.length; i++) {
            if (typeof work[i] === "undefined") return i;
        }
    }
    function doneWorking() {
        working--;
        done++;
        result.notify( +((100 * done / items.length).toFixed(1)) );
        if (!startWorking() && done === items.length) {
            result.resolve(work);
        }
    }
    function startWorking() {
        var index = getNextIndex();
        if (typeof index !== "undefined" && working < concurrent) {
            working++;
            work[index] = worker(items[index]).finally(doneWorking);
            return true;
        }
    }
    while (startWorking());
    return result.promise;
}

它接受

• 要工作的items数组(URL，在您的例子中)，
• worker (必须是接受项并返回承诺的函数)
• 以及在任何给定时间内要处理的concurrent项的最大值。

它回来了

• 一个承诺和
• 当所有的工人都做完后，就决定许下一系列的承诺。

它不会失败，您必须检查单个承诺，以确定操作的总体状态。

在您的例子中，您可以这样使用它，例如，对15个并发请求：

// myfunction returns a Q promise and takes a considerable 
// amount of time to resolve (approximately 2-5 minutes)

qMap(urls, myfunction, 15)
.progress(function (percentDone) {
    console.log("progress: " + percentDone);
})
.done(function (urlPromises) {
    console.log("all done: " + urlPromises);
});