Node.js (ES6) - Promise.all 的 Promise 控制流

Question

我正在尝试创建一个处理并行 HTTP 请求的队列。

为每个 HTTP 请求生成一个请求对象，例如

{
    method: 'POST',
    uri: 'http://posttestserver.com/post.php',
    body: {
        some: 'foo'
    },
    json: true
}

我传入这些对象的数组并希望将它们添加到队列中，然后执行实际的请求。

问题：Promise.all 语句在返回之前等待所有请求完成。

问题：我希望将所有任务添加到队列中并在每个请求完成后返回，而不是全部等待。谁能建议如何将发出请求的逻辑与将任务添加到队列中分开？

注意：promise-queue 和 request-promise 都返回 Promise。

这是我所拥有的：

"use strict";

const Queue = require("promise-queue");
const rp = require('request-promise');

const maxConcurrent = 10;
const maxQueue = Infinity;
const queue = new Queue(maxConcurrent, maxQueue);

var options = [
    {
        method: 'POST',
        uri: 'http://posttestserver.com/post.php',
        body: {
            some: 'foo'
        },
        json: true
    },{
        method: 'POST',
        uri: 'http://posttestserver.com/post.php',
        body: {
            some: 'bar'
        },
        json: true
    }
];

Promise.all(options.map(function(task) {
        return queue.add(function() {
            return rp(task);
        })
    }))
    .then(function(response) {
        log(response);
    })
    .catch(function(err) {
        log(err);
    });


function log(data) {
    console.log(JSON.stringify(data, undefined, 4));
}

Answer 1

我建议在 forEach 循环中调用 Promise 链，如下所示：

var queueTask = function(task) {
  // returns a promise
  return queue.add(function() { return rp(task); });
};

var logResponse = function(response) {
  console.log(response);
};

var catchErrors = function(err) {
  console.log(err);
};

options.forEach(opt => {
  queueTask()
    .then(logResponse)
    .catch(catchErrors);
})

更新

在 cmets 中提到，上面的 sn-p 无法让您知道最后一个任务何时完成。这不是原始问题的要求，但由于一些 cmets 说它可能有用，因此您可以使用 Promise 来实现这一点：

var runTasks = new Promise(function(resolve, reject) {

  var queueTask = function(task) {
    // returns a promise
    return queue.add(function() { return rp(task); });
  };

  var logResponse = function(response) {
    console.log(response);
  };

  var catchErrors = function(err) {
    console.log(err);
  };

  options.forEach((opt, i, arr) => {
    
    queueTask()
      .then(logResponse)
      .catch(catchErrors);
    
    if (i === arr.length-1) { resolve(); }
    
  });
  
});

var functionToRunWhenLastTaskCompletes = function() {
  console.log('All tasks are complete.');
};

// runTasks is a promise here, not a function
// so you don't need to call it like a function
runTasks
  .then(functionToRunWhenLastTaskCompletes)