限制javascript节点js中的并发性答案

【问题标题】：limiting concurrency in javascript node js限制javascript节点js中的并发性
【发布时间】：2020-01-22 17:24:25
【问题描述】：

我有以下代码：

const rl = require('readline').createInterface({
        input: require('fs').createReadStream(__dirname + '/../resources/profiles.txt'),
        terminal: true
    });

    for await (const line of rl) {
        scrape_profile(line)
    }

scrape_profile 是一个向网络发出一些请求并执行一些处理的函数。现在的问题是我想限制每 30 秒执行 5 个 scrape_profile .. 到目前为止，如果我有一个 1000 行的文本文件，它将继续执行 1000 个并发请求.. 我该怎么做限制这个？

【问题讨论】：

标签： javascript node.js concurrency

【解决方案1】：

如果您一次将整个文件异步读取到内存中，我不完全确定您为什么要使用 readlineInterface，所以对于我的回答，我已将其替换为对 fs.readFileSync 的调用处理有限值比处理流要容易得多，而且问题没有明确说明需要流式传输的文件 IO。

你可以试试Bluebird Promise.reduce:

const fs = require('fs');
const lines = fs.readFileSync('./test.txt').toString().split('\r\n');
const Promise = require('bluebird');
const BATCHES = 5;

const scrape_profile = file => new Promise((resolve, reject) => {
    setTimeout(() => {
        console.log("Done with", file);
        resolve(Math.random());
    }, Math.random() * 1000);
});

const runBatch = batchNo => {
    const batchSize = Math.round(lines.length / BATCHES);
    const start = batchSize * batchNo;
    const end = batchSize * (batchNo + 1);
    const index = start;
    return Promise.reduce(lines.slice(start, end), (aggregate, line) => {
        console.log({ aggregate });
        return scrape_profile(line)
            .then(result => {
                aggregate.push(result);
                return aggregate;
            });
    }, []);
}

runBatch(0).then(/* batch 1 done*/)
runBatch(1).then(/* batch 2 done*/)
runBatch(2).then(/* batch 3 done*/)
runBatch(3).then(/* batch 4 done*/)
runBatch(4).then(/* batch 5 done*/)
 // ... preferably use a for loop to do this

这是一个完整的例子；您应该能够在本地运行它（使用包含任何内容的名为“test.txt”的文件），对于每一行，它将花费随机时间生成一个随机数；它运行 5 个单独的批次。您需要更改BATCHES 的值以反映您需要的批次数

【讨论】：

【解决方案2】：

您可以使用 setinterval 30 秒来执行一次 scrape_profile 的循环 5 次，您的循环使用的行数就像您指定 1000 行而不停止，然后循环 5 次并将其放入您使用 setinterval 调用的函数，当然也将当前行的索引保留为变量，以便从您离开的地方继续

【讨论】：