[譯] RxJS: 使用超直觀的交互圖來學習組合操做符

原文連接: blog.angularindepth.com/learn-to-co…

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

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

在開發複雜度至關高的應用時，一般數據源都不止一個。這些數據源多是多個像 Firebase 這樣的外部數據點，也多是若干個用戶與之交互的 UI 組件。序列組合 ( sequence composition ) 是一項可讓你跨多個數據源來建立複雜查詢的技術，它是經過將這些相關的多個數據流組合成單個數據流來實現的。RxJS 提供了各式各樣的操做符來幫助你完成此項任務，在本文中咱們將介紹一些最經常使用的操做符。

在本文的寫做過程當中，爲了更好地展示出全部操做符之間的區別，我設計創造了一些超級直觀的數據流動圖，這讓我幾乎成爲了一名兼職的專業動畫師。可是，全部圖表都是以 GIF 動圖的形式嵌入到本文中的，因此須要一點時間才能所有加載出來。還請耐心等待。

在本文出現的代碼中，我都將使用 pipeable 操做符，若是不熟悉的話，能夠點擊這裏查看。我還會使用一個自定義的 stream 操做符，它會以訂閱時傳入的第一個參數做爲名稱來異步地生成不斷髮出值的流。

下面是本文中用到的圖表類型的說明:

併發地合併多個流

咱們第一個要介紹的操做符就是 merge 。此操做符能夠組合若干個流，而後併發地發出每一個輸入流中的全部值。一旦輸入流中產生了值，這些值會做爲結果流的一部分而被髮出。這種過程在文檔中一般被稱之爲打平 ( flattening ) 。

當全部輸入流完成時，結果流就會完成，如何任意輸入流報錯，那麼結果流就會報錯。若是某個輸入流沒有完成的話，那麼結果流便不會完成。

若是你只是對來自多個流中全部的值感興趣，就好像它們是由一個流所產生的，而不關心值發出的順序，那麼請使用 merge 操做符。

在下面的動圖中，能夠看到 merge 操做符組合了兩個流 A 和 B ，這兩個流各自生成 3 個值，每當發出值時值便會傳遞到結果流中。

下面是與動圖配套的代碼示例:

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 200, 3, 'partial');
merge(a, b).subscribe(fullObserver('merge'));
// 還可使用實例操做符
// a.pipe(merge(b)).subscribe(fullObserver('merge'));
複製代碼

可編輯的 stackblitz 在線 demo: combining-sequences-merge.stackblitz.io

順序地鏈接多個流

接下來要介紹的操做符是 concat 。它按順序訂閱每一個輸入流併發出其中全部的值，同一時間只會存在一個訂閱。只有當前輸入流完成的狀況下才會去訂閱下一個輸入流並將其值傳遞給結果流。

當全部輸入流完成時，結果流就會完成，如何任意輸入流報錯，那麼結果流就會報錯。若是某個輸入流沒有完成的話，那麼結果流便不會完成，這意味着某些流永遠都不會被訂閱。

若是值發出的順序很重要，而且你想要傳給操做符的第一個輸入流先發出值的話，那麼請使用 concat 操做符。舉個例子，有兩個流，一個從緩存中獲取值，一個從遠程服務器獲取值。若是你想要將二者組合起來並確保緩存中的值先發出的話，就可使用 concat 。

在下面的動圖中，能夠看到 concat 操做符組合了兩個流 A 和 B ，這兩個流各自生成 3 個值，先是 A 發出的值傳遞到結果流，而後纔是 B 。

下面是與動圖配套的代碼示例:

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 200, 3, 'partial');
concat(a, b).subscribe(fullObserver('concat'));
// 還可使用實例操做符
// a.pipe(concat(b)).subscribe(fullObserver(‘concat’));
複製代碼

可編輯的 stackblitz 在線 demo: concat.stackblitz.io

讓多個流競爭

接下來要介紹的操做符 race 引入了一個至關有趣的概念。它自己並對流進行任何組合，而是選擇第一個產生值的流。一旦第一個流發出值後，其餘的流就會被取消訂閱，徹底忽略掉。

當被選中的流完成時，結果流也隨之完成，若是被選中的流報錯，那麼結果流也將報錯。一樣，若是被選中的流不完成，那麼結果流便永遠不會完成。

若是有多個提供值的流時此操做符會很是有用，舉個例子，某個產品的服務器遍及世界各地，但因爲網絡條件，延遲是不可預測的，而且差別巨大。使用 race 的話，能夠向多個數據源發送一樣的請求，隨後消費首個響應請求的結果。

在下面的動圖中，能夠看到 race 操做符組合了兩個流 A 和 B ，這兩個流各自生成 3 個值，但只有 A 的值被髮出了，由於它首先發出了值。

下面是與動圖配套的代碼示例:

const a = intervalProducer(‘a’, 200, 3, ‘partial’);
const b = intervalProducer(‘b’, 500, 3, ‘partial’);
race(a, b).subscribe(fullObserver(‘race’));
// 還可使用實例操做符
// a.pipe(race(b)).subscribe(fullObserver(‘race’));
複製代碼

可編輯的 stackblitz 在線 demo: combining-sequences-race-b-is-ignored.stackblitz.io

使用高階 observables 來組合未知數量的流

以前介紹過的操做符，不管是靜態的仍是實例的，都只能用來組合已知數量的流，但若是預先並不知道用來組合的所有流呢，該怎麼辦？實際上，這種一種很是常見的異步場景。舉個例子，某個網絡請求會根據返回值的結果來發起一些其餘的請求。

RxJS 提供了一些接收流中流的操做符，也稱之爲高階 observables 。這些操做符將接收內部流的值並按照前一章節所介紹過的組合規則來進行操做。

如何任何內部流報錯的話，這些操做符也將報錯，而且它們只能使用實例操做符。如今咱們來一個個地進行介紹。

MergeAll

此操做符會合並全部內部流發出的值，合併方式就如同 merge 操做符，是併發的。

在下面的動圖中，能夠看到高階流 H ，它會生成兩個內部流 A 和 B 。 mergeAll 操做符將合併這兩個流中的值，每當發出值時值便會傳遞到結果流中。

下面是與動圖配套的代碼示例:

const a = stream(‘a’, 200, 3);
const b = stream(‘b’, 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(mergeAll()).subscribe(fullObserver(‘mergeAll’));
複製代碼

可編輯的 stackblitz 在線 demo: merge-all.stackblitz.io.

ConcatAll

此操做符將合併全部內部流發出的值，合併方式就如同 concat 操做符，是按順序鏈接。

在下面的動圖中，能夠看到高階流 H ，它會生成兩個內部流 A 和 B 。 concatAll 操做符首先從流 A 中取值，而後再從流 B 中取值，並將全部值傳遞到結果流中。

下面是與動圖配套的代碼示例:

const a = stream(‘a’, 200, 3);
const b = stream(‘b’, 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(concatAll()).subscribe(fullObserver(‘concatAll’));
複製代碼

可編輯的 stackblitz 在線 demo: concat-all.stackblitz.io.

SwitchAll

有時候從全部內部流中接收值並不是是咱們想要的效果。在某些場景下，咱們可能只對最新的內部流中的值感興趣。一個比較好的例子就是搜索。當用戶輸入關鍵字時，就向服務器發送請求，由於請求是異步的，因此返回的請求結果是一個 observable 。在請求結果返回以前，若是用戶更新了搜索框中的關鍵字會發生什麼狀況？第二個請求將會發出，如今已經有兩個請求發送給服務器了。可是，第一次搜索的結果用戶已經再也不關心了。更有甚者，若是第一次的搜索結果要是晚於第二次的搜索結果的話 (譯者注: 好比服務器是分佈式的，兩次請求請求的不是同一個節點)，那麼用戶看到的結果將是第一次的，這會讓用戶感到困擾。咱們不想讓這種事情發生，這也正是 switchAll 操做符的用武之地。它只會訂閱最新的內部流並忽略(譯者注: 忽略 = 取消訂閱)前一個內部流。

在下面的動圖中，能夠看到高階流 H ，它會生成兩個內部流 A 和 B 。switchAll 操做符首先從流 A 中取值，當發出流 B 的時候，會取消對流 A 的訂閱，而後從流 B 中取值，並將值傳遞到結果流中。

下面是與動圖配套的代碼示例:

const a = stream(‘a’, 200, 3);
const b = stream(‘b’, 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(switchAll()).subscribe(fullObserver(‘switchAll’));
複製代碼

可編輯的 stackblitz 在線 demo: switch-all.stackblitz.io.

concatMap、 mergeMap 和 switchMap

有趣的事是映射操做符 concatMap、 mergeMap 和 switchMap 的使用頻率要遠遠高於它們所對應的處理高階 observable 的操做符 concatAll、 mergeAll 和 switchAll 。可是，若是你細想一下，它們幾乎是同樣的。全部 *Map 的操做符實際上都是經過兩個步驟來生成高階 observables 的，先映射成高階 observables ，再經過相對應的組合邏輯來處理高階 observables 所生成的內部流。

咱們先來看下以前的 meregeAll 操做符的代碼示例:

const a = stream('a', 200, 3);
const b = stream('b', 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(mergeAll()).subscribe(fullObserver('mergeAll'));
複製代碼

map 操做符生成了高階 observables ，而後 mergeAll 操做符將這些內部流的值進行合併，使用 mergeMap 能夠輕鬆替換掉 map 和 mergeAll ，就像這樣:

const a = stream('a', 200, 3);
const b = stream('b', 200, 3);
const h = interval(100).pipe(take(2), mergeMap(i => [a, b][i]));

h.subscribe(fullObserver('mergeMap'));
複製代碼

兩段代碼的結果是徹底相同的。concatMap 和 switchMap 亦是如此，請你們自行試驗。

經過配對的方式來組合流

以前的操做符都是讓咱們能夠打平多個流並將這些流中的值原封不動地傳遞給結果流，就好像全部值來自同一個流。咱們接下來要介紹的這組操做符都接收多個流做爲輸入，但不一樣之處在於它們將每一個流中的值進行配對，而後生成單個組合值(譯者注: 默認是數組)來做爲結果流中的值。

每一個操做符均可選擇性地接收一個投射函數 ( projection function ) 做爲最後的參數，該函數定義額告終果流中的值如何進行組合。在本文的示例中，我都將使用默認的投射函數，它只是簡單地經過逗號做爲分隔符將值鏈接起來。在本節的最後我將展現如何使用自定義的投射函數。

CombineLatest

第一個介紹的操做符是 combineLatest 。使用它能夠取多個輸入流中的最新值，並將這些值轉換成一個單個值傳遞給結果流。RxJS 會緩存每一個輸入流中的最新值，只有當全部輸入流都至少發出一個值後纔會使用投射函數(從以前緩存中取出最新值)來計算出結果值，而後經過結果流將計算的結果值發出。

當全部輸入流完成時，結果流就會完成，如何任意輸入流報錯，那麼結果流就會報錯。若是某個輸入流沒有完成的話，那麼結果流便不會完成。換句話說，如何任何輸入流沒發出值就完成了，那麼結果流也將完成，而且在完成的同時不會發出任何值，由於沒法從已完成的輸入流中取值放入到結果流中。還有，若是某個輸入流即不發出值，也不完成，那麼 combineLatest 將永遠不會發出值以及完成，緣由同上，它將一直等待所有的輸入流都發出值。

若是你須要對某些狀態的組合進行求值，而且當其中某個狀態發生變化時再次進行求值，則此運算符很是有用。最簡單的例子就是監控系統。每一個服務均可以用一個流來表示，流返回布爾值以標識服務是否可用。當全部服務均可用時，監控狀態會是綠燈，因此投射函數應該只是簡單地執行邏輯與操做便可。

在下面的動圖中，能夠看到 combineLatest 操做符組合了兩個流 A 和 B 。一旦全部的輸入流都至少發出一個值後，結果流會將這些值組合後發出:

下面是與動圖配套的代碼示例:

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 500, 3, 'partial');
combineLatest(a, b).subscribe(fullObserver('latest'));
複製代碼

可編輯的 stackblitz 在線 demo: combine-latest.stackblitz.io.

Zip

zip 操做符的合併方式也很是有趣，它的機制在某種程度上相似於衣服或者包上的拉鍊。它將兩個及兩個以上的輸入流中的對應值組合成一個元祖(兩個輸入流的狀況下爲一對)。它會等待全部的輸入流中都發出相對應的值後，再使用投射函數來將其轉變成單個值，而後在結果流中發出。只有從每一個輸入流中湊齊對應的新值時，結果流纔會發出值，所以若是其中一個輸入流比另外一個的值發出地更快，那麼結果值發出的速率將由兩個輸入流中的較慢的那個決定。

當任意輸入流完成時而且與之配對的值從其餘輸入流發出後，結果流也將完成。若是任意輸入流永遠不完成的話，那麼結果流也將永遠不會完成，若是任意輸入流報錯的話，結果流也將報錯。

使用 zip 操做符能夠很簡單地實現使用定時器來生成範圍值的流。下面是基礎示例，其中使用投射函數來只返回 range 流中的值:

zip(range(3, 5), interval(500), v => v).subscribe();
複製代碼

在下面的動圖中，能夠看到 zip 操做符組合了兩個流 A 和 B 。一旦相對應的值配對成功，結果流就會發出組合值:

下面是與動圖配套的代碼示例:

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

zip(a, b).subscribe(fullObserver('zip'));
複製代碼

可編輯的 stackblitz 在線 demo: zip.stackblitz.io.

forkJoin

有時候，有一組流，但你只關心每一個流中的最後一個值。一般這些流也只有一個值。舉個例子，你想要發起多個網絡請求，並只想當全部請求都返回結果後再執行操做。此操做符的功能與 Promise.all 相似。可是，若是流中的值多於一個的話，除了最後一個值，其餘都將被忽略掉。

只有當全部輸入流都完成時，結果流纔會發出惟一的一個值。若是任意輸入流不完成的話，那麼結果流便永遠不會完成，如何任意輸入流報錯的話，結果流也將報錯。

在下面的動圖中，能夠看到 forkJoin 操做符組合了兩個流 A 和 B 。當全部輸入流都完成後，結果流將每一個輸入流中的最後一個值組合起來併發出:

下面是與動圖配套的代碼示例:

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

forkJoin(a, b).subscribe(fullObserver('forkJoin'));
複製代碼

可編輯的 stackblitz 在線 demo: fork-join.stackblitz.io.

WithLatestFrom

本文最後介紹的一個操做符是 withLatestFrom 。當有一個主線流，同時還須要其餘流的最新值時，可使用此操做符。withLatestFrom 與 combineLatest 有些相似，不一樣之處在於 combineLatest 是當任意輸入流發出值時，結果流都發出新的值，而 withLatestFrom 是隻有當主線流發出值時，結果流才發出新的值。

如同 combineLatest ，withLatestFrom 會一直等待每一個輸入流都至少發出一個值，當主線流完成時，結果流有可能在完成時從未發出過值。若是主線流不完成的話，那麼結果流永遠不會完成，若是任意輸入流報錯的話，結果流也將報錯。

在下面的動圖中，能夠看到 withLatestFrom 操做符組合了兩個流 A 和 B ，B 是主線流。每次 B 發出新的值時，結果流都會使用 A 中的最新值來發出組合值:

下面是與動圖配套的代碼示例:

const a = stream('a', 3000, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

b.pipe(withLatestFrom(a)).subscribe(fullObserver('latest'));
複製代碼

可編輯的 stackblitz 在線 demo: with-latest-from.stackblitz.io.

投射函數

正如本節開始所提到的，全部經過配對的方式來組合流的操做符均可以接收一個可選的投射函數。此函數定義告終果值是如何進行轉換的。使用投射函數能夠選擇只發出某個特定輸入流中的值或者以任意方式來鏈接值:

// 返回第二個流中的值
zip(s1, s2, s3, (v1, v2, v3) => v2)
// 使用 - 做爲分隔符來鏈接值
zip(s1, s2, s3, (v1, v2, v3) => `${v1}-${v2}-${v3}`)
// 返回單個布爾值
zip(s1, s2, s3, (v1, v2, v3) => v1 && v2 && v3)
複製代碼

若是想集中看全部的動圖，請參見 Pierre Criulanscy 的這個 gist 。