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 */);

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

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

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

Answer 1

您可以使用 mergeMap 来完成。

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

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

我将为这两个示例提供一些伪代码:

1. 在 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 */);
   

2. 在多用例中，我假设批处理的大小是恒定的:

    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 */);

编辑:优化排序，添加动态批量大小情况