用 Promise 构建一个请求队列？答案

【问题标题】：Build a request queue with promises?用 Promise 构建一个请求队列？
【发布时间】：2021-06-22 11:19:03
【问题描述】：

我的目标是针对 REST 端点运行数据导入。

我不希望在启动新请求之前等待请求得到解决。我想“模拟”并行连接。

而且我不确定我这里有基本知识问题。

此代码创建子进程：

const numchild = require('os').cpus().length;
const SIZE = 1000;
const SIZE_PER_CHILD = SIZE / numchild;

for (let i = 0; i < numchild; i++) {
  const child = child_process.fork('./child.js');
  child.send({ start: i * SIZE_PER_CHILD, end: (i + 1) * SIZE_PER_CHILD });
  // Some more code...
}

然后我想为每个子进程生成一个随机有效负载以用于导入并针对 REST 端点触发它：

process.on('message', async function({ start, end }) {
  const count = start;
  while (count < end) {
    const generatedData = 'I was generated! Yay!' + count;
    await axios.put('/api/v1/import', generatedData);
    count++;
  }
});

上述方法将等待每个导入请求完成，然后触发下一个，直到所有子导入完成。不是我想要的。

现在我要触发的端点应该能够处理比我能够生成的请求更多的请求。

我可以这样改写：

process.on('message', async function({ start, end }) {
  const count = start;
  while (count < end) {
    const generatedData = 'I was generated! Yay!' + count;
    axios.put('/api/v1/import', generatedData).then(() => console.log('I am done with this one'));
    count++;
  }
});

这种方法的问题当然是所有请求都在几秒钟内生成并针对端点触发。我猜这更像是 DOS 风格。也不是我想要的。

我希望实现的是：每个子进程有 15 个打开的连接。如果请求完成，则将下一个请求排队，直到有 15 个请求再次挂起。

所以我尝试了这个：

process.on('message', async function({ start, end }) {
  const count = start;
  let queue = [];
  while (count < end) {
    if (queue.length === 15) {
      queue = queue.filter(async (promise) => {
        const state = await promiseState(promise);
        return state !== 'fulfilled';
      });
    } else {
      const generatedData = 'I was generated! Yay!' + count;
      queue.push(axios.put('/api/v1/import', generatedData).then(() => console.log('I am done with this one')));
      count++;
    }
  }
});

function promiseState(p) {
  const t = {};
  return Promise.race([p, t])
    .then(v => (v === t) ? "pending" : "fulfilled", () => "rejected");
}

也不起作用，没有意义，对吧？过滤器函数返回承诺，因此我试图做的事情不起作用。

我有什么办法可以做到这一点？

【问题讨论】：

您可以查看p-queue，它应该可以满足您的所有需求。要么直接使用它，要么检查他们使用的代码。
老实说，我不知道我会如何使用它。遇到同样的问题...

标签： node.js asynchronous async-await promise

【解决方案1】：

试试p-queue。下面的并发设置为 3，这意味着在此队列中一次最多执行 3 个调用：

import PQueue from 'p-queue';

const queue = new PQueue({
  concurrency: 3,
});

process.on('message', async function ({ start, end }) {
  var calls = [];
  var count = start;
  while (count < end) {
    const generatedData = 'I was generated! Yay!' + count;

    calls.push(
      queue.add(() => {
        return axios
          .put('/api/v1/import', generatedData)
          .then(() => console.log('I am done with this one'));
      })
    );
    count++;
  }
  var results = await Promise.all(calls);
});

【讨论】：