RxJS实现类React-Query的Observable缓存（stale-while-revalidating）最简方案

常规的shareReplay实现的是"永久缓存"——订阅一次后就不再发起新请求，完全复用缓存值。而你要的是React-Query默认的"缓存优先、后台更新"逻辑：每次订阅先返回旧缓存（如果有），同时后台发起新请求更新缓存，后续订阅能拿到最新数据。

这里提供一种简洁的实现方式，核心利用BehaviorSubject持久化缓存，结合defer和shareReplay控制请求触发逻辑：

import { defer, BehaviorSubject, merge, tap, switchMap, shareReplay } from "rxjs";
import { fromFetch } from "rxjs/fetch";

function createStaleWhileRevalidatingSource(url) {
  // 用BehaviorSubject存储缓存，支持随时获取当前值+缓存更新通知
  const cacheStore$ = new BehaviorSubject(null);

  // 创建请求流：每次触发时发起请求，成功后更新缓存
  const fetchAndUpdateCache$ = defer(() => 
    fromFetch(url)
      .pipe(
        // 处理响应，转成JSON或抛出错误
        tap(response => {
          if (!response.ok) throw new Error(`请求失败: ${response.status}`);
        }),
        switchMap(response => response.json()),
        // 请求成功后更新缓存
        tap(data => cacheStore$.next(data))
      )
  ).pipe(
    // 确保同一时间的多个订阅只发起一次请求
    shareReplay(1)
  );

  // 组合最终的流：订阅时先返回缓存，再触发请求更新
  return defer(() => {
    const currentCache = cacheStore$.value;
    // 有缓存时：合并缓存流和请求流，先推缓存，请求完成后推新值
    // 无缓存时：直接发起请求
    return currentCache 
      ? merge(cacheStore$.asObservable(), fetchAndUpdateCache$)
      : fetchAndUpdateCache$;
  }).pipe(
    // 共享流，避免重复创建逻辑
    shareReplay(1)
  );
}

// 使用示例
const githubUsers$ = createStaleWhileRevalidatingSource("https://api.github.com/users?per_page=5");

// 第一次订阅：无缓存，直接发起请求，返回请求结果
githubUsers$.subscribe({
  next: data => console.log("第一次订阅结果:", data),
  error: err => console.error(err)
});

// 5秒后再次订阅：先返回之前的缓存，同时后台发起新请求，请求完成后推送最新数据
setTimeout(() => {
  githubUsers$.subscribe(data => console.log("第二次订阅结果（先缓存后新值）:", data));
}, 5000);

逻辑说明 #

1. BehaviorSubject缓存存储：既可以随时获取当前缓存值，又能在缓存更新时自动通知所有订阅者，完美适配"缓存更新后推新值"的需求。
2. defer触发请求：确保每次订阅都会触发新的请求（这是和shareReplay永久缓存的核心区别），但通过shareReplay(1)避免同一时间多个订阅重复发起请求。
3. merge组合流：有缓存时，先推送现有缓存作为占位符，同时后台发起请求更新缓存，请求完成后BehaviorSubject会自动推送新的缓存值，订阅者能无缝拿到最新数据；无缓存时直接发起请求。

这种实现方式逻辑清晰，没有冗余的Map操作，完全贴合你要的"简洁干净"的要求。

内容的提问来源于stack exchange，提问作者Vitor Figueredo Marques