仅使用 RxJS 运算符缓存 Http 请求

Question

我正在努力实现这里描述的内容：https://www.prestonlamb.com/blog/rxjs-cache-and-refresh-in-angular

换句话说，我想在给定的时间内（比如说 1 分钟）缓存一个 observable。在给定时间之后进行订阅时，应再次检索数据并再次缓存 1 分钟。

预期结果示例：

T 00:00: Request (1) => RETRIEVE data
T 00:10: Request (2) => data from cache
T 00:35: Request (3) => data from cache
T 00:50: Request (4) => data from cache
T 01:10: Request (5) => RETRIEVE data
T 01:15: Request (6) => data from cache
T 01:30: Request (7) => data from cache
T 02:30: Request (8) => RETRIEVE data

shareReplay 操作符在给定时间缓存数据时工作正常，但在给定时间过去后我无法重新启动它。

使用 shareRelay(1, 1000) 运算符的示例：

T 00:00: Request (1) => RETRIEVE data
T 00:10: Request (2) => data from cache
T 00:35: Request (3) => data from cache
T 00:50: Request (4) => data from cache
T 01:10: Request (5) => no response
T 01:15: Request (6) => no response
T 01:30: Request (7) => no response
T 02:30: Request (8) => no response

上面的链接尝试使用捕获空结果的第一个运算符来改变这种行为。不幸的是，它无法正常工作，因为在第一次之后没有缓存数据。

这是我使用上面链接的文章得到的（下图是描述使用的代码）

我得到的结果：

T 00:00: Request (1) => RETRIEVE data
T 00:10: Request (2) => data from cache
T 00:35: Request (3) => data from cache
T 00:50: Request (4) => data from cache
T 01:10: Request (5) => RETRIEVE data
T 01:15: Request (6) => RETRIEVE data
T 01:30: Request (7) => RETRIEVE data
T 02:30: Request (8) => RETRIEVE data

我还看到了一些有关计时器运算符的示例，但在这种情况下，即使没有订阅，数据也会每分钟检索一次。我不想每分钟刷新一次数据，我想每分钟都过期缓存。不幸的是，我丢失了计时器操作符的代码，但结果是这样的：

计时器操作符的结果：

T 00:00: Request (1) => RETRIEVE data
T 00:10: Request (2) => data from cache
T 00:35: Request (3) => data from cache
T 00:50: Request (4) => data from cache
T 01:00: NO REQUEST => RETRIEVE data
T 01:10: Request (5) => data from cache
T 01:15: Request (6) => data from cache
T 01:30: Request (7) => data from cache
T 02:00: NO REQUEST => RETRIEVE data
T 02:30: Request (8) => data from cache

任何拥有“纯”RxJS 解决方案来做我想做的事的人？

Answer 1

我认为您提供链接的解决方案有一个小错误，正如我试图在此 StackBlitz 中强调的那样。 （或者我可能误解了这个想法）

你可以试试这个：

const refetchSbj = new Subject();
const refetchData$ = refetchSbj.pipe(
    switchMap(() => service.fetchData())
  ).pipe(share());

merge(
  src$,
  refetchData$
).pipe(
  shareReplay(1, 1000),
  buffer(concat(timer(0), refetchData$)),
  tap(values => !values.length && refetchSbj.next()),
  filter(values => values.length !== 0),
  // In case there is only one value,
  map(([v]) => v),
  // Might want to add this, because each subscriber will receive the value emitted by the `shareReplay`
  take(1)
)

shareReplay 内部使用ReplaySubject，它将所有缓存值同步发送给新订阅者。 timer(0) 类似于setTimeout(fn, 0)，但这里的重要方面是它是异步，它允许buffer 收集ReplaySubject 发出的值。

buffer(concat(timer(0), refetchData$)), - 我们要确保提供给buffer 的内部可观察对象不完整，否则整个流将完成。 refetchData$ 会在这种情况下发出新获取的数据（我们稍后会看到）。

tap(values => !values.length && refetchSbj.next()) - 如果没有发出任何值，则意味着正在使用的ReplaySubject 没有任何值，这意味着时间已经过去。如果是这样，在refetchSbj 的帮助下，我们可以重新填充缓存。

---

这就是我们可视化流程的方式：

T 00:00: Request (1) => RETRIEVE data
1) `refetchSbj.next()`
2) shareReplay will send the value resulted from `service.fetchData()` to the subscriber
3) the newly fetched value will be added to the `buffer`, and then the `refetchData$` from `concat(timer(0), refetchData$)` will emit(this is why we've used `share()`), meaning that `values` will not be an empty array
4) take(1) is reached, the value will be sent to the subscriber and then it will complete, so the `ReplaySubject` from `shareReplay()` will have no subscribers.

T 00:10: Request (2) => data from cache
`values` will not be empty, so `refetchSbj` won't emit and `take(1)` will be reached

T 00:35: Request (3) => data from cache
T 00:50: Request (4) => data from cache
T 01:10: Request (5) => RETRIEVE data
Same as `Request (1)`
T 01:15: Request (6) => data from cache
T 01:30: Request (7) => data from cache
T 02:30: Request (8) => RETRIEVE data

Answer 2

我会考虑以下策略。

首先你创建一个函数createCachedSource，它返回一个Observable，它的缓存是通过shareReplay(1)实现的。

然后我会使用这个函数来设置一个变量source$，这是客户端用来订阅获取数据的变量。

现在的诀窍是在所需的时间间隔重置source$，再次使用createCachedSource。

代码中的所有这些概念都是这样的

// this function created an Observable cached via shareReplay(1)
const createCachedSource = () =>
  of(1).pipe(
    tap((d) => console.log(`Go out and fetch ${d}`)),
    shareReplay(1),
    tap((d) => console.log(`Return cached data ${d}`))
  );

// source$ is the Observable the clients must subscribe to
let source$ = createCachedSource();

// every 1 sec reset the cache by creating a new Observable and setting it as value of source$
interval(1000)
  .pipe(
    take(10),
    tap(() => (source$ = createCachedSource()))
  )
  .subscribe();

// here we simulate 30 subscriptions to source$, one every 300 mS
interval(300)
  .pipe(
    take(30),
    switchMap(() => source$)
  )
  .subscribe();

Answer 3

Observable const shared$ = data$.pipe(shareReplay(1, 1000)) 将缓存一个值 1000 毫秒。在那之后，它只会向未来的订阅者发送完整的通知（如果data$ 已完成）。通过检查 shared$ 是否完成，您知道缓存已过期，您必须创建一个新的 shared$ Observable 供当前和未来的订阅者使用。

您可以使用更高阶的 Observable 将这些 shared$ Observables 提供给您的订阅者。

const createShared = () => data$.pipe(shareReplay(1, 1000))
const sharedSupplier = new BehaviorSubject(createShared())

const cache = sharedSupplier.pipe(
  concatMap(shared$ => shared$.pipe(
    tap({ complete: () => sharedSupplier.next(createShared()) }),
  )),
  take(1) // emit once and complete so subscribers don't receive values from newly created shared Observables
)

https://stackblitz.com/edit/ggasua-2ntq7n

---

使用您发布的图片中的运算符，您还可以这样做：

const cache = sharedSupplier.pipe(
  concatMap(shared$ => shared$.pipe(
    first(null, defer(() => (sharedSupplier.next(createShared()), EMPTY))),
    mergeMap(d => isObservable(d) ? d : of(d))
  )),
  take(1)
)

但这是更多的代码，结果是一样的。