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

接著上一節的操作符繼續，讓我們直奔主題。

組合類的操作符可以將不同流數據按一定的規則進行合并，從而獲得所需要的完整數據。

combineLatest

實例方法

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

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

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

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

            combineLatest

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

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

返回 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

實例方法

將多個輸入流組合成一個輸出流，輸出流上的數據是由每一條輸入流上的最后一條數據組合成的，但是組合動作只在源 Observable 發出值時發生，這里的源 Observable 指的是調用withLatestFrom的實例。也就是說只有當源 Observable 產生值時，才會產生新的值。

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

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

            withLatestFrom

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

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

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

示例

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

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

zip

實例方法

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

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

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

            zip

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

返回值 Observable 將各輸入流上對應的值組合成新的值發送給輸出流或者先經過映射函數處理后再發送給輸出流。

它也可以接受映射函數作為參數。

示例

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

實例方法

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

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

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

                merge

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

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

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

示例

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

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

forkJoin

靜態方法

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

forkJoin可以以參數或數組的形式接收任意數量的輸入流。如果沒有轉入輸入流，輸出流將會立即發出結束通知。

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

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

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

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

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

                forkJoin

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

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

可以接收映射函數作為參數。

示例

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

實例方法

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

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

            map(v => v * 10);

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

返回值 Observable 發出映射后的值的流。

示例

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

mergeMap

實例方法

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

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

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

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

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

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

返回值 Observable 一條合并了所有映射函數生成的內部流的流，內部流上的值都會在這條流上發出。

示例

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

實例方法

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

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

    -10---10---10-| // 內部流

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

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

輸出流上的值是由映射函數在輸入流的值的基本上生成的內部流所發出的，輸出流只允許觀察一條內部流，當有新的內部流到達時，輸出流將會取消對之前的內部流的訂閱，轉而訂閱這個最新的內部流，并發出它上面的值。

返回值 Observable 僅從最新的內部流上取值的流。

示例

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

concatMap

實例方法

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

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

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

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

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

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

示例

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

groupBy

實例方法

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

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

            groupBy(v => v%2);

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

返回值 Observable 發出分組后的流的高階流，分組的流都有一個唯一的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

實例方法

在源 Observable 上應用一個累積函數，當源 Observable 結束時把累積的值推送給輸出流，可以接受一個初始的累積值。

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

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

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

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

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

返回值 Observable 把源 Observable 上的值都經過累積函數計算后得到的值作為值發送的流。

示例

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

實例方法

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

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

示例

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

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