无需外部状态在RxJS管道中实现带缓存的switchMap（含长耗时场景）

直接上可运行的RxJS管道代码，所有逻辑完全封装在管道内部：

import { distinctUntilChanged, switchMap, shareReplay, withLatestFrom, map } from 'rxjs';

const createDataPipe = <TParam, TData>(fetchData: (param: TParam) => Observable<TData>) => {
  return (params$: Observable<TParam>) => {
    // 内部创建缓存流：仅当参数变化时触发API请求，缓存最新结果
    const cachedData$ = params$.pipe(
      distinctUntilChanged(),
      switchMap(param => fetchData(param)),
      shareReplay(1) // 缓存最新值，默认refCount: false，避免订阅丢失导致缓存清空
    );

    // 原始参数流每次触发时，都获取缓存流的最新值
    return params$.pipe(
      withLatestFrom(cachedData$),
      map(([_, data]) => data)
    );
  };
};

关键细节说明： #

• distinctUntilChanged()：过滤重复参数，确保只有参数真正变化时才触发新API调用，避免无效请求。
• switchMap()：参数变化时切换到新的API请求Observable；如果前一个长耗时API还在处理，会自动取消旧订阅，防止旧数据覆盖新数据。
• shareReplay(1)：缓存对应参数的最新响应结果，默认配置下无活跃订阅时缓存依然保留，不会因订阅丢失导致缓存状态异常。
• withLatestFrom(cachedData$)：让原始参数流的每个事件都关联缓存流的最新值，实现“每次参数传入（无论是否变化）都重发最新值”的需求。

更紧凑的单管道写法： #

如果不需要复用逻辑，也可以直接把所有逻辑写在同一个管道链中：

import { distinctUntilChanged, switchMap, shareReplay, withLatestFrom, map, defer } from 'rxjs';

const result$ = defer(() => {
  const cachedData$ = params$.pipe(
    distinctUntilChanged(),
    switchMap(param => fetchData(param)),
    shareReplay(1)
  );

  return params$.pipe(
    withLatestFrom(cachedData$),
    map(([_, data]) => data)
  );
});

长耗时请求的额外保障： #

如果担心switchMap的自动取消逻辑不够直观，也可以显式控制旧请求的取消：

const cachedData$ = params$.pipe(
  distinctUntilChanged(),
  switchMap((param, index) => 
    fetchData(param).pipe(
      takeUntil(params$.pipe(skip(index + 1))) // 新参数到来时取消当前请求
    )
  ),
  shareReplay(1)
);

这种写法和原生switchMap行为一致，但更清晰地展示了旧请求的取消逻辑，避免长耗时请求未完成时的状态混乱。

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