为最大请求/秒的批次处理作业选择适当的异步方法

Question

我需要执行对一些外部API的循环调用，并有一定的延迟，以防止“超出用户速率限制”限制。

谷歌地图地理编码API对req/sec敏感，允许10 req/sec。我应该为数百个联系人进行地理编码，这样的延迟是必需的。所以，我需要有一个10异步地理编码函数，每个函数延迟1秒。所以，我收集数组中的所有触点，然后以异步方式在数组中循环。

一般来说，我需要有N个同时线程，每个线程的末尾延迟为D毫秒。整个循环遍历一系列User实体。每个线程处理单个实体，像往常一样。

我想应该有这样一个代码：

const N = 10;   # threads count
const D = 1000; # delay after each execution

var processUser = function(user, callback){ 
  someBusinessLogicProc(user, function(err) {
    setTimeout(function() {
      return callback(err);
    }, D);
  });      
 }

 var async = require('async') ;
 var people = new Array(900);

 async.batchMethod(people, processUser, N, finalCallback);

在这个伪代码中，batchMethod是我想要的方法。

Answer 1

延迟结果并不是你真正想要的。相反，您希望跟踪您发送的内容，以及何时发送，以便一旦您达到每秒请求数的界限，就可以发送另一个请求。

下面是一个函数的一般概念，它将控制速率限制为每秒固定数量的请求。这将使用promise，并要求您提供一个返回promise的请求函数（如果您现在不使用promise，您只需要将请求函数包装为promise）。

// pass the following arguments:
//   array - array of values to iterate
//   requestsPerSec - max requests per second to send (integer)
//   maxInFlight - max number of requests in process at a time
//   fn - function to process an array value
//        function is passed array element as first argument
//        function returns a promise that is resolved/rejected when async operation is done
// Returns: promise that is resolved with an array of resolves values
//          or rejected with first error that occurs
function rateLimitMap(array, requestsPerSec, maxInFlight, fn) {
    return new Promise(function(resolve, reject) {
        var index = 0;
        var inFlightCntr = 0;
        var doneCntr = 0;
        var launchTimes = [];
        var results = new Array(array.length);

        // calculate num requests in last second
        function calcRequestsInLastSecond() {
            var now = Date.now();
            // look backwards in launchTimes to see how many were launched within the last second
            var cnt = 0;
            for (var i = launchTimes.length - 1; i >= 0; i--) {
                if (now - launchTimes[i] < 1000) {
                    ++cnt;
                } else {
                    break;
                }
            }
            return cnt;            
        }

        function runMore() {
            while (index < array.length && inFlightCntr < maxInFlight && calcRequestsInLastSecond() < requestsPerSec) {
                (function(i) {
                    ++inFlightCntr;
                    launchTimes.push(Date.now());
                    fn(array[i]).then(function(val) {
                        results[i] = val;
                        --inFlightCntr;
                        ++doneCntr;
                        runMore();
                    }, reject);
                })(index);
                ++index;
            }
            // see if we're done
            if (doneCntr === array.length) {
                resolve(results);
            } else if (launchTimes.length >= requestsPerSec) {
                // calc how long we have to wait before sending more
                var delta = 1000 - (Date.now() - launchTimes[launchTimes.length - requestsPerSec]);
                if (delta >= 0) {
                    setTimeout(runMore, ++delta);
                }

            }
        }
        runMore();
    });
}

示例用法：

rateLimitMap(inputArrayToProcess, 9, 20, myRequestFunc).then(function(results) {
    // process array of results here
}, function(err) {
    // process error here
});

Github上有一个更高级的函数版本，名为rateMap（）。

这段代码背后的基本思想是：

1. 您将传入一个数组以进行迭代
2. 它返回一个promise，其解析值是一个结果数组（按顺序）
3. 您传递的requestsPerSec的最大数量为曾经命中的次数
4. 您在飞行中同时传递最大数量的请求
5. 您传递一个函数，该函数将从正在迭代的数组中传递一个元素，并且必须返回一个promise
6. 它在最后一次发送请求时保留一个时间戳数组
7. 为了查看是否可以发送另一个请求，它在数组中向后查看，并统计在最后一秒钟发送了多少个请求
8. 如果该数字低于阈值，则会发送另一个
9. 如果该数字达到阈值，那么它将计算出您需要等待多长时间才能发送另一个数字，并为该时间量设置计时器
10. 完成每个请求后，它会检查是否可以发送更多请求
11. 如果任何请求拒绝其promise，则返回的promise立即拒绝。如果您不希望它在出现第一个错误时停止，那么请修改传入函数，使其不拒绝，而是使用某个值进行解析，以便在以后处理结果时将其标识为失败的请求

这是一个工作模拟：https://jsfiddle.net/jfriend00/3gr0tq7k/

注意：如果你传入的maxInFlight值高于请求PerSec值，那么这个函数基本上只发送请求PerSec请求，然后一秒钟后，发送另一个请求PerSec请求，因为这是保持在<代码>请求PerSec边界。如果maxInFlight值与请求PerSec相同或更低，那么它将发送请求PerSec，然后当每个请求完成时，它将查看是否可以发送另一个请求。