原文：Advanced caching with RxJS

作者：Dominic Elm

在构建Web应用程序时，我们应该始终把性能放在首位。我们可以做很多事情来加速Angular应用程序，比如树抖动(tree shaking)、AoT(提前编译)、延迟加载模块或缓存。为了了解提高Angular应用程序性能的实践，我们强烈建议您查看Minko Gechev的《Angular 性能检查清单》。在这篇文章中，我们关注缓存。

事实上，缓存(caching)是改善网站体验的最有效方法之一，尤其是当用户使用带宽受限的设备或速度较慢的网络时。有几种方法可以缓存数据或资源。静态资源通常使用标准浏览器缓存或 Service Worker 进行缓存。虽然 Service Worker 也可以缓存API请求，但它们通常更适合于缓存图像、HTML、JS或CSS文件等资源。为了缓存应用程序数据，我们通常使用自定义机制。无论我们使用什么机制，缓存通常都会提高应用程序的响应能力，降低网络成本，并具有在网络中断期间内容可用的优势。换句话说，当内容被缓存到离消费者更近的地方时，比如在客户端，请求不会导致额外的网络活动，并且可以更快地检索缓存的数据，因为我们节省了整个网络往返的时间。在这篇文章中，我们将使用RxJS和Angular提供的工具开发一种高级缓存机制。

动机

我们时不时会遇到这样一个问题：如何在大量使用Observable的 Angular 应用程序中缓存数据。大多数人对如何用Promise缓存数据有很好的理解，但是当涉及到函数式反应式编程时，由于其复杂性(大型API)、思维方式的根本转变(从命令式到声明式)以及大量的概念，就感到不知所措了。因此，很难将现有的基于Promise的缓存机制转换为Observable，而想用高级一点的机制就更难了。

在Angular应用程序中，我们通常通过HttpClientModule附带的HttpClient执行HTTP请求。它的所有API都是基于Observable的，这意味着像get、post、put或delete这样的方法返回一个Observable。因为Observable本质上是懒惰的，只有当我们调用subscribe时才会发出请求。但是，对同一个Observable多次调用subscribe将导致源Observable被一次又一次地重新创建，因此，对每个订阅执行一个请求。我们称之为“冷Observable”。

这里简单解释一下什么叫“冷Observable”。根据RxJS的定义，冷Observable在订阅时开始运行，也就是说，当Subscribe被调用时，可观察序列才开始向观察者推送值。这与热Observable不同，比如鼠标移动事件或股票行情，它们甚至在订阅活动之前就已经产生了值。

这种行为使实现基于Observable的缓存机制变得很棘手。如果采用简单的方法，通常需要相当数量的样板文件(boilerplate)，我们可能最终会绕过RxJS，这是可行的，但如果我们想利用Observable的能力，则不是推荐的方法。

需求

在深入研究代码之前，我们先定义我们的高级缓存机制的需求。我们想构建一个名为“笑话世界”的应用程序。这是一个简单的应用程序，随机显示给定类别的笑话(为了保持简单和集中，只有一个类别)。

这个应用程序有三个组件：AppComponent、DashboardComponent和JokeListComponent。AppComponent是我们的入口点，它呈现一个工具栏以及一个基于当前路由器状态填充的。DashboardComponent只是显示一个类别列表。从这里，我们可以导航到JokeListComponent，在屏幕上呈现笑话列表。笑话本身是从使用Angular的HttpClient服务的服务器获取的。为了保持组件的责任集中并分离关注点，我们希望创建一个负责请求数据的JokeService。然后，组件可以简单地注入服务并通过其public API访问数据。以上所有这些只是我们应用程序的体系结构，还没有涉及缓存。

接下来，从dashboard导航到列表视图时，我们会优先从缓存中请求数据，而不是每次都从服务器请求数据。这个缓存的底层数据每10秒更新一次。当然，我们可以使用更复杂的方法来更新缓存(例如web socket推送更新)，每隔10秒轮询一次新数据并不是一个可靠的策略，这里只是为了尽量保持简单，以集中讨论缓存机制。

然后，我们会收到一些更新通知。对于我们的应用程序，我们希望UI(JokeListComponent)中的数据不是在缓存更新时自动更新，而是等待用户强制执行UI更新。为什么？想象一下，如果数据自动更新，一个用户可能正在读其中一个笑话，然后突然它就消失了，那将是糟糕的用户体验。因此，当有新数据可用时，我们的用户会收到通知，而不是自动更新。另外，为了让这个应用更有趣，我们希望用户能够强制缓存更新。这与单独更新UI不同，因为强制更新意味着从服务器新请求数据，更新缓存，然后相应地更新UI。

总结一下我们的需求：

• 应用程序包含两个组件，从组件A导航到B组件应该优先从缓存请求B的数据，而不是每次从服务器请求数据

• 缓存每10秒更新一次

• UI中的数据不会自动更新，需要用户强制执行更新

• 用户可以强制进行更新，这将导致发起请求，并实际更新缓存和UI

‍‍‍‍‍app 效果图

实现基本的缓存

让我们从简单开始，逐步优化直到最终的完整解决方案。

第一步是创建一个Service。

接下来，我们将添加两个interface，一个用于描述笑话(joke)的数据结构，另一个用于强类型化HTTP请求的响应。这不仅是为了满足TypeScript，但最重要的是使用起来更加方便和明显。

export interface Joke {  id: number;  joke: string;  categories: Array;}export interface JokeResponse {  type: string;  value: Array;}

现在让我们实现JokeService。我们不想暴露数据是从缓存提供还是从服务器新请求的实现细节，因此我们只需暴露一个属性jokes，它返回一个检测笑话列表的Observable。为了执行HTTP请求，我们需要确保在Service的构造函数中注入HttpClient服务。以下是JokeService的大致结构：

import { Injectable } from '@angular/core';import { HttpClient } from '@angular/common/http';@Injectable()export class JokeService {  constructor(private http: HttpClient) { }  get jokes() {    ...  }}

接下来，我们实现一个私有方法requestJokes()，它使用HttpClient执行GET请求来检索笑话列表。

import { map } from 'rxjs/operators';@Injectable()export class JokeService {  constructor(private http: HttpClient) { }  get jokes() {    ...  }  private requestJokes() {    return this.http.get(API_ENDPOINT).pipe(      map(response => response.value)    );  }}

有了它，我们就拥有了实现jokes的getter方法所需的一切。一种简单的做法是简单地返回 this.requestJokes()，但这样没用。我们知道，HttpClient的所有public方法，例如get，都返回冷Observable。这意味着为每个订阅者重新发出整个数据流，从而导致HTTP请求的开销。而我们的应用程序想做的是利用缓存尽量减少网络请求数量并加提升载速度。所以我们想让数据流变成热Observable。不仅如此，每个新订阅服务器都应该接收最新的缓存值。有一个非常方便的运算符叫做shareReplay。此运算符返回共享订阅基础源的Observable，即this.requestJokes()返回的Observable。另外，我们还接受了一个可选的bufferSize参数。bufferSize确定重播缓冲区的最大元素个树的值，即为每个订阅服务器缓存和重播的元素数。对于我们的场景，我们只想重放最新的值，因此，将bufferSize设置为1。我们看看代码：

import { Observable } from 'rxjs/Observable';import { shareReplay, map } from 'rxjs/operators';const API_ENDPOINT = 'https://api.icndb.com/jokes/random/5?limitTo=[nerdy]';const CACHE_SIZE = 1;@Injectable()export class JokeService {  private cache$: Observable>;  constructor(private http: HttpClient) { }  get jokes() {    if (!this.cache$) {      this.cache$ = this.requestJokes().pipe(        shareReplay(CACHE_SIZE)      );    }    return this.cache$;  }  private requestJokes() {    return this.http.get(API_ENDPOINT).pipe(      map(response => response.value)    );  }}

getter中的私有cache$属性和if语句是怎么回事？答案很简单。如果我们直接返回this.requestJokes().pipe(shareReplay(CACHE_SIZE))，那么每个订阅服务器都会创建一个新的缓存实例。但是，我们希望在所有订阅服务器之间共享一个实例。因此，我们将实例保存在私有属性cache$中，并在第一次调用getter时立即初始化它。所有后续使用者都将接收共享实例，而不必每次重新创建缓存。

看看更直观的表示：

这个流程图描述了场景中涉及的对象，即请求一个笑话列表，以及对象之间交换的消息序列。我们把它分解一下以了解整个过程：

我们从仪表板(DashboardComponent)开始导航到列表组件(JokeListComponent)。在组件初始化并Angular调用ngOnInit生命周期钩子之后，我们通过调用JokeService公开的getter函数jokes来请求笑话列表。因为这是我们第一次请求数据，所以缓存本身是空的，还没有初始化，这意味着JokeService.cache$未定义。在内部我们调用requestJokes()。这将给我们一个从服务器发出数据的Observable。

同时，我们应用shareReplay操作符来获得所需的行为。shareReplay操作符自动在原始源和所有订阅服务器之间创建ReplaySubject。一旦订阅者的数量从零变为一，它就会将主体连接到Observable的底层源，并广播其所有值。所有的订阅者都将连接到中间的主体，因此实际上只有一个订阅底层的冷Observable。这被称为多播，并定义了简单缓存的基础。一旦数据从服务器返回，就会被缓存。注意，缓存在图中是一个独立的对象，表示在使用者(订阅者)和底层源(HTTP请求)之间创建的ReplaySubject。下一次我们请求列表组件的数据时，我们的缓存将重放最新的值并将其发送给使用者。不需要额外的HTTP调用。

这个过程很简单，对吧。要真正分解这个过程，让我们进一步研究一下缓存在Observable级别上是如何工作的。为此，我们使用弹珠图来呈现流的实际工作方式：

上图清楚地表明，订阅底层只有一个Observable，所有的消费者都只订阅共享的Observable，那就是replySubject。我们还可以看到，只有第一个订阅者触发HTTP调用，所有其他订阅者都重放最新的值。

最后，让我们看看JokeListComponent如何显示数据。第一步是注入JokeService。之后，在ngOnInit中，我们用它的public getter函数返回的值初始化一个属性jokes$。这将返回Array类型的Observable，这正是我们想要的。

@Component({  ...})export class JokeListComponent implements OnInit {  jokes$: Observable<Array>;  constructor(private jokeService: JokeService) { }  ngOnInit() {    this.jokes$ = this.jokeService.jokes;  }  ...}

请注意，我们并没有订阅jokes$。而是在模板中使用异步管道。

<mat-card *ngFor="let joke of jokes$ | async">...mat-card>

酷！可以前往stackblitz查看我们的简单缓存代码和效果。为了验证是不是只发了一次网络请求，可以打开Chrome开发者工具切换到“网络工具”选项卡。从仪表板开始，跳转到列表视图，然后来回导航。

自动更新数据

到目前为止，我们已经用几行代码构建了一个简单的缓存机制。实际上，大部分繁重的工作都是由shareReplay操作符完成的，它负责缓存和重放最近的值。它可以正常工作，但数据实际上从未在后台更新。如果数据可能每隔几分钟就会更改一次呢？我们当然不想为了从服务器获取最新数据而强迫用户重新加载整个页面。

为了自动更新，我们的缓存在后台每10秒更新一次。作为一个用户，我们不必重新加载页面，如果数据发生变化，用户界面将相应地更新。同样，在实际应用程序中，我们很可能甚至不使用轮询，而是使用服务器推送通知。对于我们的小演示应用程序，刷新间隔10秒就可以了。

实现起来相当容易。简而言之，我们想创建一个Observable，它发出一系列间隔一定时间的值。我们希望每X毫秒产生一个值。为此，我们有几种选择。第一个选项是使用interval。此操作符采用可选参数周期，该参数周期定义每次发射之间的时间。考虑以下示例：

import { interval } from 'rxjs/observable/interval';interval(10000).subscribe(console.log);

在这里，我们设置了一个Observable，它发出无限的整数序列，每个值每10秒发出一次。这也意味着第一个值在给定的时间间隔内有点延迟。为了更好地演示这种行为，让我们看一下interval的弹珠图。

正如所料，第一个值是延迟了的。如果我们从仪表板导航到list组件来阅读一些有趣的笑话，那么我们必须等待10秒，然后才能从服务器请求数据并将其呈现在屏幕上。这不是我们想要的交互。为了解决这个问题，我们可以通过引入另一个名为startWith(value)的操作符来解决这个问题，该操作符首先将给定的值作为初始值发出。但我们还有更好的选择！有一个操作符，它可以在给定的持续时间(初始延迟)之后发出值，然后在每个周期(常规间隔)之后发出一系列值，它就是 timer。

很酷，但这真的解决了我们的问题吗？是的。如果我们将初始延迟设置为0，并将周期设置为10秒，那么最终的行为将与使用interval(10000).pipe(startWith(0))相同，而且只需要使用一个运算符。

我们把它加入现有的缓存机制中。我们设置一个timer，并且对于每一个tick，发出一个HTTP请求，从服务器获取新的数据。也就是说，对于每一个tick，我们需要switchMap到一个Observable，在订阅时，会得到一个新的笑话列表。使用switchMap有一个积极的副作用，我们可以避免竞争条件。这是该运算符的性质决定的，因为它将取消订阅先前的Observable，只从最近的Observable中发出值。

缓存的其余部分保持不变，这意味着我们的流仍然是多播的，所有订阅者共享一个底层源。同样，shareReplay的性质将向现有订阅者广播新值，并将最新值重播给新订户。

正如我们在上图中看到的，timer每10秒发出一个值。对于每一个值，我们都会switchMap到一个内部的Observable来获取我们的数据。因为我们使用的是switchMap，所以我们避免了竞争条件，因此消费者只收到值1和3。当我们已经从第2个Observable中取消订阅时，我们已经从第2个值中跳过了。

我们相应地更新JokeService：

import { timer } from 'rxjs/observable/timer';import { switchMap, shareReplay } from 'rxjs/operators';const REFRESH_INTERVAL = 10000;@Injectable()export class JokeService {  private cache$: Observable<Array>;  constructor(private http: HttpClient) { }  get jokes() {    if (!this.cache$) {      // Set up timer that ticks every X milliseconds      const timer$ = timer(0, REFRESH_INTERVAL);      // For each tick make an http request to fetch new data      this.cache$ = timer$.pipe(        switchMap(_ => this.requestJokes()),        shareReplay(CACHE_SIZE)      );    }    return this.cache$;  }  ...}

厉害！想自己试试吗？请前往stackblitz查看演示。在仪表板中，转到列表组件，然后观察神奇的发生。等几秒钟，这样你就可以看到正在运行的更新。记住，缓存每10秒刷新一次，但请随意调整刷新间隔。

发送更新通知

让我们回顾一下我们目前的实现。当我们从JokeService请求数据时，我们优先从缓存中请求数据，而不是每次都从服务器请求数据。此缓存的底层数据每10秒刷新一次，此时数据将传播到组件，从而使UI自动更新。

这样的实现很不优雅。假设我们是一个用户，正在阅读其中一个笑话，突然间它就不见了，因为用户界面是自动更新的。这太烦人了，用户体验也不好。因此，当有新数据可用时，我们的用户应该接收通知而不是直接更新。换句话说，我们希望由用户强制执行UI更新。

事实证明，我们不必为了实现这一点而修改我们的Service。逻辑很简单。毕竟，我们的Service不必担心发送通知，视图应该负责何时以及如何更新屏幕上的数据。首先，我们必须获得一个初始值来向用户显示一些内容，否则屏幕直到缓存第一次更新之前都是空白的。为初始值设置流与调用getter函数一样简单。另外，由于我们只对第一个值感兴趣，所以可以使用take运算符。为了使这个逻辑可重用，我们创建了一个helper方法getDataOnce()。

import { take } from 'rxjs/operators';@Component({  ...})export class JokeListComponent implements OnInit {  ...  ngOnInit() {    const initialJokes$ = this.getDataOnce();    ...  }  getDataOnce() {    return this.jokeService.jokes.pipe(take(1));  }  ...}

根据我们的需求，我们知道我们只想在用户真正执行更新时更新UI，而不是自动更改。用户如何执行您要求的更新？我们在UI中加一个按钮“更新”，此按钮与通知一起显示。现在，我们不必担心通知，而是关注单击该按钮时更新UI的逻辑。为了实现这一点，我们需要一种从DOM事件创建Observer的方法，特别是通过单击按钮。有几种方法，但最常见的方法是使用Subject作为模板和组件类中的视图逻辑之间的桥梁。简而言之，Subject是同时实现Observer和Observable类型的类型。Observable定义数据流并生成数据，而Observer可以订阅Observable值并接收数据。Subject的好处是我们可以在模板中使用事件绑定，然后在事件被触发时调用next。这将使将指定的值广播给所有正在侦听值的观察器。注意，如果Subject是void类型，我们也可以省略该值。事实上，我们的情况也是如此。

我们实例化一个Subject：

import { Subject } from 'rxjs/Subject';@Component({  ...})export class JokeListComponent implements OnInit {  update$ = new Subject<void>();  ...}

然后绑定到模板：

<div class="notification">  <span>There's new data available. Click to reload the data.span>  <button mat-raised-button color="accent" (click)="update$.next()">    <div class="flex-row">      <mat-icon>cachedmat-icon>      UPDATE    div>  button>div>

看到怎么使用事件绑定语法来捕获上的click事件了吗？当我们点击按钮时，我们只需传播一个虚值来通知所有活动的Observer。我们称它为虚值(ghost value)是因为我们实际上没有传入任何值，或者至少只传递了一个void类型的值。还有另一种方法，是将@ViewChild()修饰符与RxJS中的fromEvent运算符结合使用。然而，这需要我们跟DOM搅合在一起，并从视图中查询HTML元素。而使用Subject，我们实际上只是把两边连接在一起，除了添加到按钮的事件绑定之外，根本不涉及DOM。

好的，设置完视图，我们现在可以处理负责更新UI的逻辑了。那么更新用户界面意味着什么呢？缓存在后台自动更新，我们希望在单击该按钮时呈现缓存中最新的值，这意味着数据流的源是Subject。每次在update$上广播一个值，我们都希望将这个值映射到一个Observable，它给我们提供最新的缓存值。换句话说，我们处理的是一个所谓的高阶Observable，一个发出Observable的Observable。根据前文，switchMap正好解决了这个问题。这次我们将改用mergeMap。该操作符的行为与switchMap非常相似，不同之处在于它不从先前映射的内部Observable中取消订阅，而只是将内部发射合并到Observable的输出中。事实上，当从缓存请求最新的值时，HTTP请求已经完成并且缓存已成功更新。因此，这里的竞态条件问题我们其实并没有解决。虽然它看起来是异步的，但实际上有点同步，因为值将在同一时间发出。

import { Subject } from 'rxjs/Subject';import { mergeMap } from 'rxjs/operators';@Component({  ...})export class JokeListComponent implements OnInit {  update$ = new Subject<void>();  ...  ngOnInit() {    ...    const updates$ = this.update$.pipe(      mergeMap(() => this.getDataOnce())    );    ...  }  ...}

太好了！对于每一个“更新”，我们使用我们先前实现的helper方法从缓存中请求最新值。现在，我们还差一小步就可以为屏幕上呈现的笑话创建流。我们所要做的就是将最初的笑话列表与我们的update$流合并。

import { Observable } from 'rxjs/Observable';import { Subject } from 'rxjs/Subject';import { merge } from 'rxjs/observable/merge';import { mergeMap } from 'rxjs/operators';@Component({  ...})export class JokeListComponent implements OnInit {  jokes$: Observable<Array>;  update$ = new Subject<void>();  ...  ngOnInit() {    const initialJokes$ = this.getDataOnce();    const updates$ = this.update$.pipe(      mergeMap(() => this.getDataOnce())    );    this.jokes$ = merge(initialJokes$, updates$);    ...  }  ...}

使用helper方法getDataOnce()将每个更新事件映射到最新的缓存值，它在内部使用take(1)，它将获取第一个值，然后完成流。这是至关重要的，因为否则我们最终将得到一个正在进行的流或到缓存的实时连接，破坏了只通过单击“更新”按钮来执行UI更新的逻辑。另外，由于缓存的底层是多映射的数据源，所以总是重新订阅缓存以获得最新值是完全安全的。

我们先将刚才实现的内容可视化为一个弹珠图。

正如上图所示，initialJokes$是至关重要的，否则我们只能在点击“更新”按钮时看到屏幕上的内容。虽然数据已经在后台每10秒更新一次，但我们还不能按这个按钮，因为按钮是通知的一部分，我们从来没有真正向用户显示它。为此，我们必须创建一个负责显示或隐藏通知的Observable。本质上，我们需要一个发射true或false的流。我们希望值在有更新时为true，在用户单击“更新”按钮时为false。此外，我们希望跳过缓存发出的第一个(初始)值，因为它不是真正的刷新。如果我们从流的角度来思考，我们可以将其分解成多个流，并将它们合并在一起，将它们变成一个单独的Observable流。最后一个流就有了显示或隐藏通知所需的行为。

import { Observable } from 'rxjs/Observable';import { Subject } from 'rxjs/Subject';import { skip, mapTo } from 'rxjs/operators';@Component({  ...})export class JokeListComponent implements OnInit {  showNotification$: Observable<boolean>;  update$ = new Subject<void>();  ...  ngOnInit() {    ...    const initialNotifications$ = this.jokeService.jokes.pipe(skip(1));    const show$ = initialNotifications$.pipe(mapTo(true));    const hide$ = this.update$.pipe(mapTo(false));    this.showNotification$ = merge(show$, hide$);  }  ...}

这里，我们监听缓存发出的所有值，但跳过第一个，因为它不是刷新。对于initialNotifications$上的每个新值，我们将其映射为true以显示通知。一旦我们点击通知中的“更新”按钮，update$上就会产生一个值，我们可以简单地将其映射为false，从而导致通知消失。

我们在JokeListComponent的模板中使用showNotification$来切换显示或隐藏通知的类。

import { Subject } from 'rxjs/Subject';import { timer } from 'rxjs/observable/timer';import { switchMap, shareReplay, map, takeUntil } from 'rxjs/operators';const REFRESH_INTERVAL = 10000;@Injectable()export class JokeService {  private reload$ = new Subject<void>();  ...  get jokes() {    if (!this.cache$) {      const timer$ = timer(0, REFRESH_INTERVAL);      this.cache$ = timer$.pipe(        switchMap(() => this.requestJokes()),        takeUntil(this.reload$),        shareReplay(CACHE_SIZE)      );    }    return this.cache$;  }  forceReload() {    // Calling next will complete the current cache instance    this.reload$.next();    // Setting the cache to null will create a new cache the    // next time 'jokes' is called    this.cache$ = null;  }  ...}

import { Subject } from 'rxjs/Subject';@Component({  ...})export class JokeListComponent implements OnInit {  forceReload$ = new Subject<void>();  ...  forceReload() {    this.jokeService.forceReload();    this.forceReload$.next();  }  ...}

class=  <div class="flex-row">    <mat-icon>cachedmat-icon>    FETCH NEW JOKES  div>button>

import { Subject } from 'rxjs/Subject';import { merge } from 'rxjs/observable/merge';import { mergeMap } from 'rxjs/operators';@Component({  ...})export class JokeListComponent implements OnInit {  update$ = new Subject<void>();  forceReload$ = new Subject<void>();  ...  ngOnInit() {    ...    const updates$ = merge(this.update$, this.forceReload$).pipe(      mergeMap(() => this.getDataOnce())    );    ...  }  ...}

import { Observable } from 'rxjs/Observable';import { Subject } from 'rxjs/Subject';import { merge } from 'rxjs/observable/merge';import { take, switchMap, mergeMap, skip, mapTo } from 'rxjs/operators';@Component({  ...})export class JokeListComponent implements OnInit {  showNotification$: Observable<boolean>;  update$ = new Subject<void>();  forceReload$ = new Subject<void>();  ...  ngOnInit() {    ...    const reload$ = this.forceReload$.pipe(switchMap(() => this.getNotifications()));    const initialNotifications$ = this.getNotifications();    const show$ = merge(initialNotifications$, reload$).pipe(mapTo(true));    const hide$ = this.update$.pipe(mapTo(false));    this.showNotification$ = merge(show$, hide$);  }  getNotifications() {    return this.jokeService.jokes.pipe(skip(1));  }  ...}