使用RXJS实现高级缓存

引言

在构建Web应用程序时，性能应始终是头等大事。我们可以采取许多措施来加快Angular应用程序的运行速度，例如Tree-Shaking，AoT（提前编译），懒加载模块或缓存。为了提高关于Angular应用程序性能实践方面的全面了解，我们强烈建议您查看 Minko Gechev撰写的Angular Performance Checklist。

在这篇文章中，我们专注于缓存。

实际上，缓存是提升我们的网站的性能的最有效的方法，尤其是用户在使用带宽受限或者低速网络的时候。
有几种缓存数据或资源的方法。静态资源最常用的是使用标准浏览器缓存或Service Worker。尽管Service Workers也可以缓存API请求，但它们通常对于缓存图像，HTML，JS或CSS文件等资源更为有用。为了缓存应用程序数据，我们通常使用自定义机制。
无论我们使用哪种机制，缓存通常都会提高应用程序的响应速度，降低网络成本，并具有在网络中断期间网站内容仍可用的优势。换句话说，当网站内容因缓存而距离使用者更近时，例如在客户端，请求不会引起额外的网络活动，并且缓存的数据可以更快地查找到，因为我们节省了整个网络往返的时间。

在本文中，我们将使用RxJS和Angular提供的工具开发高级缓存机制。

我认为如果你读完本文并完全理解，那么你将对RxJS的使用有深刻的认识。你也会更加清楚在你的项目里何时何地去使用RxJS。同时你会了解到许多常用的操作符具体的用法和意义。

动机

时不时地我们的脑海中经常会浮现一个问题，即如何在到处使用Observable的Angular应用程序中缓存数据。大多数人对如何使用Promises缓存数据有很好的了解，但是由于复杂性（大型API），思维方式的根本转变（从命令式到声明式）以及众多概念，在函数式/响应式编程方面会感到不知所措。因此，很难将基于Promises的现有缓存机制实际转换为Observables，尤其是如果您希望该机制更高级的话。

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

如果你对这个概念不熟悉的话，我们已经写了一篇文章Cold vs Hot Observables。

Angular的这种基于Observable的Http请求可能导致使用Observables实现缓存机制变得棘手。虽然也有简单的方法，但是通常需要大量的样板文件，可能最终可以绕过RxJS。这虽然可行，但是如果我们想要利用Observable的强大功能，不推荐用这样的方法。简单来说，我们都不会想驾驶一辆马车引擎的法拉利，对吧？

需求

在深入研究代码之前，让我们先定义这种高级缓存机制的需求。

我们想要构建一个名为World of Jokes的应用程序。这是一个简单的应用，可以随机显示给定类别的笑话。为了简单明了和集中注意力，只有一个类别。

该应用程序包含三个组件：AppComponent，DashboardComponent和JokeListComponent。

AppComponent是我们的入口点，它呈现工具栏以及根据当前路由器状态填充的
<router-outlet>。DashboardComponent仅显示类别列表。从这里，我们可以导航到JokeListComponent，然后将笑话列表呈现到屏幕上。

这些笑话本身是使用Angular的HttpClient服务从服务器中提取的。为了使组件的职责集中并且分离关注点，我们希望创建一个JokeService来处理请求数据的任务。然后，该组件可以简单地注入服务并通过其公共API访问数据。

以上所有只是我们应用程序的架构，还没有涉及缓存。

从仪表板导航到列表视图时，我们更喜欢从缓存请求数据，而不是每次都从服务器请求数据。此缓存的基础数据将每10秒更新一次。

当然，对于生产应用而言，每10秒轮询一次新数据并不是一个可靠的策略，我们宁愿使用更复杂的方法来更新缓存（例如，Web套接字推送更新）。但是，我们将在此处尝试简化操作，以专注于缓存方面。

无论如何，我们都会收到某种更新通知。对于我们的应用程序，我们希望UI（JokeListComponent）中的数据在缓存更新时不自动更新，而是等待用户强制执行UI更新。为什么？设想一个用户可能正在读其中一个笑话，但是突然之间，数据消失了，因为数据是自动更新的。那将是超级烦人和糟糕的用户体验。因此，只要有新数据可用，我们的用户就会收到通知。但是是否执行更新将由用户来决定。

为了使其更加有趣，我们希望用户也能够强制更新。这与仅更新UI不同，因为强制更新意味着先要从服务器请求数据，然后更新缓存，然后相应地更新UI。

让我们总结一下我们要构建的内容：

我们的应用包含两个组件，从组件A导航到组件B时，应优先从缓存请求B的数据，而不是每次从服务器请求B的数据
缓存每10秒更新一次
UI中的数据不会自动更新，并且需要用户强制执行更新
用户可以随时进行强制更新，这将导致网络请求，然后更新缓存和UI

最终效果会是这样的：

实现基础的缓存

让我们从简单开始，一步步到最终的，成熟的解决方案。

第一步是创建新服务。

接下来，我们将添加两个接口，一个接口描述Joke，另一个接口用于封装HTTP请求返回数据的类型。这比较符合TypeScript的代码风格，但最重要的是，它使代码开发更方便和易读。

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

现在，我们实现JokeService。我们不想显示数据是从缓存中提供还是从服务器中请求的实现细节，因此我们仅暴露一个Joke属性，返回一个获取Jokes列表的Observable。

为了执行HTTP请求，我们需要确保在服务的构造函数中注入HttpClient服务。

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<JokeResponse>(API_ENDPOINT).pipe(map(response => response.value));}
}

有了这个，我们就拥有实现Joke的getter方法的一切了。

一种简单的方法是直接返回this.requestJokes()，但这并不能解决问题。我们已经知道HttpClient公开的所有方法（例如get）都返回Code Observables。这意味着将为每个订阅者重新发射整个数据流，从而导致HTTP请求的开销。毕竟，缓存是为了加快应用程序的加载时间，并将网络请求数量限制为最小。

相反，我们想让流变得很热。不仅如此，每个新订阅者都应收到最新的缓存值。其实有一个非常方便的操作符叫做 shareReplay。此运算符返回一个Observable，该Observable共享一个对基础源（从this.requestJokes()返回的Observable）的单独订阅。

另外，shareReplay接受一个可选参数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<Array<Joke>>;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<JokeResponse>(API_ENDPOINT).pipe(map(response => response.value));}
}

我们已经讨论过上面代码里的大部分内容。但是，等等，私有属性cache$和getter里的if语句是咋回事？答案很简单。如果我们直接返回this.requestJokes().pipe(shareReplay(CACHE_SIZE))，则每个订阅者都将创建一个新的缓存实例。但是，我们希望在所有订阅者之间共享一个实例。因此，我们将实例保存在私有属性cache$中，并在首次调用getter时对其进行初始化。所有后续的订阅者都可以接收到共享实例，而无需每次都重新创建缓存。

让我们更加形象的看一下我们刚刚实现的东西：

上图描述了我们的场景中涉及的对象，即请求一个Joke列表以及这些对象之间交换消息的次序。让我们细分一下，以了解发生了什么。

我们从导航到列表组件的dashboard开始。

在初始化组件和Angular调用ngOnInit生命周期之后，我们通过调用JokeService暴露的getter函数jokes来请求jokes列表。由于这是我们第一次请求数据，因此缓存本身为空且尚未初始化，这意味着JokeService.cache$现在是undefined。在get函数内部，我们调用requestJokes()。这将为我们提供一个可以从服务器发出数据的Observable。同时，我们使用shareReplay运算符来做我们期望的事情。

shareReplay运算符会在原始源和所有将来的订阅者之间自动创建一个ReplaySubject(关于这个译者之前写过博客介绍：彻底理解RxJS里面的Observable 、Observer 、Subject
)。一旦订阅者数量从零增加到一，它将把Subject连接到基础源Observable并广播其所有值。所有将来的订阅者都将被连接到介于两者之间的那个Subject，因此实际上只对基础源Code Observable进行了一个订阅。这称为多播，它是我们实现简单缓存的基础。

一旦数据从服务器返回就会被缓存。

注意cache$在消息交换序列图中是一个独立的对象，它应该是用来代表在使用者（subscribers）和基础源（HTTP请求）之间创建的ReplaySubject的。

下一次我们在列表页面请求数据，缓存会立即重放最近的数据而且把数据发送给消费者。没有发生Http请求。

很简单，是吧？

为了真正理解这一点，让我们更进一步，看一下缓存在Observable级别的工作方式。为此，我们使用marble图来可视化观察数据流是如何工作的：

marble图清楚地表明，基础源头Observable只有一个订阅，而所有消费者都简便的订阅了共享Observable，即ReplaySubject。我们还可以看到，只有第一个订阅者触发了HTTP调用，所有其他订阅者都获得了最新值的重播。

最后，让我们看一下JokeListComponent以及如何显示数据。第一步是注入JokeService。之后，在ngOnInit内部，我们初始化属性jokes$,它是使用服务暴露的getter函数返回的值。getter会返回类型为Array<Joke>的Observable，而这正是我们想要的。

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

请注意，我们并非必须订阅jokes$。相反，我们在模板中使用了async管道，因为事实证明该管道非常好玩。好奇？查看这篇文章，
了解有关AsyncPipe的三件事。

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

Cool！这是我们的简单缓存的实现。要验证该Http请求是否仅发出一次，请打开Chrome的DevTools，点击“Network”标签，然后选择XHR。在Dashboard上开始，进入列表视图，然后来回导航。

自动更新

到目前为止，我们已经在几行代码中构建了一个简单的缓存机制。实际上，大部分繁重的工作都是由 shareReplay运算符完成的，该运算符负责缓存和重播最新值。

这样可以很好地工作，但是数据永远不会在后台实际更新。如果数据可能每隔几分钟更改一次怎么办？我们当然不希望强迫用户重新加载整个页面只是为了从服务器获取最新数据。

如果我们的缓存每10秒在后台更新一次，那会很酷吗？当然！作为用户，我们不必重新加载页面，并且如果数据已更改，则用户界面也会相应更新。同样，在实际应用程序中，我们很可能甚至不使用轮询，而是使用服务器推送通知。对于我们的小型演示应用程序，刷新间隔为10秒就可以了。

实现起来很容易。简而言之，我们想创建一个Observable，它发出以给定时间间隔隔开的一系列数据，或者简单地说，我们想每X毫秒产生一个值。为此，我们有几种选择。

第一种选择是使用interval。该运算符采用一个可选的参数period，该周期定义了每次发射之间的时间。示例：

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

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

是的，正如预期的那样。第一个值延迟了10s，这不是我们想要的。为什么？因为如果我们来到dashboard并导航至joke列表组件以阅读一些有趣的joke，那么我们将需要等待10秒钟，然后才从服务器请求数据并将其呈现到屏幕上。

我们可以通过引入另一个称为startWith(value)的运算符来解决此问题，该运算符一开始会发出给定的value作为初始值。例如interval(10000).pipe(startWith(0))。

但是我们可以做得更好。

如果我告诉你，有一个操作符可以在给定的持续时间（初始延迟）之后，然后在每个周期（间隔）之后发出一系列值呢？欢迎认识timer。

很酷，但这可以解决我们的问题吗？是的，当然可以。如果将initialDelay设置为零（0）并将period设置为10秒，则最终会出现与使用interval(10000).pipe(startWith(0))相同的行为，但我们仅使用了一个运算符。

让我们将其集成到现有的缓存机制中。

我们必须设置一个timer，对于每个滴答，我们都想发出一个HTTP请求以从服务器获取新数据。也就是说，对于每个滴答声，我们都需要switchMap一个Observable，订阅时去获取一个新的joke列表。使用switchMap有一个积极的影响，那就是避免多次请求形成竞争情况。这是由于该运算符的性质，它会unsubscribe之前的Observable，而仅从最近的Observable发射数据。

我们其余的缓存保持不变，这意味着我们的流仍然是多播的，并且所有订阅者共享一个基础源。

再次说明， shareReplay会将新的值广播给已有的订阅者，将最近的值重播给新的订阅者。

正如我们在上图看到的，Timer每10s发射一个值，对每个值我们都将其转换为内部Observable来获取数据。因为我们使用了switchMap，所以我们避免了竞争情况。因此消费者只接收到1和3。第二个内部Observable被跳过了因为当值到达的时候我们已经取消订阅了。

让我们将我们学习到的知识应用起来并相应的更新JokeService。

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

发送更新通知

让我们回顾一下到目前为止已完成的工作。

当我们从JokeService请求数据时，我们总是优先从缓存请求数据，而不是每次都从服务器请求数据。此缓存的基础数据每10秒刷新一次，当这种情况发生时，数据将传播到组件，导致UI自动更新。

这样不是很友好。假设我们是一个正在阅读其中一个笑话的用户，但是突然之间，数据消失了，因为用户界面已自动更新。这太烦人了，用户体验也很差。

因此，我们的用户应该优先收到有新数据可用的通知提示。换句话说，我们想让用户自己去更新UI变化。

事实上我们不必去接触服务来实现这个功能。逻辑也很简单。毕竟服务不应该关心发送通知，视图应该负责何时以及如何去更新屏幕上的数据。

首先，我们必须获取一个初始值才能向用户显示某些内容，否则屏幕将一直空白，直到第一次缓存的更新。我们稍后会明白为什么。为初始值设置数据流就像调用getter函数一样容易。另外，由于我们只对第一个值感兴趣，所以可以使用take运算符。

为了使这个逻辑可复用我们创建一个函数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的逻辑即可。

为了使其生效，我们需要一种通过DOM操作事件（尤其是通过点击按钮）创建Observable的方法。有几种方法，但是一种非常普遍的方法是使用Subject作为模板与组件类中的视图逻辑之间的桥梁。简而言之，Subject可以同时实现 Observer和Observable。Observables定义数据流并产生数据，而Observers可以订阅可观察对象并接收数据。

Subject的好处是，我们可以简单地在模板中使用事件绑定，然后在触发事件时调用next。这时它会将指定的值广播到所有正在监听值的观察者。注意，如果Subject为void类型，我们也可以省略该值，这对于我们的情况正好是对的。

关于Subject的内容，在之前译者的文章彻底理解RxJS里面的Observable 、Observer 、Subject
中也都解释过。

我们继续，实例化一个新的Subject：

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

现在我们可以在模版里面使用它：

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

看到我们如何使用事件绑定语法捕获<button>上的click事件了吗？当我们单击按钮时，我们只是传播一个虚值，使所有正常的观察者得到通知。之所以称其为虚值，是因为我们实际上没有传入任何值，或者至少没有传入void类型的值。

另一种方法是将@ViewChild()装饰器与RxJS的fromEvent操作符结合使用。但是，这需要我们直接操作DOM并从视图中查询HTML元素。使用Subject，我们实际上只是在两个方面架起了桥梁，除了我们要添加到按钮上的事件绑定之外，根本没有接触DOM。

好的，现在视图这边的工作都搞好了，让我们来切换到负责更新UI的逻辑。

所以更新UI意味着什么呢？缓存会在后台自动更新，当我们点击那个按钮的时候我们希望渲染缓存中的最新值，对吧？这意味着在这种情况下，我们的源数据流是Subject。 每次在update$上广播一个值时，也就是每次点击’Update’按钮时，我们都希望将此次发出的值映射到一个Observable上，从而为我们提供最新的缓存值。

之前我们已经知道switchMap操作符可以完全解决这种问题。这次我们使用mergeMap来代替它。这个操作符和switchMap非常相似，区别在于它不会取消订阅先前投影的内部Observable，而只是将内部发射合并到输出Observable中。

也就是this.update$.pipe(mergeMap(()=>this.getDataOnce()));。而getDataOnce方法也就是this.jokeService.jokes.pipe(take(1));

实际上，当从缓存中请求最新值时，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()));...}...
}

Cool！对于每次“更新”，我们都使用我们之前实现的getDataOnce方法从缓存中请求最新值。

从这里之后我们只差一小步就能拿到需要在屏幕上渲染的Jokes列表。我们要做的就是把初始的Jokes列表和我们的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<Joke>>;update$ = new Subject<void>();...ngOnInit() {const initialJokes$ = this.getDataOnce();const updates$ = this.update$.pipe(mergeMap(() => this.getDataOnce()));this.jokes$ = merge(initialJokes$, updates$);...}...
}

重要的是，我们使用了getDataOnce()使得每个更新事件发生时，都会去获取最新的缓存值。如果我们还记得的话，getDataOnce在内部使用take(1)，它只取第一个值然后就会结束流。这很关键，否则我们将获得持续不断的流或与缓存的实时连接，这会导致我们只希望通过单击“更新”按钮来执行UI更新的逻辑无法实现。

另外，由于底层缓存是多播的，因此始终重新订阅该缓存以获取最新值是完全没有问题的。

在我们继续实现Notification数据流之前，我们先来花一点时间使用marble图看看我们已经实现的东西。

如上图所示，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$监听缓存发出的所有值，但是跳过了第一个值，因为它不是刷新。对于initialNotifications$上发出的每个新值，我们将其映射为true以显示通知。一旦我们单击通知中的“更新”按钮，就会在update$上生成一个值，我们可以简单地将其映射为false，从而使通知消失。

我们在JokeListComponent 的模版里面使用showNotification$来做为一个开关，显示或者隐藏通知。

<div class="notification" [class.visible]="showNotification$ | async">...
</div>

好极了！我们真的很接近最终解决方案。但是在继续之前，让我们尝试一下并进行现场演示。请花一些时间，并逐步逐步执行代码。

手动获取最新数据

太棒了！我们已经走了很长一段路，并且已经为我们的缓存实现了一些非常酷的功能。要结束本文并把我们的缓存提升到一个全新的水平，我们还有一件事要做。作为用户，我们希望能够在任何时间点强制进行更新。

其实并没有那么复杂，但是我们必须同时涉及组件和服务才能使其正常工作。

让我们从我们的服务开始。我们需要一个API来强制缓存去重新加载数据。从技术上讲，我们将结束当前的缓存并将其设置为null。这意味着下次我们从服务中请求数据时，我们将建立一个新的缓存，获取数据并将其存储给以后的订阅者。每次我们执行更新时，创建新的缓存都没什么大不了的，因为它会完成并最终被垃圾回收。实际上，这具有积极的作用，即我们也必须重置计时器，这是绝对需要的。假设我们已经等待了9秒钟，然后点击“获取新Joke”。我们希望数据会刷新，但是1秒钟后不会弹出通知。也就是说，我们希望重新启动timer，以便在强制执行更新之后又需要10秒钟才触发自动更新。

销毁缓存的另一个原因是，与保持缓存始终运行的机制相比，它的复杂度要低得多。因为如果是缓存始终运行，则缓存需要知道是否执行了重新加载。

让我们创建一个Subject类reload$用来告诉缓存可以结束了。我们将利用takeUntil操作符将其放入我们的cache$流中。此外，我们实现了在内部将缓存设置为null，并在reload$上广播事件的forceReload方法。

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 instancethis.reload$.next();// Setting the cache to null will create a new cache the// next time 'jokes' is calledthis.cache$ = null;}...
}

仅此一项还不够，所以让我们继续在JokeListComponent中使用服务里的forceReload。为此，我们将实现一个forceReload()函数，只要我们单击“FETCH NEW JOKES”的按钮，就会调用该函数。此外，我们需要创建一个Subject，用作EventBus以更新UI并显示通知。

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

通过此操作，我们可以连接JokeListComponent模板中的按钮，以强制重新加载数据。我们要做的就是使用Angular的事件绑定语法监听click事件并调用forceReload()。

<button class="reload-button" (click)="forceReload()" mat-raised-button color="accent"><div class="flex-row"><mat-icon>cached</mat-icon>FETCH NEW JOKES</div>
</button>

这样已经可以生效了，但前提是我们先回到仪表板，然后再回到列表视图。这当然不是我们想要的。当我们强制重新从后端加载数据时，我们希望UI立即更新。

还记得我们之前实现的流update$吗？当我们单击“更新”按钮时，会从缓存中请求最新数据this.jokeSevice.jokes.pipe(take(1))。而现在我们需要完全相同的行为，因此我们可以继续扩展此流。这意味着我们必须同时合并update$和forceReload$，因为这两个流是用于更新UI的源头。

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()));...}...
}

很简单对吧，不过我们还没完成。其实我们这样破坏了notifications$。一切都很正常，直到我们点击“Fetch new Jokes”按钮。数据在屏幕上和缓存中都已经更新了，但是当我们等待10S之后并没有出现通知框。因为强制更新会销毁缓存实例(forceReload方法里的this.cache$ = null;)，意味着我们在组件里的initialNotifications$（this.jokeService.jokes.pipe(skip(1));）也就接收不到任何值。那么我们如何修复这个问题呢？

很简单！我们监听forceReload$上的事件，并为每个值切换到新的通知流。
重要的是我们要unsubscribe上一个流。听起来这里好像很需要switchMap，不是吗？也就是this.forceReload$.pipe(switchMap(() => this.getNotifications()));

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));}...
}

每当forceReload$发出一个值时，也就是用户每次点击FETCH NEW JOKES按钮，我们就会从先前的Observable取消订阅并切换到新的通知流。 请注意，这里有一些代码是我们需要做两次的，即this.jokeService.jokes.pipe(skip(1))。我们没有重复代码，而是创建了一个函数getNotifications()，该函数仅返回一个Jokes流，但跳过第一个值。最后，我们将initialNotifications$和reload$合并到一个称为show$的流中。这个流负责在屏幕上显示通知。也无需取消订阅initialNotifications$，因为此流在下一次订阅重新创建缓存之前就已经结束了。其余的保持不变。

让我们花点时间看一下我们刚刚实现的内容的更直观的表示。

正如我们在marble图中看到的那样，initialNotifications$对于显示通知非常重要。如果我们缺少此流，则仅当我们强制更新缓存时才会看到通知。就是说，当我们自己直接从服务器请求新数据时，我们必须不断切换到新的通知流，因为先前的Observable会结束，并且不再发出值。

我们已经做到了，并使用RxJS和Angular提供的工具实现了复杂的缓存机制。回顾一下，我们的服务公开了一个流，该流向我们提供了一个Jokes列表。 HTTP请求每10秒钟定期触发一次以更新缓存。为了改善用户体验，我们显示了一条通知，以便用户自己强制更新UI。最重要的是，我们还为用户提供了一种按需直接从服务器请求新数据的方法。

太棒了！这是最终的解决方案。

最终代码文件：

app.component.ts

import { Component } from '@angular/core';
import { Router, NavigationEnd, RouterEvent } from '@angular/router';
import { Observable } from 'rxjs/Observable';
import { filter, map } from 'rxjs/operators';@Component({selector: 'my-app',templateUrl: './app.component.html',styleUrls: [ './app.component.scss' ]
})
export class AppComponent  {isRoot: Observable<boolean>;constructor(private router: Router) {}ngOnInit() {this.isRoot = this.router.events.pipe(filter(x => x instanceof NavigationEnd),map((x: RouterEvent) => x.url != '/'));}
}

app.component.html

<mat-toolbar color="primary"><mat-toolbar-row class="flex"><span class="stretch">World of Jokes</span><button mat-icon-button><mat-icon aria-label="Login">menu</mat-icon></button></mat-toolbar-row><mat-toolbar-row class="cta-row" *ngIf="isRoot | async"><button class="back-button" mat-button routerLink="/"><mat-icon aria-label="Login">keyboard_arrow_left</mat-icon>Back to Dashboard</button></mat-toolbar-row>
</mat-toolbar><router-outlet></router-outlet>

dashboard.component.ts

import { Component, OnInit } from '@angular/core';
@Component({selector: 'app-dashboard',templateUrl: './dashboard.component.html',styleUrls: ['./dashboard.component.scss']
})
export class DashboardComponent implements OnInit {constructor() { }ngOnInit() {}
}

dashboard.component.html

<div class="container"><mat-card><div class="card-content"><span class="stretch">Chuck Norris</span><a color="accent" routerLink="jokes" mat-button>VIEW</a></div></mat-card>
</div>

joke.service.ts

import { Injectable } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { Observable } from 'rxjs/Observable';
import { Subject } from 'rxjs/Subject';
import { timer } from 'rxjs/observable/timer';
import { switchMap, shareReplay, map, takeUntil } from 'rxjs/operators';export interface Joke {id: number;joke: string;categories: Array<string>;
}export interface JokeResponse {type: string;value: Array<Joke>;
}const API_ENDPOINT = 'https://api.icndb.com/jokes/random/5?limitTo=[nerdy]';
const REFRESH_INTERVAL = 10000;
const CACHE_SIZE = 1;@Injectable()
export class JokeService {private cache$: Observable<Array<Joke>>;private reload$ = new Subject<void>();constructor(private http: HttpClient) { }// This method is responsible for fetching the data.// The first one who calls this function will initiate // the process of fetching data.get jokes() {if (!this.cache$) {// Set up timer that ticks every X millisecondsconst timer$ = timer(0, REFRESH_INTERVAL);/* For each timer tick make an http request to fetch new dataWe use shareReplay(X) to multicast the cache so that all subscribers share one underlying source and do not re-create the source over and over again. We use takeUntil to completethis stream when the user forces an update.*/this.cache$ = timer$.pipe(switchMap(() => this.requestJokes()),takeUntil(this.reload$),shareReplay(CACHE_SIZE));}return this.cache$;}// Public facing API to force the cache to reload the dataforceReload() {this.reload$.next();this.cache$ = null;}// Helper method to actually fetch the jokesprivate requestJokes() {return this.http.get<JokeResponse>(API_ENDPOINT).pipe(map(response => response.value));}
}

joke-list.component.ts

import { Component, OnInit } from '@angular/core';
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';import { Memoize } from 'lodash-decorators';import { JokeService, Joke } from '../joke.service';@Component({selector: 'app-joke-list',templateUrl: './joke-list.component.html',styleUrls: ['./joke-list.component.scss']
})
export class JokeListComponent implements OnInit {jokes$: Observable<Array<Joke>>;showNotification$: Observable<boolean>;update$ = new Subject<void>();forceReload$ = new Subject<void>();constructor(private jokeService: JokeService) { }ngOnInit() {const initialJokes$ = this.getDataOnce();const updates$ = merge(this.update$, this.forceReload$).pipe(mergeMap(() => this.getDataOnce()));this.jokes$ = merge(initialJokes$, updates$);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$);}getDataOnce() {return this.jokeService.jokes.pipe(take(1));}getNotifications() {return this.jokeService.jokes.pipe(skip(1));}forceReload() {this.jokeService.forceReload();this.forceReload$.next();}@Memoize()getVotes(id: number) {return Math.floor(10 + Math.random() * (100 - 10));}
}

joke-list.component.html

<div class="notification" [class.visible]="showNotification$ | async"><span>There is new data available. Click to reload the data.</span><button mat-raised-button color="accent" (click)="update$.next()"><div class="flex-row"><mat-icon>cached</mat-icon>UPDATE</div></button>
</div><main><button class="reload-button" (click)="forceReload()" mat-raised-button color="accent"><div class="flex-row"><mat-icon>cached</mat-icon>FETCH NEW JOKES</div></button><mat-card *ngFor="let joke of jokes$ | async"><div class="joke-content flex"><span class="vote">{{ getVotes(joke.id) }}</span><span class="stretch" [innerHTML]="joke.joke"></span><button mat-icon-button><mat-icon class="heart" aria-label="Like">favorite</mat-icon></button></div></mat-card>
</main>

展望

如果您以后需要一些家庭作业，请考虑以下改进建议：

添加错误处理
将组件中的逻辑重构为服务以使其可重用