RxJS 在并行批量/队列中运行异步任务 Array.prototype.map答案

【问题标题】：RxJS run async tasks Array.prototype.map in parallel bulks / queueRxJS 在并行批量/队列中运行异步任务 Array.prototype.map
【发布时间】：2021-08-05 06:50:49
【问题描述】：

假设我有一个变量数组，如下所示：[Sasha、Misha、Caitlyn、...String] (string[]) 等等。它有一个很大的 .length 大约 10k 元素左右。

我想和他们一起运行一个异步并行任务，但不是一次全部运行，比如Promise.all，而是批量运行，比如这样：

0 <= return await result
1 <= return await result, then next 2 (or N)
 2
 3
  4
  5
   6
   7

当然，我可以通过多种方式来实现，例如使用 for-loop 迭代原语并在内部做出承诺，然后运行它们，或者使用 p-limit，例如，但我听说 RxJS 及其运算符可以提供帮助我和那个。

根据 RxJS bufferCount 看起来就像我正在寻找的东西，但我仍然找不到必要的例子。

附：如果可能的话，我不喜欢在另一个变量中重新创建我的基元数组，并且有两个不同的数组，大约 20k 个原始数组和大约 20k 个 promise。我更喜欢按批量 (N) 迭代原语，然后形成 Promise，等待它们响应，然后迭代到下一个批量 (N)

【问题讨论】：

标签： javascript node.js typescript rxjs

【解决方案1】：

据我了解，bufferCount 也是一个选项，但使用 MergeMap 与第二个参数一起使用似乎更容易实现。

import { from } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

async function t(members: string[]) {
  await from(members).pipe(
    // try / catch block is optional
    mergeMap(async obj => { // remember to use `async` for awaiting result
      console.log(obj + '1') // any async action you do, http requests, or DB updates
    }, 2), // where 2 represent parallel
  ).toPromise()
};

t(['a', 'b', 'c', 'e', 'd', 'e', 'f', 'g', 'h', 'i']);

【讨论】：

mergeMap 的并发行为与bufferCount 的并发行为不同。使用并发参数 N，mergeMap 将一次接收一个新值，直到它同时处理 N 个值。当一个完成时，它会接收另一个。

【解决方案2】：

这取决于您想要阻止异步调用的具体程度。

假设您有 N 个输入值，并且您希望按大小分组处理它们 M.

您是否要阻止下一组的处理，直到该组中的所有项目当前组都做完了吗？（这更符合您的“返回等待结果，然后是下一个 N" 要求）。

如果是这样，那么bufferCount 就是这样。

from(members).pipe(
  bufferCount(N),

  // concatMap completes an async operation for each group in sequence. Nothing
  // happens with the next group until the current group is done.
  concatMap(async (groupOfN) => {
    // process group
  });
)

【讨论】：