使用RxJS捕获错误后，如何继续执行顺序Observable任务？

需求说明 #

• 需顺序执行X个包含RxJS Observable链的函数调用；
• 当Observable链某环节出错时，弹出对话框询问用户是否继续执行；
• 用户选择停止则终止所有任务，选择继续则执行后续任务。

尝试方案及问题 #

• 在函数内部使用catchError，但无法根据用户选择控制流程，会直接继续执行；
• 将错误抛至subscribe的error回调，虽能触发弹窗，但无法恢复执行后续任务。

补充执行顺序要求（以3个任务为例） #

* post1
* get1
* delete1
* post2
* get2 <--- 此处出错
* 弹出对话框
===================
* 若选择继续
* delete2
* post3
* get3
* delete3
===================
* 若选择停止
* delete2

任务数量不固定，任意任务的get环节都可能出错，请问该需求是否可实现？

代码示例 #

export class AppComponent {
  constructor(private postService: PostService, public dialog: MatDialog) {}

  startEverything(): void {
    const sameFunc = (id1: number, id2: number) => {
      return this.postService.addPost(id1).pipe(
        switchMap((post: Post) => {
          return this.postService.getPosts().pipe(take(1));
        }),
        switchMap((posts: Post[]) =>
          this.postService.getPost(id2).pipe(
            map((post) => post),
            // catch any errors here and throw to down below
            catchError((error) => {
              console.log('error: ', error);
              throw new Error(error);
            })
          )
        ),
        switchMap((queryId: string) =>
          this.postService.deletePost(9).pipe(catchError(() => of(undefined)))
        )
      );
    };

    // creating an array of three function calls, second one will fail
    const requests = [sameFunc(1, 1), sameFunc(2, 999), sameFunc(3, 3)];

    from(requests)
      .pipe(
        // run three function calls sequentially
        concatAll()
      )
      .subscribe({
        next: (x) => console.log('Observer got a next value: ' + x),
        error: (err) => {
          console.error('Observer got an error: ' + err);
          // pop up dialog here with STOP/CONTINUE
          const dialogRef = this.dialog.open(DialogContentExampleDialog);

          dialogRef.afterClosed().subscribe((result) => {
            console.log(`Dialog result: ${result}`);
            if (result) {
              // continue with next concatAll() <== can I do this??
            }
          });
        },
        complete: () => {
          console.log('Observer got a complete notification');
        },
      );
  }
}

解决方案：需求可实现，具体修改如下 #

核心思路是在单个任务的Observable内部处理错误，结合用户弹窗选择决定流的走向，同时确保出错任务的收尾操作（如delete2）无论用户选择如何都会执行：

1. 重构任务函数的错误处理逻辑 #

修改sameFunc，在getPost出错时，先执行deletePost操作，再弹出对话框根据用户选择决定是否继续后续任务：

const sameFunc = (id1: number, id2: number) => {
  // 提取delete操作，确保出错时也能执行
  const deleteTask = () => this.postService.deletePost(9).pipe(catchError(() => of(undefined)));

  return this.postService.addPost(id1).pipe(
    switchMap((post: Post) => this.postService.getPosts().pipe(take(1))),
    switchMap((posts: Post[]) => 
      this.postService.getPost(id2).pipe(
        catchError((error) => {
          console.log('error: ', error);
          // 先执行delete操作，再处理用户选择
          return deleteTask().pipe(
            switchMap(() => {
              // 将对话框转换为Observable
              const dialogRef = this.dialog.open(DialogContentExampleDialog);
              return dialogRef.afterClosed().pipe(
                switchMap((result) => {
                  if (result) {
                    // 用户选择继续，返回空值让当前任务完成，触发下一个任务
                    return of(undefined);
                  } else {
                    // 用户选择停止，抛出错误终止整个流
                    throw new Error('User chose to stop');
                  }
                })
              );
            })
          );
        })
      )
    ),
    // 正常流程下执行delete操作
    switchMap((queryId: string) => deleteTask())
  );
};

2. 保持顺序执行的流逻辑不变 #

from(requests).pipe(concatAll())的逻辑无需修改，因为每个任务的错误已经被内部处理：

• 若用户选择继续，当前任务的Observable会正常完成，concatAll会自动执行下一个任务
• 若用户选择停止，当前任务的Observable会抛出错误，终止整个流

关键说明 #

• 把delete操作单独提取，确保无论是正常流程还是出错流程，都能执行收尾操作，符合补充的执行顺序要求
• 将对话框的afterClosed转换为Observable，嵌入到RxJS的流中，实现同步的流程控制
• 避免在subscribe的error回调中处理，因为一旦流进入错误状态就无法恢复，必须在流内部通过catchError处理并决定流的走向

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