如何实现RxJS的catchUnhandledError操作符？仅捕获未处理错误

首先，咱们得先搞清楚你当前问题的根源：你的通用错误处理(errorHandler)放在了apiQuery内部的管道末尾，这意味着所有API请求的错误会先被这个通用处理拦截，下游switchMap里的特定catchError根本没机会接触到错误。RxJS的管道是上游到下游依次处理的，内部流的错误会先流经内部流的管道，再传递到外部流，所以你现在的顺序完全反了。

接下来给你两个可行的解决方案，都能避免你提到的繁琐重复问题：

方案一：调整错误处理顺序，让通用处理成为兜底 #

这个思路最直接，核心是让特定错误处理先执行，处理不了的错误重新抛出，最后由通用错误处理兜底。

步骤1：重构apiQuery，把通用错误处理作为兜底 #

把原来apiQuery里的errorHandler改成兜底逻辑，不要直接处理所有错误，而是只处理那些下游没处理的：

import { ajax, Observable, throwError, catchError, EMPTY, of } from 'rxjs';

function apiQuery<T>(search: string, order: string, page: number): Observable<T> {
  return ajax({ url: 'url', method: 'GET' }).pipe(
    map(r => r.response as T),
    // 把通用错误处理放在这里作为兜底
    catchError((err) => {
      // 你的复杂日志逻辑
      console.error('Unhandled API error:', err);
      alert('unhandled error');
      return EMPTY;
    })
  );
}

步骤2：下游特定错误处理需重新抛出未处理的错误 #

在switchMap的内部管道里，处理特定错误后，把不属于该类型的错误重新抛出，让上游的兜底逻辑处理：

combineLatest([ page, filter ]).pipe(
  switchMap(([ page, { search, order }]) => apiQuery(search, order, page).pipe(
    map(/* 合并响应与筛选条件 */),
    catchError((err) => {
      // 仅处理HTTP 422错误
      if (err.status === 422) {
        // 你的422错误处理逻辑，比如返回默认值或触发提示
        return of(/* 处理后的结果 */);
      }
      // 不是422，重新抛出给下一个catchError或兜底逻辑
      return throwError(() => err);
    }),
    catchError((err) => {
      // 仅处理HTTP 409错误
      if (err.status === 409) {
        // 你的409错误处理逻辑
        return of(/* 处理后的结果 */);
      }
      // 不是409，重新抛出给兜底逻辑
      return throwError(() => err);
    })
  ))
);

这样一来，错误会按顺序流经：422处理 → 409处理 → 通用兜底处理，完全符合你想要的“仅捕获未处理错误”的逻辑。

方案二：实现自定义catchUnhandledError操作符 #

如果你希望有一个专门的操作符来封装兜底逻辑，可以实现一个catchUnhandledError，它的核心是：允许错误先流经下游的所有处理逻辑，只有当下游没有处理该错误时，才执行自身的处理逻辑。

自定义操作符实现 #

import { Observable, Subscriber, throwError } from 'rxjs';

function catchUnhandledError<T>(handler: (err: any) => Observable<T>): (source: Observable<T>) => Observable<T> {
  return (source) => {
    return new Observable<T>((subscriber) => {
      // 保存原始的error回调，用于最终未处理的错误
      const originalError = subscriber.error;
      
      // 替换subscriber的error方法，拦截未处理的错误
      subscriber.error = (err) => {
        // 执行兜底处理逻辑
        handler(err).subscribe({
          next: (val) => subscriber.next(val),
          complete: () => subscriber.complete(),
          // 如果兜底处理也出错，再调用原始的error回调
          error: (innerErr) => originalError.call(subscriber, innerErr)
        });
      };

      // 订阅源Observable，让错误正常流经下游
      return source.subscribe(subscriber);
    });
  };
}

使用方式 #

在apiQuery里用这个操作符替代原来的catchError，然后下游的特定处理逻辑依然保持原样（记得重新抛出未处理的错误）：

// 重构errorHandler，使用自定义操作符
function errorHandler(errorMessage?: string) {
  return <T>(source: Observable<T>): Observable<T> => source.pipe(
    catchUnhandledError(() => {
      console.error('Unhandled error:', errorMessage);
      alert('unhandled error');
      return EMPTY;
    })
  );
}

// apiQuery保持原来的结构
function apiQuery<T>(search: string, order: string, page: number): Observable<T> {
  return ajax({ url: 'url', method: 'GET' }).pipe(
    map(r => r.response as T),
    errorHandler('An error message')
  );
}

// 下游的特定处理逻辑和方案一一致
combineLatest([ page, filter ]).pipe(
  switchMap(([ page, { search, order }]) => apiQuery(search, order, page).pipe(
    map(/* 合并响应与筛选条件 */),
    catchError((err) => {
      if (err.status === 422) {
        return of(/* 处理结果 */);
      }
      return throwError(() => err);
    }),
    catchError((err) => {
      if (err.status === 409) {
        return of(/* 处理结果 */);
      }
      return throwError(() => err);
    })
  ))
);

这个操作符的原理是：它不会立即处理源Observable的错误，而是让错误正常传递给下游；只有当下游没有任何catchError处理这个错误（最终触发了subscriber.error），才会执行兜底逻辑。

如何避免代码重复？ #

如果你的项目中有很多类似的API调用，可以把特定错误处理和兜底逻辑封装成一个高阶函数：

function handleApiResponse<T>(source$: Observable<T>) {
  return source$.pipe(
    catchError((err) => {
      if (err.status === 422) {
        return of(/* 422处理 */);
      }
      return throwError(() => err);
    }),
    catchError((err) => {
      if (err.status === 409) {
        return of(/* 409处理 */);
      }
      return throwError(() => err);
    }),
    // 如果用方案一，这里直接加兜底逻辑；如果用方案二，apiQuery已经包含了
    catchError(() => {
      alert('unhandled error');
      return EMPTY;
    })
  );
}

然后在调用时直接使用：

combineLatest([ page, filter ]).pipe(
  switchMap(([ page, { search, order }]) => 
    handleApiResponse(apiQuery(search, order, page).pipe(map(/* 合并逻辑 */)))
  )
);

这样就彻底避免了重复代码，同时保持了逻辑清晰。

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