RxJS：并行执行 concatMap

Question

是否可以并行执行高阶可观察对象，但在合并结果时仍保留顺序？

我有这样的东西：

invoker$: Observable<void>;
fetch: (index: number) => Observable<T[]>;

invoker$
  .pipe(
    concatMap((_, index) => fetch(index)),
    scan((acc: T[], values) => [...acc, ...values], [])
  )
  .subscribe(/* Do something with the array */);

这个想法是让一个可观察对象调用回调（例如，后端调用需要花费大量时间）生成一个新的可观察对象，该可观察对象会发出单个值（某种泛型类型的数组）。返回值应连接到另一个数组中，同时保留其原始获取顺序。

但是，我希望并行触发这些请求。因此，如果快速调用 invoker$，请求将并行进行，结果在完成时合并。

我的理解是 concatMap 将等待一个 observable 完成，然后再开始下一个。 mergeMap 将并行执行，但不会执行任何操作来保留顺序。

Answer 1

我相信您要找的接线员是forkJoin。此运算符将输入一个可观察对象列表，并行触发它们，并在每个可观察对象完成后 return 列出每个可观察对象的最后发射值。

forkJoin({
   invoker: invoker$,
   fetch: fetch$,
})
.subscribe(({invoker, fetch}) => {
   console.log(invoker, fetch);
});

Answer 2

您可以使用 mergeMap 来完成。

首先，您需要将索引与异步响应一起向下传递。然后您可以根据上一步的索引进行排序。那么你有两个选择：

如果流需要在所有请求发出后结束并且只处理一次所有响应，您可以使用 reduce https://rxmarbles.com/#reduce
如果流需要继续处理另一批请求，您需要使用扫描和稍后过滤，直到达到所需的事件计数。 https://rxmarbles.com/#scan and https://rxmarbles.com/#filter

我会给你一些pseudo-code两个例子：

在 reduce 的情况下，一旦所有请求都发送完毕，流就会结束：

 invoker$
  .pipe(
   mergeMap((_, index) => fetch(index).then(value => {value, index})),
   reduce((acc: T[], singleValue) => [...acc, ...singleValue], []),
   map(array => array.sort(/*Sort on index here*/).map(valueWithIndex => valueWithIndex.value))
  )
  .subscribe(/* Do something with the array */);

在 multiple-use 的情况下，我假设批处理的大小是常数：

 invoker$
  .pipe(
   mergeMap((_, index) => fetch(index).then(value => {value, index})),
   scan((acc: T[], singleValue) => {
       let resp = [...acc, ...singleValue];
       // The scan can accumulate more than the batch size,
       // so we need to limit it and restart for the new batch
       if(resp.length > BATCH_SIZE) {
          resp = [singleValue];
       }

       return resp;
   }, []),
   filter(array => array.length == BATCH_SIZE),
   map(array => 
      array
         .sort(/*Sort on index here*/)
         .map(valueWithIndex => valueWithIndex.value))
  )
  .subscribe(/* Do something with the array */);

2.1。如果批量大小是动态的：

    invoker$
     .pipe(
      mergeMap((_, index) => fetch(index).then(value => {value, index})),
      withLatestFrom(batchSizeStream),
      scan((acc: [T[], number], [singleValue, batchSize]) => {
          let resp = [[...acc[0], ...singleValue], batchSize];
          // The scan can accumulate more than the batch size,
          // so we need to limit it and restart for the new batch
          // NOTE: the batch size is dynamic and we do not want to drop data
          // once the buffer size changes, so we need to drop the buffer 
          // only if the batch size did not change
          if(resp[0].length > batchSize && acc[1] == batchSize) {
             resp = [[singleValue], batchSize];
          }

          return resp;
      }, [[],0]),
      filter(arrayWithBatchSize => 
        arrayWithBatchSize[0].length >= arrayWithBatchSize[1]),
      map(arrayWithBatchSize => 
        arrayWithBatchSize[0]
           .sort(/*Sort on index here*/)
           .map(valueWithIndex => valueWithIndex.value))
     )
     .subscribe(/* Do something with the array */);

编辑：优化排序，添加动态批量大小案例

Answer 3

似乎这种行为是由 concatMapEager operator from the cartant/rxjs-etc library 提供的 - 由 Nicholas Jamieson 编写 (cartant) 谁是核心 RxJS 团队的开发人员。

RxJS：并行执行 concatMap

RxJS: execute concatMap i parallel

rxjs

rxjs-observables

concatmap

mergemap