如何为RxJS管道添加两个错误处理器？附现有实现方案

现有方案的潜在问题 #

你当前的拆分管道方案存在关键隐患：retrying$和errorCommunicating$会分别订阅source$，这意味着每次foo$被订阅时，source$会被触发两次（比如source$是HTTP请求时，会发起两次重复请求）。另外全局变量needsRetrying与Subject的组合也增加了状态管理复杂度，容易出现状态不一致的问题。

更优实现思路 #

无需拆分管道，可通过统一状态流传递成功/错误信息，同时内置可控重试机制，既避免重复订阅源，又简化逻辑：

• 用结构化状态对象封装结果，明确区分成功、错误状态
• 利用retryWhen结合信号触发实现可控重试
• 在管道内将错误转化为状态值而非终止流，确保消费者感知错误的同时不中断重试逻辑

优化后的代码实现 #

import { catchError, Observable, of, retryWhen, Subject, map } from "rxjs";

/** 定义状态类型，明确区分成功/错误状态 */
type FooState<T> = 
  | { type: 'success', data: T, error: null }
  | { type: 'error', data: null, error: unknown };

/** 可重复执行的源Observable */
declare const source$: Observable<unknown>;

/** 重试触发信号 */
const retryTrigger$ = new Subject<void>();

// 构建统一的状态流
const foo$: Observable<FooState<unknown>> = source$.pipe(
  // 成功时包装为成功状态
  map(data => ({ type: 'success', data, error: null })),
  // 错误时包装为错误状态，不终止流
  catchError(error => of({ type: 'error', data: null, error })),
  // 重试逻辑：仅当收到重试信号时重新订阅源
  retryWhen(errors => errors.pipe(() => retryTrigger$))
);

/** 触发重试的方法 */
export function triggerRetry() {
  retryTrigger$.next();
}

/** 获取统一状态流 */
export function getFoo$() {
  return foo$;
}

方案优势 #

• 无重复订阅：整个流仅订阅一次source$，不会产生重复副作用
• 状态清晰：通过类型明确的状态对象，消费者可直接判断当前状态
• 逻辑简洁：移除全局变量，用retryWhen和Subject实现可靠的可控重试
• 兼容性强：foo$始终为同一个Observable，满足多消费端复用需求

消费者使用示例 #

getFoo$().subscribe(state => {
  if (state.type === 'success') {
    console.log('收到数据:', state.data);
  } else {
    console.error('发生错误:', state.error);
    // 消费者可自主选择是否触发重试
    // triggerRetry();
  }
});

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