javascript - 使用 Promises 在 for 循环中调用 API 的最佳方法

Question

我有 5 亿个对象，其中每个对象都有 n 个联系人，如下所示

var groupsArray = [
                    {'G1': ['C1','C2','C3'....]},
                    {'G2': ['D1','D2','D3'....]}
                     ...
                    {'G2000': ['D2001','D2002','D2003'....]}
                     ...
                ]

我在 nodejs 中有两种基于常规 promise 的实现方式，另一种使用 Bluebird ，如下所示

定期 promise

...
var groupsArray = [
                    {'G1': ['C1','C2','C3']},
                    {'G2': ['D1','D2','D3']}
                ]

function ajax(url) {
  return new Promise(function(resolve, reject) {
        request.get(url,{json: true}, function(error, data) {
            if (error) {
                reject(error);
            } else {
                resolve(data);  
            }
        });
    });
}
_.each(groupsArray,function(groupData){
    _.each(groupData,function(contactlists,groupIndex){
        // console.log(groupIndex)
        _.each(contactlists,function(contactData){
            ajax('http://localhost:3001/api/getcontactdata/'+groupIndex+'/'+contactData).then(function(result) {
                console.log(result.body);
              // Code depending on result
            }).catch(function() {
              // An error occurred
            });
        })
    })
})
...

使用 bluebird 方式我已经使用并发来检查如何控制 promise 队列

...
_.each(groupsArray,function(groupData){
    _.each(groupData,function(contactlists,groupIndex){
        var contacts = [];
        // console.log(groupIndex)
        _.each(contactlists,function(contactData){
            contacts.push({
                contact_name: 'Contact ' + contactData
            });
        })
        groups.push({
            task_name: 'Group ' + groupIndex,
            contacts: contacts
        });
    })
})

Promise.each(groups, group => 
    Promise.map(group.contacts,
         contact => new Promise((resolve, reject) => {
                /*setTimeout(() => 
                    resolve(group.task_name + ' ' + contact.contact_name), 1000);*/
                request.get('http://localhost:3001/api/getcontactdata/'+group.task_name+'/'+contact.contact_name,{json: true}, function(error, data) {
                    if (error) {
                        reject(error);
                    } else {
                        resolve(data);  
                    }
                });
}).then(log => console.log(log.body)), 
{
    concurrency: 50
}).then(() => console.log())).then(() => {
    console.log('All Done!!');
});
...

我想知道何时使用 promises 处理循环内的 1 亿个 api 调用。请告知异步调用 API 并稍后处理响应的最佳方式。

Answer 1

我的回答使用常规的 Node.js promise (这可能很容易适应 Bluebird 或其他库)。

您可以使用 Promise.all 一次触发所有 Promise:

var groupsArray = [
                    {'G1': ['C1','C2','C3']},
                    {'G2': ['D1','D2','D3']}
                ];


function ajax(url) {
  return new Promise(function(resolve, reject) {
        request.get(url,{json: true}, function(error, data) {
            if (error) {
                reject(error);
            } else {
                resolve(data);  
            }
        });
    });
}

Promise.all(groupsArray.map(group => ajax("your-url-here")))
    .then(results => {
        // Code that depends on all results.
    })
    .catch(err => {
        // Handle the error.
    });

使用 Promise.all 尝试并行运行所有请求。当您有 5 亿个请求同时尝试所有请求时，这可能不会很好地工作!

一种更有效的方法是使用 JavaScript reduce 函数将您的请求一个接一个地排序:

// ... Setup as before ...

const results = [];

groupsArray.reduce((prevPromise, group) => {
            return prevPromise.then(() => {
                return ajax("your-url-here")
                    .then(result => {
                        // Process a single result if necessary.
                        results.push(result); // Collect your results.
                    });
            });
        },
        Promise.resolve() // Seed promise.
    );
    .then(() => {
        // Code that depends on all results.
    })
    .catch(err => {
        // Handle the error.
    });

此示例将 promise 链接在一起，以便下一个 promise 仅在前一个 promise 完成后才开始。

不幸的是，排序方法会非常慢，因为它必须等到每个请求都完成后再开始一个新请求。虽然每个请求都在进行中(发出 API 请求需要时间)，但您的 CPU 处于空闲状态，而它可能正在处理另一个请求!

解决此问题的更有效但复杂的方法是结合使用上述方法。您应该对您的请求进行批处理，以便每批(比如 10 个)中的请求并行执行，然后这些批处理一个接一个地排序。

自己实现这个很棘手 - 虽然这是一个很好的学习练习- 结合使用 Promise.all 和 reduce 函数，但我建议使用库 async-await-parallel .有很多这样的库，但我用的是这个，它运行良好，很容易完成你想要的工作。

你可以像这样安装库:

npm install --save async-await-parallel

以下是您将如何使用它:

const parallel = require("async-await-parallel");

// ... Setup as before ...

const batchSize = 10;

parallel(groupsArray.map(group => {
        return () => { // We need to return a 'thunk' function, so that the jobs can be started when they are need, rather than all at once.
            return ajax("your-url-here");               
        }
    }, batchSize)
    .then(() => {
        // Code that depends on all results.
    })
    .catch(err => {
        // Handle the error.
    });

这样更好，但是发出如此大量的请求仍然是一种笨拙的方法!也许您需要提高赌注并考虑投入时间进行适当的异步作业管理。

我一直在使用 Kue最近用于管理一组工作进程。将 Kue 与 Node.js 集群库一起使用可以让您在多核 PC 上获得适当的并行性，然后如果您需要更多的处理能力，您可以轻松地将它扩展到多个基于云的虚拟机。

参见 my answer here对于一些 Kue 示例代码。