摘要：輸出流只有在所有的輸入流都完成以后才能完成，任何一條輸入流上的錯誤都將立即推送到輸出流上。如果沒有轉(zhuǎn)入輸入流，輸出流將會立即發(fā)出結(jié)束通知。返回值以數(shù)組形式獲取到的每一個輸入流的值，或者來自映射函數(shù)的值。返回值僅從最新的內(nèi)部流上取值的流。

接著上一節(jié)的操作符繼續(xù)，讓我們直奔主題。

組合類的操作符可以將不同流數(shù)據(jù)按一定的規(guī)則進(jìn)行合并，從而獲得所需要的完整數(shù)據(jù)。

combineLatest

實(shí)例方法

將多個輸入流組合成一個輸出流，輸出流上的數(shù)據(jù)是由每一條輸入流上的最后一條數(shù)據(jù)組合成的數(shù)據(jù)。

無論什么時間只要輸入流上發(fā)出了數(shù)據(jù)，它就會把所有輸入流上的最新數(shù)據(jù)組合成一條新的數(shù)據(jù)推送到輸出流上。關(guān)鍵點(diǎn)在于任意一條被組合的流發(fā)出數(shù)據(jù)，輸出流上都將有新的數(shù)據(jù)產(chǎn)生。

    -a------b----c-----d------e---|

    --1--2-----3--4-----5---6---|

            combineLatest

    --a1-a2-b2--b3-c3-c4----d4-d5---e5-e6-|

另外它可以接受一個映射函數(shù)作為最后一個參數(shù)，所有輸入流的值會依次傳入映射函數(shù)作為它的參數(shù)。

返回 Observable 一條包含所有輸入流最新值的流，或者是所有最新值的映射值的流。

示例

const weight = of(70,72,76,79,75);

const height = of(1.76,1.77,1.78);

const bmi = weight.combineLatest(height, (w, h) => w/h*h);

bmi.subscribe(x => console.log("BMI is " + x);

withLatestFrom

實(shí)例方法

將多個輸入流組合成一個輸出流，輸出流上的數(shù)據(jù)是由每一條輸入流上的最后一條數(shù)據(jù)組合成的，但是組合動作只在源 Observable 發(fā)出值時發(fā)生，這里的源 Observable 指的是調(diào)用withLatestFrom的實(shí)例。也就是說只有當(dāng)源 Observable 產(chǎn)生值時，才會產(chǎn)生新的值。

    -a-----b--------------c------------d--e---|-->

    ----1-----2-----3--4--------5----|---->

            withLatestFrom

    ------b1------------c4-----------d5--e5---|-->

返回值 Observable 一條包含所有輸入流最新值的流，或者是所有最新值的映射值的流。

和combineLatest一樣它也可以接受映射函數(shù)作為參數(shù)。

示例

const clicks = fromEvent(document, "click");
const timer = interval(1000);
const result = clicks.withLatestFrom(timer)

result.subscribe(x => console.log(x));

zip

實(shí)例方法

如果所有的輸入都是流，那么將所有輸入流對應(yīng)位置上的值組合成一個新值，將這個新值作為輸出流上的值。此外zip操作符還可以直接使用可轉(zhuǎn)化為流的數(shù)據(jù)作為參數(shù)，比如數(shù)組，promise，字符串等。

--a------b------c------d---2--|->

-----e------f------g-----h--|->

            zip

----ae------fb-----gc----hd-|->

返回值 Observable 將各輸入流上對應(yīng)的值組合成新的值發(fā)送給輸出流或者先經(jīng)過映射函數(shù)處理后再發(fā)送給輸出流。

它也可以接受映射函數(shù)作為參數(shù)。

示例

const obs1 = from([1,2,3,4]);
const obs2 = from(["a","b","c"]);

obs1.zip(obs2)
    .subscribe(v => console.log(v));

const obs = interval(1000);
const promise = new Promise(resolve => {
    setTimeout(() => resolve("hello"), 2000);
});

obs.zip(promise, (obs, promise) => promise + obs)
    .subscribe(v => console.log(v));

merge

實(shí)例方法

創(chuàng)建一條可以同時把所有輸入流的值推送出去的流。它把多條輸入流合并成一條輸入流，所有輸入流的值都可以在這條輸出流上得到。

    -a----------b----------c---|----->

    ------d---------e----------e----|-->

                merge

    -a----d----b----e-----c----e----|-->

合并所有輸入流的訂閱，不管是源 Observable 還是作為參數(shù)輸入的 Observable，只是把值依次推送到輸出流上，不作任何額外的更改。輸出流只有在所有的輸入流都完成以后才能完成，任何一條輸入流上的錯誤都將立即推送到輸出流上。

返回值 Observable 可以發(fā)送所有輸入流上的值的Observable。

示例

const clicks = fromEvent(document, "click");
const timer = interval(1000);
const clicksOrTimer = clicks.merge(timer);

clicksOrTimer.subscribe(x => console.log(x));

forkJoin

靜態(tài)方法

將輸入流的最后一個值合并后傳給輸出流。它的效果等同于Promise.all()，因此在你需要多個并發(fā)請求都返回結(jié)果時可以使用它。

forkJoin可以以參數(shù)或數(shù)組的形式接收任意數(shù)量的輸入流。如果沒有轉(zhuǎn)入輸入流，輸出流將會立即發(fā)出結(jié)束通知。

它會等所有的輸入流發(fā)出結(jié)束通知，然后發(fā)出一個包含所有輸入流發(fā)出的最終值組成的數(shù)組，因此，傳入n條輸入流，得到的數(shù)組中將包含n個元素，每一個元素都是從對應(yīng)順序的輸入流上取到的最后一個值。這也就意味著輸出流只會發(fā)出一個值，緊接著就會發(fā)出結(jié)束通知。

為了使用輸出流上獲取到的數(shù)組的長度與輸入流的數(shù)量保持一致，如果某一個輸入流在發(fā)出結(jié)束通知之前沒有發(fā)出過任何有效值，那么輸出流也將會立刻結(jié)束并且不再發(fā)出任何值，盡管其它的輸入流上可能會發(fā)出有效值。反過來，假如有一條輸入流一直沒有發(fā)出結(jié)束通知，輸出流也不會發(fā)出值，除非另外一條輸入像之前描述的那樣只發(fā)了結(jié)束通知?？偟膩碚f就是為了讓輸出流發(fā)出值，所有的輸入流都必須發(fā)出結(jié)束通知而且在此之前必須至少發(fā)出一個有效值。

如果輸入流中的某一條發(fā)出了錯誤通知，輸出流將會立即發(fā)出這個錯誤通知，并且立即取消對其它流的訂閱。

    ----2---3------4----5--|-->

    ----a----b------------d---|-->

                forkJoin

    --------------------------5d|

返回值 Observable 以數(shù)組形式獲取到的每一個輸入流的值，或者來自映射函數(shù)的值。

可以接收映射函數(shù)作為參數(shù)。

示例

const observable = forkJoin(
    of(1, 2, 3, 4),
    of(5, 6, 7, 8)
);

observable.subscribe(
    value => console.log(value),
    err => {},
    () => console.log("This is how it ends!")
);

const observable = forkJoin(
    interval(1000).take(3),
    interval(500).take(4)
);

observable.subscribe(
    value => console.log(value),
    err => {},
    () => console.log("This is how it ends!")
);

這里我們主要介紹那些可以處理高階流的操作符。

map

實(shí)例方法

輸出流上的值是使用一個映射函數(shù)將輸入流上的值映射后得到新的值。

    --------1-----------2----------3----------|--->

            map(v => v * 10);

    --------10----------20---------30---------|---->

返回值 Observable 發(fā)出映射后的值的流。

示例

fromEvent(document, "click")
    .map(event => event.clientX)
    .subscribe(v => console.log(v));

mergeMap

實(shí)例方法

將所有輸入流的值都合并到一條流上。

    -1-----2-----3---------|-->

    ---2----2----2--|-----> //第一次時的內(nèi)部流，第2，3次的一樣，這個流是直正的輸入流

            mergeMap(v => Observable.of(v + 1).repeat(3));

    -2--2--2--3--3--3--4--4--4----|-->

輸出流會把所有從映射函數(shù)中返回的內(nèi)部流打平到一條流上。映射函數(shù)可以使用輸入流的值來生成內(nèi)部流。

返回值 Observable 一條合并了所有映射函數(shù)生成的內(nèi)部流的流，內(nèi)部流上的值都會在這條流上發(fā)出。

示例

of("a","b","c")
    .mergeMap(v => Rx.Observable.interval(1000).map(x => x + v ))
    .subscribe(v => console.log(v));

const source = of("Hello");

const createPromise = v => new Promise(resolve => resolve(`I got ${v} from promise`));

source.mergeMap(
    v => createPromise(v),
    (outValue, innerValue) => `Source: ${outValue},${innerValue}`
)
.subscribe(v => console.log(v));

switchMap

實(shí)例方法

在輸入流發(fā)出值時，把它映射成一條內(nèi)部流，然后把這條內(nèi)部流打平成到輸出流上，輸出流上只會發(fā)出最近的內(nèi)部流上的值。

    -1---------3-----5----|->

    -10---10---10-| // 內(nèi)部流

            switchMap(v => Observable.from([10,10,10]).map(x => x * v))

    -10---10---10-30---30-50---50---50-|

輸出流上的值是由映射函數(shù)在輸入流的值的基本上生成的內(nèi)部流所發(fā)出的，輸出流只允許觀察一條內(nèi)部流，當(dāng)有新的內(nèi)部流到達(dá)時，輸出流將會取消對之前的內(nèi)部流的訂閱，轉(zhuǎn)而訂閱這個最新的內(nèi)部流，并發(fā)出它上面的值。

返回值 Observable 僅從最新的內(nèi)部流上取值的流。

示例

fromEvent(document, "click")
    .switchMap(event => interval(1000))
    .subscribe(v => console.log(v));

concatMap

實(shí)例方法

將多個流合并到一條流上，需要等待當(dāng)前的流完成后才能開始合并下一個，合并過程按傳入的順序進(jìn)行。

    --1------------3---------5-----------|-->

    --10---10---10--|-->

                concatMap(i => 10*i----10*i---10*i)

    --10---10---10-30---30---30-50---50---50--|->

返回值 Observable 把輸入的值經(jīng)過映射成流后再連接起來的流。

示例

fromEvent(document, "click")
    .concatMap(event => Rx.Observable.interval(1000).take(3))
    .subscribe(v => console.log(v));

groupBy

實(shí)例方法

將輸入流上的值按一定的規(guī)則分組成不同的流，然后把這些流發(fā)送給輸出流，每一條流上都是一組符合相同條件的值。

----1----2----3----4----5----|->

            groupBy(v => v%2);

-----------------------------|->
       
        2---------4---------|
      1------3----------5----|

返回值 Observable 發(fā)出分組后的流的高階流，分組的流都有一個唯一的key，并且該流中的值都是輸入流上符合某一條件的值。

示例

of(
    {id: 1, name: "aze1"},
    {id: 2, name: "sf2"},
    {id: 2, name: "dg2"},
    {id: 1, name: "erg1"},
    {id: 1, name: "df1"},
    {id: 2, name: "sf2"},
    {id: 3, name: "qfs3"},
    {id: 2, name: "qsg"}
)
.groupBy(v => v.id)
.mergeMap(group => group.reduce((acc,cur) => [...acc, cur],[]))
.subscribe(v => console.log(v))

reduce

實(shí)例方法

在源 Observable 上應(yīng)用一個累積函數(shù)，當(dāng)源 Observable 結(jié)束時把累積的值推送給輸出流，可以接受一個初始的累積值。

    --------1-------2-------3---|-->

            reduce((acc,cur) => acc + cur, 0);

    ----------------------------4|

這個操作符的行為與數(shù)組的reduce方法相同。使用一個累積函數(shù)來處理流上的每一值，前一次累積后的值將作為下一次累積的初始值，把通過累積得到的最終值推送給輸出流。注意，reduce操作符只會在源 Observable 發(fā)出結(jié)束通知后發(fā)出一個值。

源 Observable 上發(fā)出的值都會被累積函數(shù)處理。如果提供了初始值，這個值將成為整個累積過程的初始值，如果沒有提供的話，從源 Observable 發(fā)出的第一個值就是整個累積過程的初始值。

返回值 Observable 把源 Observable 上的值都經(jīng)過累積函數(shù)計(jì)算后得到的值作為值發(fā)送的流。

示例

const clicksInFiveSeconds = fromEvent(document, "click")
                            .takeUntil(interval(5000));
const ones = clicksInFiveSeconds.mapTo(1);
const seed = 0;
const count = ones.reduce((acc,cur) => acc + cur, seed);

count.subscribe(v => console.log(v));

scan

實(shí)例方法

在輸入流上應(yīng)用一個累積函數(shù)，每一次累積的結(jié)果都發(fā)送給輸出流，累積時可以接受一個可選的初始值。 和reduce操作符類似，不同的是會把每一次累積的結(jié)果都發(fā)送出去。

返回值 Observable 輸出累積值的流。

示例

fromEvent(document, "click")
    .mapTo(1)
    .scan((acc,cur) => acc + cur, 0)
    .subscribe(v => console.log(v))

操作符就介紹到這里，都是一些非常非常非常常用的，另外一些沒有介紹到的并非沒有用，需要的請自行搜索，通過操作符之間的相互組合我們可以實(shí)現(xiàn)非常高效便捷的數(shù)據(jù)處理。