[譯] RxJS 高級緩存

時間 2019-11-08

標籤 rxjs 高級緩存简体版

原文原文鏈接

原文連接: blog.thoughtram.io/angular/201…html

本文爲 RxJS 中文社區翻譯文章，如需轉載，請註明出處，謝謝合做！git

若是你也想和咱們一塊兒，翻譯更多優質的 RxJS 文章以奉獻給你們，請點擊【這裏】github

舒適提示: 文章較長，原文中寫的是40分鐘閱讀，建議你們午後有大把空閒時間再慢慢讀來編程

開發 Web 應用時，性能始終都是重中之重。要想提高 Angular 應用的速度，咱們能夠作一些工做，好比要搖樹優化 (tree-shaking)、AoT (ahead-of-time)、模塊的懶加載以及緩存。想要對 Angular 應用的性能提高的實戰技巧有一個全面瞭解的話，咱們強烈推薦你參考由 Minko Gechev 撰寫的 Angular 性能檢測表。在本文中，咱們將專一於緩存。api

實際上，緩存是提高網站用戶體驗的最有效的方式之一，尤爲是當用戶使用寬帶受限的設備或網絡環境較差。瀏覽器

緩存數據或資源的方式有不少種。靜態資源一般都是由標準的瀏覽器緩存或 Service Workers 來進行緩存。雖然 Service Workers 也能夠緩存 API 請求，可是對於圖像、HTML、JS 或 CSS 文件等資源的緩存，它們一般更爲有用。咱們一般使用自定義機制來緩存應用的數據。緩存

不管咱們使用的是什麼機制，緩存一般都是提高應用的響應能力、減小網絡花銷，並具備內容在網絡中斷時可用的優點。換句話說，當內容被緩存的更接近消費者時，好比在客戶端，請求將不會致使額外的網絡活動，而且能夠更快速地檢索緩存數據，從而節省了網絡往返的整個過程。安全

在本文中，咱們將使用 RxJS 和 Angular 提供的工具來開發一個高級緩存機制。服務器

動機

不時地就會有人問，如何在大量使用 Observables 的 Angular 應用中緩存數據？大多數人對於如何使用 Promises 來緩存數據有不錯的理解，但當切換至響應式編程時，便會由於它的複雜度 (龐大的 API)、思惟轉化 (從命令式到聲明式) 和衆多概念而感到不知所措。所以，很難將一個基於 Promises 的現有緩存機制轉換成基於 Observables 的，當你想要緩存機制變得更高級點時更是如此。網絡

在 Angular 應用中一般使用 HttpClientModule 中的 HttpClient 來執行 HTTP 請求。HttpClient 的全部 API 都是基於 Observable 的，也就是說像 get 、post 、put 或 delete 等方法返回的都是 Observable 。由於 Observables 天生是惰性的，因此只有當咱們調用 subscribe 時纔會真正發起請求。可是，對同一個 Observable 調用屢次 subscribe 會致使源 Observable 一遍又一遍地從新建立，每一個訂閱 (subscription) 上執行一個請求。咱們稱之爲冷的 Observables 。

若是你對此徹底沒有概念的話，咱們以前寫過一篇此主題的文章: 冷的 vs 熱的 Observables 。(譯者注: 想了解冷的 vs 熱的 Observables，還能夠推薦閱讀這篇文章)

這種行爲將致使使用 Observables 來實現緩存機制變得很棘手。簡單的方法每每就須要至關數量的樣板代碼, 咱們可能會選擇繞過 RxJS, 這也是可行的，但若是咱們想要最終駕馭 Observables 的強大力量時，這種方式是不推薦的。說白了就是咱們不想開配備小型摩托車引擎的法拉利，對吧？

需求

在深刻代碼以前，咱們先來爲要實現的高級緩存機制制定需求。

咱們想要開發的應用名爲笑話世界。這是一個簡單的應用，它只是根據制定的分類來隨機展現笑話。爲了讓應用更簡單、更專一，咱們只設定一個分類。

應用有三個組件: AppComponent、 DashboardComponent 和 JokeListComponent 。

AppComponent 組件是應用的入口，它渲染工具欄和 <router-outlet>，後者會根據當前路由器狀態來填充內容。

DashboardComponent 組件只展現分類的列表。在這能夠導航至 JokeListComponent 組件，它負責將笑話列表渲染到屏幕中。

笑話是使用 Angular 的 HttpClient 服務從服務器拉取的。要保持組件的職責單一和概念分離，咱們想建立一個 JokeService 來負責請求數據。而後組件只需經過注入此服務即可以經過它的公有 API 來訪問數據。

以上就是咱們這個應用的架構，目前尚未涉及到緩存。

當從分類列表頁導航至笑話列表頁時，咱們更傾向於請求緩存中的最新數據，而不是每次都向服務器發起請求。而緩存的底層數據會每10秒鐘自動更新。

固然，對於生產級應用來講，每隔10秒輪詢新數據並不是是個好選擇，通常來講會使用一種更成熟的方式來更新緩存 (例如 Web Socket 推送更新)。但在這裏咱們將保持簡單性，以便於專一於緩存自己。

咱們將會以某種形式來接收更新通知。對於這個應用來講，當緩存更新時，咱們不想 UI (JokeListComponent) 中的數據自動更新，而是等待用戶來執行 UI 的更新。爲何要這樣作？想象一下，用戶可能正在讀某條笑話，而後忽然間由於數據的自動更新這條笑話就消失了。這樣的結果就是因爲這種較差的用戶體驗，讓用戶很生氣。所以，咱們的作法是每當有新數據時提示用戶更新。

爲了更好玩一些，咱們還想要用戶可以強制緩存更新。這與僅更新 UI 不一樣，由於強制更新意味着從服務器請求最新數據、更新緩存，而後相應地更新 UI 。

來總結下咱們將要開發的內容點:

應用有兩個組件 A 和 B，當從 A 導航至 B 時應該從緩存中請求 B 的數據，而不是每次都請求服務器
緩存每隔10秒自動更新
UI 中的數據不會自動更新，而是須要用戶執行更新操做
用戶能夠強制更新，這將會發起 HTTP 請求以更新緩存和 UI

下面是應用的預覽圖:

實現基礎緩存

咱們先從簡單的開始，而後一步步地打造出最終成熟的解決方案。

第一步是建立一個新的服務。

接下來，咱們來添加兩個接口，一個是描述 Joke 的數據結構，另外一個是用來強化 HTTP 請求響應的類型。這會讓 TypeScript 很開心，但最重要的是開發人員使用起來也更加方便和清晰。

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

export interface JokeResponse {
  type: string;
  value: Array<Joke>;
}
複製代碼

如今咱們來實現 JokeService 。對於數據是來自緩存仍是服務器，咱們並不想暴露實現的細節，所以，咱們只暴露一個 jokes 的屬性，它返回的是包含笑話列表的 Observable 。

爲了發起 HTTP 請求，咱們須要確保在服務的構造函數中注入 HttpClien 服務。

下面是 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<JokeResponse>(API_ENDPOINT).pipe(
      map(response => response.value)
    );
  }
}
複製代碼

完成這一步後，咱們只剩 jokes 的 getter 方法沒有完成了。

一個簡單的方法就是直接返回 this.requestJokes()，但這樣並不會生效。從文章開頭中咱們已經得知 HttpClient 暴露出的全部方法，例如 get 返回的是冷的 Observables 。這意味着爲每次訂戶都會從新發出整個數據流，從而致使屢次的 HTTP 請求。畢竟，緩存的理念是提高應用的加載速度並將網絡請求的數量限制到最小。

相反的，咱們想讓流變成熱的。不只如此，咱們還想讓每一個新訂閱者都接收到最新的緩存數據。有一個很是方便的操做符叫作 shareReplay 。它返回的 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)
    );
  }
}
複製代碼

Ok，上面代碼中的大部分咱們都已經討論過了。可是等下，那個私有屬性 cache$ 和 getter 方法中的 if 語句是作什麼的？答案很簡單。若是直接返回 this.requestJokes().pipe(shareReplay(CACHE_SIZE)) 的話，那麼每次訂閱都將建立一個緩存實例。但咱們想要的是全部訂閱者都共享同一個實例。所以，咱們將這個共享的實例保存在私有屬性 cache$ 中，並在首次調用 getter 方法時對其進行初始化。後續的全部消費者都將共享此實例而無需每次都從新建立緩存。

經過下面的圖來更直觀地看下咱們剛剛實現的內容:

在上圖中，咱們能夠看到描述咱們場景中所涉及到的對象的序列圖，即請求笑話列表和在對象之間交換消息的隊列。咱們分解來看，以便更好地瞭解咱們正在作什麼。

咱們從 DashboardComponent 導航至 JokeListComponent 開始提及。

組件初始化後 Angular 會調用 ngOnInit 生命週期鉤子，這裏咱們將調用 JokeService 暴露的 jokes 的 getter 方法來請求笑話列表。由於這是首次請求數據，因此緩存自己還未初始化，也就是說 JokeService.cache$ 是 undefined 。在內部咱們會調用 requestJokes()，它會返回一個將會發出服務端數據的 Observable 。同時咱們還應用了 shareReplay 操做符來獲取預期效果。

shareReplay 操做符會自動在原始數據源和全部後來的訂閱者之間建立一個 ReplaySubject 。一旦訂閱者的數量從 0 增長至 1，就會將 Subject 與底層源 Observable 進行鏈接，而後廣播出它的全部重放值。後續的全部訂閱者都將與中間人 Subject 進行鏈接，所以底層的冷的 Observable 只有一個訂閱。這就是多播，它是咱們這個簡單緩存機制的基礎。(譯者注: 想深刻了解多播，推薦這篇文章)

一旦服務端返回數據，數據就會被緩存。

注意，在序列圖中 Cache 是一個獨立的對象，它被表示成一個 ReplaySubject，它位於消費者 (訂閱者) 和底層數據源 (HTTP 請求) 之間。

當再次爲 JokeListComponent 組件請求數據時，緩存將會重放最新值並將其發送給消費者。這樣就不會再發起額外的 HTTP 請求。

很簡單，是吧？

要想了解更多細節，咱們還需更進一步，來看看在 Observable 級別緩存是如何工做的。所以，咱們將使用彈珠圖 (marble diagram) 來對流的工做原理進行可視化展現:

彈珠圖看上去十分清晰，底層的 Observable 確實只有一個訂閱，全部消費者訂閱的都是這個共享 Observable，即 ReplaySubject 。咱們還能夠看到只有第一個訂閱者觸發了 HTTP 請求，而其餘訂閱者得到的只是緩存重放的最新值。

最後，咱們來看看 JokeListComponent 以及如何展示數據。首先是注入 JokeService 。而後在 ngOnInit 生命週期中對 jokes$ 屬性進行初始化，初始值爲由服務所暴露的 getter 方法所返回的 Observable， Observable 的類型爲 Array<Joke>，這正是咱們想要的數據。

@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>
複製代碼

酷！這就是咱們的簡單緩存了。想要驗證請求是否只發起一次，能夠打開 Chrome 的開發者工具，而後點擊 Network 標籤頁並選擇 XHR 。從分類列表頁開始，導航至笑話列表頁，而後再返回分類列表頁，反反覆覆幾回。

第 1 階段在線 Demo: 點擊查看。

自動更新

到目前爲止，咱們已經經過了少許的代碼開發出了一個簡單的緩存機制，大部分的髒活都是由 shareReplay 操做符完成的，它負責緩存和重放最新值。

目前徹底能夠正常運行，可是在後臺的數據源卻永遠不會更新。若是數據可能每隔幾分鐘就發生變化怎麼辦？咱們可不想強迫用戶去刷新整個頁面才能從服務器得到最新數據。

若是咱們的緩存能夠在後臺每10秒更新一次豈不是很好？徹底贊成！做爲用戶，咱們沒必要從新加載頁面，若是數據發生變化的話，UI 會相應地更新。重申下，在真實的應用中咱們基本上不會使用輪詢，而是使用服務器推送通知。但對於咱們這個小 Demo 應用來講，間隔 10 秒的定時器已經足夠了。

實現起來也至關簡單。總而言之，咱們想要建立一個 Observable，它發出一系列根據給定時間間隔隔開的值，或者簡單點說，咱們想要每 x 毫秒就生成一個值。咱們有幾種實現方式。

第一種選擇是使用 interval 。此操做符接收一個可選參數 period，它定義了每次發出值間的時間間隔。參考下面的示例:

import { interval } from 'rxjs/observable/interval';

interval(10000).subscribe(console.log);
複製代碼

這裏咱們設置的 Observable 會發出無限序列的整數，每次發出值會間隔 10 秒。也就是說第一個值將會在 10 秒發出。爲了更好地演示，咱們來看下 interval 操做符的彈珠圖:

呃，果然如此。第一個值是「延遲」發出的，而這並不是咱們想要的效果。爲何這麼說？由於若是咱們從分類列表頁導航至笑話列表頁時，咱們必須等待 10 秒後纔會向服務器發起數據請求以渲染頁面。

咱們能夠經過引入另外一個名爲 startWith(value) 的操做符來修復此問題，這樣一開始就會先發出給定的 value，即初始值。可是，咱們能夠作的更好！

若是我告訴你還有另一個操做符，它能夠先根據給定的時間 (初始延遲) 發出值，而後再根據時間間隔 (常規的定時器) 來不停地發出值。timer 瞭解一下。

彈珠圖時刻！

酷，可是它真的解決了咱們問題了嗎？是的，沒錯。若是咱們將初始延遲設置爲 0，並將時間間隔設置爲 10 秒，這樣它的行爲就和 interval(10000).pipe(startWith(0)) 是同樣的，但卻只使用了一個操做符。

咱們來使用 timer 操做符並將其運用在咱們現有的緩存機制當中。

咱們須要設置一個定時器，而後每次時間一到就發起 HTTP 請求來從服務器拉取最新數據。也就是說，對於每一個時間點咱們都須要使用 switchMap 來切換成一個獲取笑話列表的 Observable 。使用 swtichMap 有一個好的反作用就是能夠避免條件競爭。這是因爲這個操做符的本質，它會取消對前一個內部 Observable 的訂閱，而後只發出最新內部 Observable 中的值。

咱們緩存的其他部分都保持原樣，咱們的流仍然是多播的，全部的訂閱者都共享同一個底層數據源。

一樣的，shareReplay 會將最新值發送給現有的訂閱者，併爲後來的訂閱者重放最新值。

正如在彈珠圖中所展現的，timer 每 10 秒發出一個值。每一個值都將轉換成拉取數據的內部 Observable 。由於使用的是 switchMap，咱們能夠避免競爭條件，所以消費者只會收到值 1 和 3 。第二個內部 Observable 的值會被「跳過」，這是由於當新值發出時咱們其實已經取消對它的訂閱了。

下面來將咱們剛剛所學到的應用到 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<Joke>>;

  constructor(private http: HttpClient) { }

  get jokes() {
    if (!this.cache$) {
      // 設置每 X 毫秒發出值的定時器
      const timer$ = timer(0, REFRESH_INTERVAL);

      // 每一個時間點都會發起 HTTP 請求來獲取最新數據
      this.cache$ = timer$.pipe(
        switchMap(_ => this.requestJokes()),
        shareReplay(CACHE_SIZE)
      );
    }

    return this.cache$;
  }

  ...
}
複製代碼

酷！是否想本身試試呢？常常嘗試下面的在線 Demo 吧。從分類列表頁導航至笑話列表頁，而後見證奇蹟的誕生。耐心等待幾秒後就能看見數據更新了。記住，雖然緩存是每 10 秒刷新一次，但你能夠在在線 Demo 中自由更改 REFRESH_INTERVAL 的值。

第 2 階段在線 Demo: 點擊查看。

發送更新通知

咱們來簡單回顧下到目前爲止咱們所開發的內容。

當從 JokeService 請求數據時，咱們老是但願請求緩存中的最新數據，而不是每次都請求服務器。緩存的底層數據每隔 10 秒刷新一次，數據傳播到組件後將使得 UI 自動更新。

這是有些失敗的。想象一下，咱們就是用戶，當咱們正在看某條笑話時忽然笑話就消失了，這是由於 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 定義了數據流並生成數據，而觀察者能夠訂閱 Observables 並接收數據。

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>cached</mat-icon>
      UPDATE
    </div>
  </button>
</div>
複製代碼

來看下咱們是如何使用事件綁定語法來捕獲 <button> 上的點擊事件的？當點擊按鈕時，咱們只是傳播一個幽靈值從而通知全部活動的觀察者。咱們稱之爲幽靈值是由於實際上並無傳任何值，或者說傳遞的值的類型爲 void 。

另外一種方式是使用 @ViewChild() 裝飾器和 RxJS 的 fromEvent 操做符。可是，這須要咱們在組件類中「混入」 DOM 並從視圖中查詢 HTML 元素。使用 Subject 的話，咱們只須要將二者橋接便可，除了咱們在按鈕上添加的事件綁定以外，根本不會觸及 DOM 。

好了，設置好視圖後，咱們就能夠切換至處理 UI 更新的邏輯了。

那麼更新 UI 意味着什麼？緩存是在後臺自動更新的，而咱們想要點擊按鈕時才渲染從緩存中拿到的最新值，是這樣吧？這意味着咱們的源頭流是 Subject 。每次 update$ 上發出值時，咱們就將其映射成給出最新緩存值的 Observable 。換句話說，咱們使用的是 高階 Observable ( Higher Order Observable ) ，即發出 Observables 的 Observable 。

在此以前，咱們應該知道 switchMap 正好能夠解決這種問題。但此次，咱們將使用 mergeMap 。它的行爲與 switchMap 很相似，它不會取消前一個內部 Observable 的訂閱，而是將內部 Observable 的發出值合併到輸出 Observable 中。

事實，從緩存中請求最新值時，HTTP 請求早已完成，緩存也已經成功更新。所以，咱們並不會面臨條件競爭的問題。雖然這看上去仍是異步的，但某種程度上來講，它實際上是同步的，由於值是在同一個 tick 中發出的。

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())
    );
    ...
  }
  ...
}
複製代碼

酷！每次「更新」時咱們都是從緩存中請求的最新值，而緩存使用的是咱們以前實現的輔助方法。

到這裏，還差一小步就能夠完成負責將笑話渲染到屏幕上的流。咱們所須要作的只是合併 initialJokes$ 和 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() 來將每次更新事件映射成最新的緩存值，這點很重要。回想一下，在這個方法內部使用了 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$ 來切換 class 以顯示/關閉通知。

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

耶！目前，咱們已經十分接近最終的解決方案了。在繼續前進以前，咱們來試玩下在線 Demo 。不用着急，再來一步步地過遍代碼。

第 3 階段在線 Demo: 點擊查看。

按需拉取新數據

酷！一路走來咱們已經爲咱們的緩存實現了一些很酷的功能。要結束本文並將緩存再提高一個等級的話，咱們還須要作一件事。做爲用戶，咱們想要可以在任什麼時候間點來強制更新數據。

這並無什麼複雜的，但要完成此功能咱們須要同時修改組件和服務。

咱們先從服務開始。咱們須要一個面向公衆的 API 來強制緩存重載數據。從技術上來講，咱們會完成當前緩存，並將其設置爲 null 。這意味着下次咱們從服務中請求數據時會設置一個新的緩存，它會從服務器拉取數據並保存起來以便爲後來的訂閱者服務。每次強制更新時建立一個新緩存並非什麼大問題，由於舊的緩存將會完成並最終被垃圾收集。實際上，這樣作還有一個有用的反作用，就是重置了定時器，這決定是咱們想獲得的效果。好比說，咱們等待 9 秒後點擊「強制更新」按鈕。咱們所指望的數據刷新了，但咱們不想看到 1 秒後彈出更新通知。咱們想要讓計時器從新開始，這樣當強制更新後再過 10 秒才應該觸發自動更新。

銷燬緩存的另外一個緣由是相比於不銷燬緩存的版本，它的複雜度要小得多。若是是後者的話，緩存須要知道重載數據是不是強制執行的。

咱們來建立一個 Subject，它用來通知緩存以完成。這裏咱們利用了 takeUnitl 操做符並將其加入到 cache$ 流中。此外，咱們還實現了一個公開的 API ，它使用 Subject 來廣播事件，同時將緩存設置爲 null 。

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() {
    // 調用 `next` 以完成當前緩存流
    this.reload$.next();

    // 將緩存設置爲 `null`，這樣下次調用 `jokes` 時
    // 就會建立一個新的緩存
    this.cache$ = null;
  }

  ...
}
複製代碼

光在服務中實現並無什麼做用，咱們還須要在 JokeListComponent 中來使用它。爲此，咱們將實現一個函數 forceReload()，當點擊「強制更新」按鈕時會調用此函數。此外，咱們還須要建立一個 Subject 做爲事件總線 ( Event Bus )，用於更新 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$ 嗎？當咱們點擊「更新」按鈕時，它會請求緩存中的最新數據。事實上，咱們須要的也是一樣的行爲，所以咱們能夠繼續使用並擴展此流。這意味着咱們須要合併 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())
    );
    ...
  }
  ...
}
複製代碼

就是這麼簡單，難道不是嗎？是的，但尚未結束。實際上，咱們這樣作只會「破壞」通知。在咱們點擊「強制更新」按鈕以前，一切都是好用的。一旦點擊按鈕後，屏幕和緩存中的數據依舊照常更新，但當等待了 10 秒後卻並無通知彈出。問題在於強制更新將會完成緩存流，這意味着在組件中不會再接收到值。通知流 ( initialNotifications$ ) 基本就是死掉了。這不是正確的結果，那麼咱們如何來修復它呢？

至關簡單！咱們監聽 forceReload$ 發出的事件，將其每一個發出的值都切換成一個新的通知流。這裏取消對前一個流的訂閱很重要。耳邊是否迴盪起鈴聲？就好像在告訴咱們這裏須要使用 switchMap 。

咱們來動手實現代碼！

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$ 發出值，咱們就取消對前一個 Observable 的訂閱，而後切換成一個全新的通知流。注意，這裏有一行代碼咱們須要調用兩次，就是 this.jokeService.jokes.pipe(skip(1)) 。爲了不重複，咱們建立了函數 getNotifications()，它返回笑話列表的流，但會跳過第一個值。最後，咱們將 initialNotifications$ 和 reload$ 合併成一個名爲 show$ 的流。這個流負責在屏幕上顯示通知。另外沒有必要取消 initialNotifications$ 的訂閱，由於它會在緩存從新建立以前完成。其他的都保持不變。

嗯，咱們作到了。咱們來花點時間看看咱們剛剛實現內容的彈珠圖。

正如在圖中所看見的，對於顯示通知來講，initialNotifications$ 十分重要。若是沒有這個流的話，咱們只能在強制緩存更新後纔有機會看到通知。也就是說，當咱們按需請求最新數據時，咱們必須不斷地切換成新的通知流，由於前一個(舊的) Observable 已經完成並再也不發出任何值。

就是這樣！咱們使用 RxJS 和 Angular 提供的工具實現了一個複雜的緩存機制。簡答回顧下，咱們的服務暴露出一個流，它爲咱們提供笑話列表。每隔 10 秒會觸發 HTTP 請求來更新緩存。爲了提高用戶體驗，咱們提供了更新通知，這樣用戶能夠執行更新 UI 的操做。在此之上，咱們還爲用戶提供了一種按需請求最新數據的方式。

太棒了！這就是完整的解決方案。花費幾分鐘再來看一遍代碼。而後嘗試不一樣的場景以確認是否一切都能正常運行。

第 4 階段在線 Demo: 點擊查看。