使用 RxJS 对发出的值进行排队，就像可以使用 p-queue 一样

Question

使用p-queue，我可以限制我在一段时间内启动一些异步操作（例如，API 请求）的次数，以及一次可以运行多少个这些异步操作。

效果很好，但我觉得我应该可以用RxJS 做同样的事情。我很难弄清楚如何做到这一点。我对 RxJS 还是很陌生，我还没有找到任何可以做我想做的事情的例子。

我看到buffer 和throttleTime 等运算符，这些似乎是可行的方法，但我很难将所有这些信息放在一起。

我将如何复制 p-queue 的配置：

{
    concurrency:    2 /* at a time */
    , intervalCap: 10 /* per every… */
    , interval: (  15 /* seconds */ * 1000 /* milliseconds */)
    , carryoverConcurrencyCount: true
}

…使用RxJS?

RxJS 解决方案应该：

• 当队列为空时立即允许通过值（即立即开始一个新的时间间隔，而不是等待某个时间间隔取决于队列最后一次非空的时间）
• 提供与p-queue 的carryoverConcurrencyCount 相同的功能：“……任务必须在给定的时间间隔内完成，否则将结转到下一个时间间隔计数。”

---

使用p-queue的完整示例：

// Queue/Concurrency-limit requests
const PQueue = require('p-queue') ;
const requestQueue = new PQueue({
    concurrency:    2 /* at a time */
    , intervalCap: 10 /* per every… */
    , interval: (  15 /* seconds */ * 1000 /* milliseconds */)
    , carryoverConcurrencyCount: true
}) ;

// From https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Math/random
function getRandomInt(min, max) {
  min = Math.ceil(min);
  max = Math.floor(max);
  return Math.floor(Math.random() * (max - min)) + min; //The maximum is exclusive and the minimum is inclusive
}

const queuePromises = (
    [...(Array(20)).keys()]
    .map(number => requestQueue.add(() => new Promise(
        (resolve, reject) => setTimeout(() => resolve(number), getRandomInt(0, /* up to */ 250) /* milliseconds */))
    ))
) ;

queuePromises.forEach(queuePromise => queuePromise.then(
    number => console.log(number, 'resolved') 
    , error => console.error('Individual Promise error', error)
)) ;

Promise.all(queuePromises).then(
    numbers => console.log('all are resolved', ...numbers)
    , error => console.error('All Promises error', error)
) ;

Answer 1

我不知道 p-queue 但您可能可以查看 mergeMap 运算符来完成您想要的操作，尤其是 mergeMap 的 concurrency 参数。通过concurrency 参数，您可以定义可以同时运行多少个并行执行。

因此，从您的示例开始，代码可能是这样的

const concurrency = 1;
function getRandomInt(min, max) {
  min = Math.ceil(min);
  max = Math.floor(max);
  return Math.floor(Math.random() * (max - min)) + min; //The maximum is exclusive and the minimum is inclusive
}

const queuePromises = (
    [...(Array(20)).keys()]
    .map(number => new Promise(
      (resolve, reject) => setTimeout(() => resolve(number), getRandomInt(0, /* up to */ 250) /* milliseconds */)))
) ;

from(queuePromises).pipe(
  mergeMap(qp => from(qp), concurrency)
)
.subscribe(
  number => console.log(number, 'resolved') 
  , error => console.error('Individual Promise error', error),
  () => console.log('all are resolved')
)

将concurrency 的值设置为 1 可以让您看到实际上您有按顺序按顺序到达的 Promise 的结果。