今天再來介紹幾個常用的「建立/組合」類型的 operators。

combineLatest

combineLatest 跟昨天介紹過的 zip 非常像，差別在於 zip 會依序組合，而 combineLatest 會在資料流有事件發生時，直接跟目前其他資料流的「最後一個事件」組合在一起，也因此這個 operator 是 latest 結尾，另一個不同的地方是，comeintLatest 內的參數是一個 Observable 陣列，訂閱後會把陣列內的這些 Observables 組合起來；直接看看程式碼：

const sourceA$ = interval(1000).pipe(
  map(data => `A${data + 1}`)
);
const sourceB$ = interval(2000).pipe(
  map(data => `B${data + 1}`)
);
const sourceC$ = interval(3000).pipe(
  map(data => `C${data + 1}`)
);

const subscription = combineLatest([sourceA$, sourceB$, sourceC$])
	.subscribe(data => console.log(`combineLatest 範例: ${data}`));
// combineLatest 範例: A3,B1,C1
// combineLatest 範例: A4,B1,C1
// combineLatest 範例: A4,B2,C1
// combineLatest 範例: A5,B2,C1
// combineLatest 範例: A6,B2,C1
// combineLatest 範例: A6,B3,C1
// combineLatest 範例: A6,B3,C2
// combineLatest 範例: A7,B3,C2
// combineLatest 範例: A8,B3,C2
// combineLatest 範例: A8,B4,C2
// combineLatest 範例: A9,B4,C2
// combineLatest 範例: A9,B4,C3

RxJS 5 之前參數不用放在陣列內，直接一個一個放進去就好，不過後來改成用陣列了，因此不放陣列的方法也被標記為棄用囉。

從結果可以看到每次有事件發生時都會將其他 Observable 最後發生的事件值組合起來，而 A1 發生時，因為其他 Observable 還沒有任何新事件，因此沒有辦法組合，所以直到 C1 發生時，全部 Observable 都有「最後一次事件值」，才進行組合。

彈珠圖：

sourceA$: --A1--A2--A3        --A4        --A5......           
sourceB$:   ----B1            --B2        ....
sourceC$:     ------C1                          

zip(sourceA$, sourceB$, sourceC$)
              ------**        --**        --**.......
                [A3,B1,C1]  [A4,B1,C1]  
                            [A4,B2,C1] (兩個來源 Observable 同時發生事件)

combineLatest 在實務上組合各種 Observables 的時候還蠻常用的，之後實戰練習的時候再來詳細介紹一下。

程式碼：https://stackblitz.com/edit/mastering-rxjs-operator-combinelatest

forkJoin

forkJoin 會同時訂閱傳入的 Observables，直到每個 Observable 都「結束」後，將每個 Observable 的「最後一筆值」組合起來，一樣直接上程式碼：

import { interval, forkJoin } from 'rxjs';
import { map, take } from 'rxjs/operators';

const sourceA$ = interval(1000).pipe(
  map(data => `A${data + 1}`),
  take(5)
);
const sourceB$ = interval(2000).pipe(
  map(data => `B${data + 1}`),
  take(4)
);
const sourceC$ = interval(3000).pipe(
  map(data => `C${data + 1}`),
  take(3)
);

forkJoin([sourceA$, sourceB$, sourceC$]).subscribe({
  next: data => console.log(`forkJoin 範例: ${data}`),
  complete: () => console.log('forkJoin 結束')
});
// forkJoin 範例: A5,B4,C3
// forkJoin 結束

因為要等所有 Observables 都結束，因此這裡示範使用了 take() 這個 operators 會在事件發生指定次數後結束整個 Observable。

上面的程式碼，最後結束的會是sourceC$ 的 C3，此時 sourceA$ 和 sourceB 都已經結束了，事件值分別是 A5 和 B4，因此最後訂閱時會得到一個 [A5, B4, C3] 然後結束。

彈珠圖：

sourceA$: --A1--A2--A3--A4--A5|
sourceB$: ----B1  ----B2  ----B3|
sourceC$:     ------C1    ------C2    ------C3|

forkJoin(sourceA$, sourceB$, sourceC#)
              ------      ------      ------**|
                                        [A5,B3,C3]

在實務上，我們會使用 forkJoin 去平行發送多個沒有順序性的 HTTP 請求，因為 HTTP 請求只會發生一次回傳就結束，如果每個請求之前沒有順序性，那麼一起發送會是比較快可以拿到全部資料的方法，例如：

const profile$ = ajax('/me/rofile');
const friends$ = ajax('/me/friends');

forkJoin(profile$, friend$).subscribe((profile, friends) => {
  // 同時處理 profile 和 friends 資料
});

程式碼：https://stackblitz.com/edit/mastering-rxjs-operator-forkjoin

race

race 本身就有「競速」的意思，因此這個 operator 接受的參數一樣是數個 Observables，當訂閱發生時，這些 Observables 會同時開跑，當其中一個 Observable 率先發生事件後，就會以這個 Observable 為主，並退訂其他的 Observables，也就是先到先贏，其他都是輸家：

const sourceA$ = interval(1000).pipe(
  map(data => `A${data + 1}`)
);
const sourceB$ = interval(2000).pipe(
  map(data => `B${data + 1}`)
);
const sourceC$ = interval(3000).pipe(
  map(data => `C${data + 1}`)
);

const subscription = race([sourceA$, sourceB$, sourceC$])
  .subscribe(data => console.log(`race 範例: ${data}`));
// A1
// A2
// A3
// ... (因為 sourceA$ 已經先到了，其他 Observables 就退訂不處理)

彈珠圖：

sourceA$: --A1--A2--A3.....
sourceB$:   ----B1.........
sourceC$:     ------C1.....

race(sourceA$, sourceB$, sourceC$)
          --A1--A2--A3..... 
            ^ sourceA$ 先到了，因此退訂 sourceB$ 和 sourceC$

這種只有一條 Observable 可以存活的感覺，讓人想忍不住到這張圖啊：

程式碼：https://stackblitz.com/edit/mastering-rxjs-operator-race

本日小節

這兩天我們介紹了「組合/建立類型」的 operators，這類型的 opereators 都是透過不同的 Observables 來源組合成新的 Observables，讓我們能更靈活的操作各種 Observables。

• combineLatest：同時訂閱所有內部 Observables，並將內部 Observables 裡面的最後一次事件資料組合起來。
• forkJoin：同時訂閱所有內部 Observables，並將內部 Observables 「完成」前的最後一個事件資料組合起來。
• race：同時訂閱所有內部 Observables，當其中一個 Observable 先發生第一次事件後，以此 Observable 為主，並將其他 Observable 取消訂閱。

跟建立類型相關的 operators 介紹到此為止，明天開始讓我們介紹這種轉換資料流向的 operators 囉。

相關資源

• Operators - combineLatest
• Operators - forkJoin
• Operators - race