如何在RxJS顺序执行的Observable序列出错时取消后续调用？

核心原理 #

concat 操作符的默认行为就是按顺序执行Observable，一旦某个Observable抛出错误，立即终止后续所有Observable的执行，这完全匹配你的需求。你当前代码里的 catchError 只是将错误流转换为正常流，用于给订阅者返回错误提示，但不会改变「后续Observable被中止」的事实。

正确实现示例 #

import { concat, of, throwError } from 'rxjs';
import { catchError, map } from 'rxjs/operators';

// 模拟业务Observable
const verifyToken = of(1, 2);
const verifyUser = of(3);
const verificationWithError = throwError(() => new Error('验证失败'));
const checkRoles = of(4, 5);

concat(
  verifyToken,
  verifyUser,
  verificationWithError, // 抛出错误后，checkRoles不会被执行
  checkRoles
)
.pipe(
  map(val => val), // 按需处理正常流数据
  catchError(err => {
    console.error('执行流程已中止:', err.message);
    return of('no more requests!!!');
  })
)
.subscribe(console.log);

执行结果 #

输出顺序为：1 → 2 → 3 → 控制台打印错误信息 → no more requests!!!，checkRoles 对应的 4,5 完全不会输出，达到了出错即中止的效果。

关键注意事项 #

• 不要给单个Observable添加局部 catchError：如果给 verificationWithError 单独加 catchError，它的错误会被吞掉，concat 会继续执行后续的 checkRoles，违背你的需求。
• 关于 forkJoin：forkJoin 是并行执行所有Observable，只要有一个出错就会立即报错，但它不保证顺序执行。如果必须顺序执行，concat 是最优选择。

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