經過超級直觀的圖表學習合併Rxjs

內容來自於Max Koretskyi aka Wizard的《 Learn to combine RxJs sequences with super intuitive interactive diagrams》

在足夠複雜的應用程序上工做時，一般會有來自多個數據源的數據。它能夠是一些多個外部數據點。序列合成是一種技術，經過將相關流組合成一個流，能夠跨多個數據源建立複雜的查詢。RxJs提供了各類各樣的操做符，能夠幫助你作到這一點，在本文中，咱們將看看最經常使用的操做符。

下面是我將在接下里的文章裏將會用到的圖表的樣例：

同時合併多個序列

咱們將要看到的第一個操做符是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’));

// can also be used as an instance operator
a.pipe(merge(b)).subscribe(fullObserver(‘merge’));

順序鏈接多個序列

接下來咱們要講到的操做符是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'));
// can also be used as an instance operator
a.pipe(concat(b)).subscribe(fullObserver(‘concat’));

多個流競爭

接下來咱們要講到的這個操做符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’));
// can also be used as an instance operator
a.pipe(race(b)).subscribe(fullObserver(‘race’));

組合爲止數量的流和高階可觀察對象

以前講到的操做，都只能組合已知數量的流。可是若是您事先不知道全部的流，而且想要合併能夠在運行時延遲評估的流，會怎麼樣呢？事實上，這是使用異步代碼時很是常見的狀況。例如，對某些資源的網絡調用可能會致使由原始請求的結果值決定的許多其餘請求。

RxJs有咱們在上面看到的操做符的變體，這些操做符采用一系列序列，被稱爲高階Observable或Observable。

MergeAll

該運算符組合全部發出的內部流，就像普通合併同樣，同時從每一個流中生成值。

在下圖中，你將看到一個高階流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’));

ConcatAll

該運算符將全部發出的內部流組合起來，就像普通concat同樣，從每一個流中順序生成值。在下圖中，您能夠看到產生兩個內部流A和B的高階流H。串聯運算符首先從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’));

SwitchAll

有時從全部內部Observable中接收值不是咱們須要的。在某些狀況下，咱們可能只對最新內部序列中的值感興趣。搜索功能是一個很好的例子。
當用戶在輸入框輸入一些值後，咱們想服務器發送一些網絡請求，但這些網絡請求是異步的。若是用戶在返回結果以前又更新了輸入框中的值，會發生什麼？第二個網絡請求被髮送了出去，因此如今咱們已經向服務器發送了兩個搜索的網絡請求。然而，咱們對第一次搜索的結果已經不感興趣了，而且，若是將兩次搜索結果都顯示給用戶，這將不符合咱們的設想。因此咱們使用switchAll操做符，它只會訂閱最新的內部流併產生值，並忽略以前的流。

在下圖中，您能夠看到產生兩個內部流A和B的高階流H。開關操做符首先從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’));

concatMap,mergeMap,switchMap

有趣的是，這些映射操做符concatMap,mergeMap,switchMap的使用頻率比和他們相對應的concatAll,'mergeMap',switchAll要高得多。然而，若是你仔細想一想，它們幾乎是同樣的。全部的*Map操做符都是由兩個parts — producing流經過映射和使用組合邏輯，在由高階Observable產生的內部流上進行觀察。

讓咱們來看看下面熟悉的代碼，它演示了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'));

這裏的map操做符產生Observable流，mergeAll合併這些Observable流。因此咱們可使用mergeMap輕鬆替代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'));

這兩個結果是徹底同樣的。concaMap和switchMap操做也是如此。你能夠本身嘗試一下。

配對序列組合

前面的操做符容許咱們展平多個序列，並經過結果流不變地傳遞來自這些序列的值，就好像它們都來自這個序列同樣。接下來咱們要看的這組運算符仍然將多個序列做爲輸入，但不一樣之處在於它們將每一個序列的值配對，爲輸出序列產生一個組合值。

每一個運算符均可以選擇一個所謂的投影函數做爲最後一個參數，該參數定義告終果序列中的值應該如何組合。在個人示例中，我將使用默認的投影函數，該函數簡單地使用逗號做爲分隔符來鏈接值。在這一節的最後，我將展現如何提供一個定製的投影函數。

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'));

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'));

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'));

WithLatestFrom

咱們在本文中最後要看的運算符是withLatestFrom。當您有一個引導流，但還須要來自其餘流的最新值時，使用該運算符。在某種程度上，它相似於combineLatest操做符，每當任何輸入流有新的排放時，都會發出新的值。withLatestFrom只有在引導流發出值後，纔會發出新值。

正如combineLatest同樣，它仍然等待來自每一個流的至少一個發射值，而且當引導流完成時，能夠在沒有單個發射的狀況下完成。若是引導流沒有完成，它將永遠不會完成，若是任何內部流出錯，它將拋出一個錯誤。

在下圖中，您能夠看到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'));

Projection function(投影函數)

如本節開頭所述，經過配對組合值的全部運算符都採用可選的投影函數。該函數定義結果值的轉換。使用此函數，您能夠選擇只從特定的輸入序列中發出一個值，或者以任何您想要的方式鏈接值:

// return value from the second sequence
zip(s1, s2, s3, (v1, v2, v3) => v2)

// join values using dash as a separator
zip(s1, s2, s3, (v1, v2, v3) => `${v1}-${v2}-${v3}`)

// return single boolean result
zip(s1, s2, s3, (v1, v2, v3) => v1 && v2 && v3)