RxJS: 詳解forkJoin, zip, combineLatest之間的區(qū)別

Mr_houzi 發(fā)布于2019-08-22 13:40 / 1392人閱讀

摘要：信號(hào)會(huì)由于發(fā)射源的不同發(fā)射到多次而和僅會(huì)發(fā)射其中一個(gè)，且只發(fā)射一次，標(biāo)志著流的結(jié)束。以此類(lèi)推直到其中某個(gè)流發(fā)出結(jié)束信號(hào)，整個(gè)被合并后的流結(jié)束，不再發(fā)射數(shù)據(jù)。

前言

forkJoin, zip, combineLatest是rxjs中的合并操作符，用于對(duì)多個(gè)流進(jìn)行合并。很多人第一次接觸rxjs時(shí)往往分不清它們之間的區(qū)別，其實(shí)這很正常，因?yàn)楫?dāng)你準(zhǔn)備用來(lái)合并的流是那種只會(huì)發(fā)射一次數(shù)據(jù)就關(guān)閉的流時(shí)（比如http請(qǐng)求），就結(jié)果而言這三個(gè)操作符沒(méi)有任何區(qū)別。

const ob1 = Rx.Observable.of(1).delay(1000);
const ob2 = Rx.Observable.of(2).delay(2000);
const ob3 = Rx.Observable.of(3).delay(3000);

Rx.Observable.forkJoin(ob1, ob2, ob3).subscribe((data) => console.log(data));
Rx.Observable.zip(ob1, ob2, ob3).subscribe((data) => console.log(data));
Rx.Observable.combineLatest(ob1, ob2, ob3).subscribe((data) => console.log(data));

// [1, 2, 3]
// [1, 2, 3]
// [1, 2, 3]
// 都是在3秒的時(shí)候打印

rxjs中很多操作符功能相近，只有當(dāng)其操作的流會(huì)多次發(fā)射數(shù)據(jù)時(shí)才會(huì)體現(xiàn)出它們之間的區(qū)別，下面我們來(lái)詳細(xì)解釋forkJoin, zip, 和combineLatest。

一個(gè)基本概念

首先我們要知道，一個(gè)流（或者說(shuō)Observable序列）的生命周期中，每次發(fā)射數(shù)據(jù)會(huì)發(fā)出next信號(hào)（Notification），結(jié)束發(fā)射時(shí)會(huì)發(fā)出complete信號(hào)，發(fā)生錯(cuò)誤時(shí)發(fā)出error信號(hào)，三個(gè)信號(hào)分別對(duì)應(yīng)observer的三個(gè)方法。next信號(hào)會(huì)由于發(fā)射源的不同發(fā)射0到多次；而complete和error僅會(huì)發(fā)射其中一個(gè)，且只發(fā)射一次，標(biāo)志著流的結(jié)束。
subscribe接收一個(gè)observer對(duì)象用來(lái)處理上述三種信號(hào)，只傳入一個(gè)函數(shù)會(huì)被認(rèn)為是next方法，因此傳入subscribe的next方法會(huì)執(zhí)行0到N次，N為序列正常發(fā)射信號(hào)的次數(shù)。

forkJoin

用forkJoin合并的流，會(huì)在每個(gè)被合并的流都發(fā)出結(jié)束信號(hào)時(shí)發(fā)射一次也是唯一一次數(shù)據(jù)。假設(shè)我們有兩個(gè)流：

const ob1 = Rx.Observable.interval(1000).map(d => `ob1:$bhzhxlb`).take(3);
const ob2 = Rx.Observable.interval(2000).map(d => `ob2:$7zvrrpd`).take(2);

Rx.Observable.forkJoin(ob1, ob2).subscribe((data) => console.log(data));
// ["ob1:2", "ob2:1"]

ob1會(huì)在發(fā)射完第三個(gè)數(shù)據(jù)時(shí)停止發(fā)射，ob2會(huì)在發(fā)射完第二個(gè)數(shù)據(jù)時(shí)停止，而forkJoin合并后的流會(huì)等到ob1和ob2都結(jié)束時(shí)，發(fā)射一次數(shù)據(jù)，也就是觸發(fā)一次subscribe里的回調(diào)，接收到的數(shù)據(jù)為ob1和ob2發(fā)射的最后一次數(shù)據(jù)的數(shù)組。

zip

zip工作原理如下，當(dāng)每個(gè)傳入zip的流都發(fā)射完畢第一次數(shù)據(jù)時(shí)，zip將這些數(shù)據(jù)合并為數(shù)組并發(fā)射出去；當(dāng)這些流都發(fā)射完第二次數(shù)據(jù)時(shí)，zip再次將它們合并為數(shù)組并發(fā)射。以此類(lèi)推直到其中某個(gè)流發(fā)出結(jié)束信號(hào)，整個(gè)被合并后的流結(jié)束，不再發(fā)射數(shù)據(jù)。

const ob1 = Rx.Observable.interval(1000).map(d => `ob1:$tzp5phz`).take(3);
const ob2 = Rx.Observable.interval(2000).map(d => `ob2:$zd7xj75`).take(2);

Rx.Observable.zip(ob1, ob2).subscribe({
  next: (data) => console.log(data),
  complete: () => console.log("complete")
});
// ["ob1:0", "ob2:0"] ob1等待ob2發(fā)射數(shù)據(jù)，之后合并
// ["ob1:1", "ob2:1"] 此時(shí)ob2結(jié)束，整個(gè)合并的流也結(jié)束
// "complete"

zip和forkJoin的區(qū)別在于，forkJoin僅會(huì)合并各個(gè)子流最后發(fā)射的一次數(shù)據(jù)，觸發(fā)一次回調(diào)；zip會(huì)等待每個(gè)子流都發(fā)射完一次數(shù)據(jù)然后合并發(fā)射，之后繼續(xù)等待，直到其中某個(gè)流結(jié)束（因?yàn)榇藭r(shí)不能使合并的數(shù)據(jù)包含每個(gè)子流的數(shù)據(jù)）。

combineLatest

combineLatest與zip很相似，combineLatest一開(kāi)始也會(huì)等待每個(gè)子流都發(fā)射完一次數(shù)據(jù)，但是在合并時(shí)，如果子流1在等待其他流發(fā)射數(shù)據(jù)期間又發(fā)射了新數(shù)據(jù)，則使用子流最新發(fā)射的數(shù)據(jù)進(jìn)行合并，之后每當(dāng)有某個(gè)流發(fā)射新數(shù)據(jù)，不再等待其他流同步發(fā)射數(shù)據(jù)，而是使用其他流之前的最近一次數(shù)據(jù)進(jìn)行合并。

const ob1 = Rx.Observable.interval(1000).map(d => `ob1:$lxdljnb`).take(3);
const ob2 = Rx.Observable.interval(2000).map(d => `ob2:$v3lrppf`).take(2);

Rx.Observable.combineLatest(ob1, ob2).subscribe({
  next: (data) => console.log(data),
  complete: () => console.log("complete")
});
// ["ob1:1", "ob2:0"] ob1等待ob2發(fā)射，當(dāng)ob2發(fā)射時(shí)ob1已經(jīng)發(fā)射了第二次數(shù)據(jù)，使用ob1的第二次數(shù)據(jù)
// ["ob1:2", "ob2:0"] ob1繼續(xù)發(fā)射第三次也是最后一次數(shù)據(jù)，ob2雖然還未發(fā)射，但是可以使用它上一次的數(shù)據(jù)
// ["ob1:2", "ob2:1"] ob2發(fā)射第二次也是最后一次數(shù)據(jù)，使ob1上一次的數(shù)據(jù)。
// "complete"

本期內(nèi)容結(jié)束，下一期會(huì)繼續(xù)帶來(lái)rxjs的其他操作符或者概念詳解。