对API调用队列进行速率限制并返回结果

如何解决对API调用队列进行速率限制并返回结果

我正在遍历一个数组，并使用async / await对每个成员进行API调用，然后将结果推送到另一个数组中，并返回该数组。

// My current function
async requestForEach(repos) {
    const result = [];
    for (const repo of repos) {
        result.push(await this.doSomething(repo.name));
    }
    return result;
}

// doSomething()
const AWS = require('aws-sdk');
const codecommit = new AWS.CodeCommit();
async doSomething(repoName){
    return (await codecommit.listBranches({
        repoName
    }).promise()).branches;
}

我的问题是速率受到限制。如果我发现并打印错误，则会得到..

ThrottlingException: Rate exceeded {
  // Call stack here
  code: 'ThrottlingException',time: 2020-08-16T15:52:56.632Z,requestId: '****-****-****-****-****',statusCode: 400,retryable: true
}

我正在使用的API的文档可以在这里找到-https://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/CodeCommit.html#listBranches-property

我研究了选项，this async library似乎是最受欢迎的选项。

使用async.queue（）。

添加到队列中的任务将并行处理（直到并发限制）。如果所有工作人员都在进行中，则将任务排队直到一个可用。工人完成一项任务后，任务的回调被调用。
// create a queue object with concurrency 2
var q = async.queue(function(task,callback) {
    console.log('hello ' + task.name);
    callback();
},2);

很明显，我无法从回调函数中获取值，那么该如何解决这个问题？

解决方法

顺序for … of循环对我来说看起来不错。您可以add a default delay for each iteration使其变慢，但是也可以稍后在由于限制而失败的情况下稍后重试请求。请注意，只有当您的应用程序中只有一个请求源（没有对requestForEach的多个并发调用）时，此方法才能很好地起作用，否则您可能需要全局协调。

async doSomething(repoName) {
    while (true) {
        try {
            const data = await codecommit.listBranches({
                repoName
            }).promise();
            return data.branches;
        } catch(err) {
            if (err.code == 'ThrottlingException') { // if (err.retryable) {
                await delay(err.retryDelay ?? 1000);
                continue;
            } else {
                throw err;
            }
        }
    }
}
function delay(time) {
    return new Promise(resolve => {
        setTimeout(resolve,time);
    });
}

递归方法可能比while (true)循环更好。请注意，在生产代码中，您将希望限制重试次数，以使循环永远不会无限运行。

看起来像想要parallelLimit。

它需要一个可选的回调来接收结果。

来自文档。

https://caolan.github.io/async/v3/docs.html#parallelLimit

回调函数所有功能成功完成后运行的可选回调。此函数获取一个结果数组（或对象），该数组包含传递给任务回调的所有结果参数。调用了（错误，结果）。

示例：

// run 'my_task' 100 times,with parallel limit of 10

  var my_task = function(callback) { ... };
  var when_done = function(err,results) { ... };

  // create an array of tasks
  var async_queue = Array(100).fill(my_task);

  async.parallelLimit(async_queue,10,when_done);

来自： how to use async.parallelLimit to maximize the amount of (paralle) running processes?

您可以如下使用Promise.all来减少API调用的等待时间

async requestForEach(repos) {
  return Promise.all(repos.map(repo => this.doSomething(repo.value)));
}

由于总通话数量出现rate limit问题，因此可以使用es6-promise-pool之类的库来管理并发请求（5/10-根据您的要求）。

并使用以下的递归和 MAX_RETRIES （从this.doSomething控制MAX_RETRIES）限制更新environment variable

async doSomething(repoName,retries = 0) {
    try {
        const data = await codecommit.listBranches({
            repoName
        }).promise();
        return data.branches;
    } catch(err) {
        if (err.code == 'ThrottlingException' && retries <= MAX_RETRIES) {
            await delay(err.retryDelay ?? 1000); // As per @Bergi's answer
            await doSomething(repoName,retries + 1); // Recursive call
        } else {
            console.log('Issue with repo: ',repoName);
            throw err; // (Or) return ''; based on requirement
        }
    }
}


// Filter out the valid results at the end - Applicable only if you use return '';
const results = await requestForEach(repos);
const finalResults = results.filter(Boolean);

这种方法可以帮助您减少按顺序循环每个请求的生产中的等待时间。

对API调用队列进行速率限制并返回结果

如何解决对API调用队列进行速率限制并返回结果

解决方法

相关推荐