如何使用RxJS实现可按条件重置、兼顾缓存与按需加载的Observable

直接使用RxJS内置运算符即可实现需求，无需自定义运算符。

最终实现代码 #

RxJS 6.x 版本 #

const myObservable = userProfile.pipe(
  switchMap(profile => loadData(profile).pipe(
    shareReplay(1, true)
  )),
  share()
);

RxJS 7.x 及以上版本 #

const myObservable = userProfile.pipe(
  switchMap(profile => loadData(profile).pipe(
    shareReplay({
      bufferSize: 1,
      refCount: true,
      resetOnRefCountZero: false
    })
  )),
  share()
);

实现原理 #

核心思路是为每个用户身份创建独立的缓存流，切换用户时直接替换整个流实例，从根源上隔离不同用户的缓存：

• 用switchMap监听用户身份变化，每次新的用户profile下发时，创建一个全新的数据流负责该用户的数据加载
• 每个用户对应的内部流用shareReplay做1份数据缓存，配置resetOnRefCountZero: false保证即便该流的临时订阅归零，缓存也不会被清空
• 最外层加share()启用refCount机制，当全局没有任何myObservable的订阅者时，自动取消对上游userProfile的订阅，避免无订阅状态下用户切换触发不必要的数据加载

解决的问题 #

• 切换用户加载数据过程中不会返回旧用户缓存：用户切换时switchMap直接销毁旧用户的流实例，新订阅者只会等待新用户流返回的最新数据
• 无订阅时不会触发预加载：外层refCount机制保证只有存在活跃订阅时才会响应上游的用户身份变化
• 订阅归零后缓存不丢失：同一用户的内部流缓存独立保留，只要用户未切换，下次订阅直接复用缓存无需重新加载
• 用户切换时自动重置缓存：新用户对应新的流实例，旧缓存随旧流自动销毁，无需手动清理

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