如何在RxJS的mergeMap并发场景中处理错误并执行全部请求？

业务场景：需要加载200张图片，要求每次并行加载10张。用RxJS的mergeMap（带并发参数）处理这类200次HTTP请求、每次并行10次的场景本应合适，但图片可能加载失败（比如请求不存在的图片会抛出HTTP错误）。当前遇到的问题：mergeMap中若某个Observable失败，只会收到错误，其余请求会被取消或阻塞。我在各执行层级添加了catchError，尝试捕获错误并返回非错误Observable，但未成功。请问有没有办法在mergeMap并发场景下，即使存在预期错误，也能执行完所有请求？

工具函数代码：

const limitedParallelObservableExecution = <T>(listOfItems: T[],observableMethod: (item: T) => Observable<unknown>,maxConcurrency: number = maxParallelUploads): Observable<unknown> => {
  if (listOfItems && listOfItems.length > 0) {
    const observableListOfItems: Observable<T> = from(listOfItems);
    return observableListOfItems.pipe(mergeMap(observableMethod, maxConcurrency), toArray());
  } else {
    return of({});
  }
};

执行代码：

limitedParallelObservableExecution<T>(queueImages, (item) =>
  methodReturnsObservable().pipe(catchError((error) => {
    return of([]);
  }))
).subscribe((value) => console.log('value: ', value));

subscribe中的console.log从未执行。

编辑补充：排查后发现问题出在methodReturnsObservable()的实现（即downloadVehicleImage方法）中，移除http.get后的pipe可解决问题，我将进一步调试。

downloadVehicleImage代码：

downloadVehicleImage(imageId: string, width: number): Observable<any> {
  const params = objectToActualHttpParams({
    width,
    noSpinner: true
  });
  return this.http.get(`${environment.WS_ENDPOINT_URI}/vehicles/image/${imageId}`, { responseType: 'blob', params }).pipe(
    mergeMap((blob) => {
      if (blob.size > 0) {
        const image = new Subject();
        const reader = new FileReader();
        reader.readAsDataURL(blob);
        reader.onloadend = () => image.next(this.domSanitizer.bypassSecurityTrustResourceUrl(reader.result as string));
        return image.asObservable();
      }
      return of();
    })
  );
}

核心问题定位 #

你当前代码的关键问题有两个：

1. downloadVehicleImage内部的Observable未正常完成：当blob.size > 0时，你创建了Subject但只调用了next()，没有调用complete()，导致这个Observable永远处于活跃状态，mergeMap会一直等待它完成，最终toArray()无法收齐所有结果，subscribe回调永远不执行。
2. 未在内部捕获HTTP请求错误：http.get如果失败，错误会直接冒泡，即使外层加了catchError，但错误导致的流终止会影响整个并发流的完成逻辑。

修复步骤 #

1. 完善downloadVehicleImage内部逻辑 #

给http.get添加内部错误捕获，同时确保Subject能正常完成，还要处理FileReader的错误：

downloadVehicleImage(imageId: string, width: number): Observable<any> {
  const params = objectToActualHttpParams({
    width,
    noSpinner: true
  });
  return this.http.get(`${environment.WS_ENDPOINT_URI}/vehicles/image/${imageId}`, { responseType: 'blob', params }).pipe(
    mergeMap((blob) => {
      if (blob.size > 0) {
        const image = new Subject();
        const reader = new FileReader();
        
        reader.readAsDataURL(blob);
        reader.onloadend = () => {
          image.next(this.domSanitizer.bypassSecurityTrustResourceUrl(reader.result as string));
          image.complete(); // 必须调用complete，让Observable结束
        };
        // 处理FileReader读取失败的情况
        reader.onerror = () => {
          console.error('Blob读取失败:', reader.error);
          image.next(null); // 或返回错误标记
          image.complete();
        };
        
        return image.asObservable();
      }
      return of(null); // 返回明确的已完成Observable
    }),
    catchError((error) => {
      console.error('图片请求失败:', error);
      return of(null); // 捕获HTTP错误，返回正常完成的Observable
    })
  );
}

2. 确保外层流正常完成 #

你的执行代码中已经在外层添加了catchError，现在内部已处理所有可能的错误源，每个子Observable都会正常完成，mergeMap和toArray()就能按预期工作，subscribe的回调会在所有请求完成后执行。

3. 可选：区分成功/失败结果 #

可以修改返回值结构，方便后续处理成功和失败的项：

// 在downloadVehicleImage的catchError中返回失败标记
catchError((error) => {
  console.error('图片请求失败:', error);
  return of({ success: false, imageId, error });
});
// 成功时返回
image.next({ success: true, imageId, data: this.domSanitizer.bypassSecurityTrustResourceUrl(reader.result as string) });

内容的提问来源于stack exchange，提问作者Luigi Cordoba Granera