从命令式到响应式（五）

摘要：输出流只有在所有的输入流都完成以后才能完成，任何一条输入流上的错误都将立即推送到输出流上。如果没有转入输入流，输出流将会立即发出结束通知。返回值以数组形式获取到的每一个输入流的值，或者来自映射函数的值。返回值仅从最新的内部流上取值的流。

接着上一节的操作符继续，让我们直奔主题。

组合类的操作符可以将不同流数据按一定的规则进行合并，从而获得所需要的完整数据。

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

操作符就介绍到这里，都是一些非常非常非常常用的，另外一些没有介绍到的并非没有用，需要的请自行搜索，通过操作符之间的相互组合我们可以实现非常高效便捷的数据处理。