如何提取RxJS多个函数中重复的错误处理及前置代码？

方案一：封装通用订阅方法 #

直接把重复的错误重置、错误回调逻辑封装成组件内的私有方法，调用时只传目标Observable和next回调即可：

// 封装通用订阅逻辑
private safeSubscribe<T>(obs$: Observable<T>, nextHandler: (value: T) => void) {
  this.error = undefined; // 前置重置错误状态
  obs$.subscribe({
    next: nextHandler,
    error: (error) => this.error = error
  });
}

// 改造原方法
f1() {
  this.safeSubscribe(obs1$, () => some_code);
}

f2() {
  this.safeSubscribe(obs2$, () => some_other_code);
}

这种方式改动最小，完全贴合你原有代码的调用习惯，把重复代码彻底抽离。

方案二：自定义RxJS操作符 #

如果你更倾向于用RxJS操作符的风格，可以封装一个包含错误重置和错误捕获的自定义操作符：

import { EMPTY, Observable, catchError } from 'rxjs';

// 自定义错误处理操作符
private handleError() {
  return (source$: Observable<any>) => {
    this.error = undefined; // 前置重置错误
    return source$.pipe(
      catchError((error) => {
        this.error = error;
        return EMPTY; // 这里根据需求返回空流或其他默认值，避免订阅终止
      })
    );
  };
}

// 改造原方法
f1() {
  obs1$.pipe(this.handleError()).subscribe(() => some_code);
}

f2() {
  obs2$.pipe(this.handleError()).subscribe(() => some_other_code);
}

这种方式把逻辑融入RxJS的流处理中，更符合响应式编程的思路，适合后续需要扩展更多流处理逻辑的场景。

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