RxJS forkJoin报错后仍执行后续请求，如何终止后续请求？

forkJoin数组中的请求即使前一个失败仍会继续执行。例如第一个请求putCategory$因名称已存在报错，但后续loadImage$请求仍会执行并加载新图片。请问原因是什么？如何实现请求序列在失败时终止后续请求？

代码示例 #

PUTCategory(category: ICategory) {
    const newImage = category.image;
    const oldImage = this.editedCategory.image;
    const imageChanged = oldImage !== newImage;
    const loadImage = imageChanged && !!newImage;
    const deleteImage = imageChanged && !!oldImage;

    const file: File = this.form.value.image;

    const putCategory$ = this.categoryService.putCategory(category).pipe(
      catchError((response: any) => {
        if (response.error.info == 'NAME_EXISTS') {
          this.throwErrorModal(
            'Категория с таким названием уже существует. Измените название и попробуйте снова.',
          );
        } else {
          this.throwErrorModal(
            'Произошла ошибка при попытке обновить категорию',
          );
        }
        return EMPTY;
      }),
    );

    const loadImage$ = loadImage
      ? this.categoryService.uploadCategoryImage(file).pipe(
          catchError(() => {
            this.throwErrorModal(
              'Произошла ошибка при попытке загрузить файл нового изображения категории',
            );
            return EMPTY;
          }),
          concatMap((response: any) => {
            const filename = response.filename;
            return this.categoryService
              .setImageToCategory(category.id, filename)
              .pipe(
                catchError(() => {
                  this.throwErrorModal(
                    'Произошла ошибка при попытке связать новое загруженное изображение и категорию',
                  );
                  return EMPTY;
                }),
              );
          }),
        )
      : of(null);

    const deleteImage$ = deleteImage
      ? this.categoryService.deleteImage(oldImage!).pipe(
          catchError(() => {
            this.throwErrorModal(
              'Произошла ошибка при попытке удалить старое изображение категории',
            );
            return EMPTY;
          }),
          concatMap(() => {
            return this.categoryService
              .setImageToCategory(category.id, '')
              .pipe(
                catchError(() => {
                  this.throwErrorModal(
                    'Произошла ошибка при попытке удаления связи старого изображения с категорией',
                  );
                  return EMPTY;
                }),
              );
          }),
        )
      : of(null);

    forkJoin([putCategory$, deleteImage$, loadImage$])
      .pipe(
        finalize(() => {
          this.awaitModal = false;
          this.cdr.markForCheck();
        }),
        tap(() => {
          this.successModal = true;
        }),
      )
      .subscribe();
  }

补充函数 #

private throwErrorModal(content: string) {
    this.errorModalContent = content;
    this.errorModal = true;
  }

原因分析 #

1. forkJoin的并行特性：forkJoin会在订阅时立即同时启动所有传入的Observable，请求一旦启动就会执行完毕，不受其他请求状态影响。
2. 错误被“吞掉”：每个请求内部的catchError都返回了EMPTY，这会把出错的Observable转换成一个正常完成的流，forkJoin会认为该请求已正常结束，不会触发整个流的中断。

解决方案 #

要实现“前一个失败则终止后续请求”，需要把并行执行改为串行执行，同时让错误能够向上传递，而不是被内部catchError吞掉。

方法1：用concat串联请求 #

concat会严格按顺序执行Observable，前一个流完成后才会启动下一个，一旦某个流出错，整个序列会立即终止。

// 修改putCategory$：出错时返回throwError而非EMPTY，让错误向上传递
const putCategory$ = this.categoryService.putCategory(category).pipe(
  catchError((response: any) => {
    const message = response.error.info === 'NAME_EXISTS'
      ? 'Категория с таким названием уже существует. Измените название и попробуйте снова.'
      : 'Произошла ошибка при попытке обновить категорию';
    this.throwErrorModal(message);
    return throwError(() => new Error(message));
  }),
);

// 同理修改loadImage$和deleteImage$，出错时返回throwError
const loadImage$ = loadImage
  ? this.categoryService.uploadCategoryImage(file).pipe(
      catchError(() => {
        const message = 'Произошла ошибка при попытке загрузить файл нового изображения категории';
        this.throwErrorModal(message);
        return throwError(() => new Error(message));
      }),
      concatMap((response: any) => {
        const filename = response.filename;
        return this.categoryService.setImageToCategory(category.id, filename).pipe(
          catchError(() => {
            const message = 'Произошла ошибка при попытке связать новое загруженное изображение и категорию';
            this.throwErrorModal(message);
            return throwError(() => new Error(message));
          }),
        );
      }),
    )
  : of(null);

const deleteImage$ = deleteImage
  ? this.categoryService.deleteImage(oldImage!).pipe(
      catchError(() => {
        const message = 'Произошла ошибка при попытке удалить старое изображение категории';
        this.throwErrorModal(message);
        return throwError(() => new Error(message));
      }),
      concatMap(() => {
        return this.categoryService.setImageToCategory(category.id, '').pipe(
          catchError(() => {
            const message = 'Произошла ошибка при попытке удаления связи старого изображения с категорией';
            this.throwErrorModal(message);
            return throwError(() => new Error(message));
          }),
        );
      }),
    )
  : of(null);

// 用concat按顺序执行：先更新分类，再删除旧图，最后上传新图
concat(putCategory$, deleteImage$, loadImage$)
  .pipe(
    finalize(() => {
      this.awaitModal = false;
      this.cdr.markForCheck();
    }),
    tap(() => {
      this.successModal = true;
    }),
    catchError(() => EMPTY) // 最外层捕获错误，避免流崩溃
  )
  .subscribe();

方法2：用switchMap实现灵活串行 #

如果需要根据前一个请求的结果判断是否执行后续请求，用switchMap链式调用更灵活：

this.categoryService.putCategory(category).pipe(
  catchError((response: any) => {
    const message = response.error.info === 'NAME_EXISTS'
      ? 'Категория с таким названием уже существует. Измените название и попробуйте снова.'
      : 'Произошла ошибка при попытке обновить категорию';
    this.throwErrorModal(message);
    return throwError(() => new Error(message));
  }),
  // 分类更新成功后，根据条件执行删除旧图请求
  switchMap(() => deleteImage ? deleteImage$ : of(null)),
  // 再根据条件执行上传新图请求
  switchMap(() => loadImage ? loadImage$ : of(null)),
  finalize(() => {
    this.awaitModal = false;
    this.cdr.markForCheck();
  }),
  tap(() => {
    this.successModal = true;
  }),
  catchError(() => EMPTY)
).subscribe();

核心要点 #

• 去掉内部catchError返回EMPTY的逻辑，换成throwError让错误向上传递，这样才能触发序列终止。
• 使用串行操作符（concat/switchMap）替代并行的forkJoin，确保请求按顺序启动，前一个失败则后续请求不会被订阅执行。

内容的提问来源于stack exchange，提问作者krabochki