1.创建数据流
2.合并数据流
（1）concat()：首尾相连—按顺序订阅Observable对象，因此若一个Observable对象不会完结，则后面的Observable对象永远没有上场的机会；
- 完结条件：最后一个Observable对象完结；
（2）merge()：第一时间订阅所有上游Observable，对其采取“先到先得”策略，任何一个Observable只要有数据推下来，就立刻传给下游Observable对象；主要用于合并产生异步数据的Observable对象；
- 完结条件：所有上游Observable对象完结；
- 可选参数concurrent：指定可以同时合并的Observable对象个数；
eg：const merge.merge(s2
, 2) — concurrent参数值为2，s1
优先合并，只有其中之一完结的时候，s3
对象；
- 应用场景：用一个事件处理函数eventHandler()处理click和touchend事件；
const click$ = Rx.Observable.fromEvent(element, 'click');
const touchend$ = Rx.Observable.fromEvent(element, 'touchend'); //fromEvent从网页中获取事件流；
Rx.Observable.merge(click$, touchend$).subscribe(eventHandler);
（3）zip()：一对一合并 — 每个上游Observable都要贡献一个元素放入一个数组中；
- 完结条件：只要任何一个上游的Observable完结，则zip()就会给下游一个complete信号；
const s1$ = Rx.Observable.interval(1000);
const s2$ = Rx.Observable.of('a', 'b', 'c');
Rx.Observable.zip(s1$, s2$).subscribe(console.log, null, () => console.log('complete')); // [0, 'a'], [1, 'b'], [2, 'c'], complete
（4）combineLatest()：合并最后一个数据 — 反复使用上游产生的最新数据产生数组，只要上游不产生新的数据，则会反复使用这个上游产生的最后一次的数据；
- 完结条件：所有的上游Observable对象完结，combineLatest()会给下游一个complete信号；
// s1$是同步数据流，当它产生数据的时候combineLatest还未来得及订阅s2$，因此当订阅s2$时，s1$的最新值为3，s2$虽然也是同步数据流，但它产生一个数据都会有准备好的s1$的最新值，所以s2$产生的每个数据都会引发下游一个数据的产生；
const s1$ = Rx.Observable.of(1, 2, 3);
const s2$ = Rx.Observable.of('a', 'b');
const res$ = s1$.combineLatest(s2$); // [3, 'a'], [3, 'b'], complete；
const original$ = Observable.timer(0, 1000);
const s1$ = original$.map(x => x+'a');
const s2$ = original$.map(x => x +'b');
s1$.combineLatest(s2$).subscribe(console.log, nul, () =>console.log('complete'));
// ['0a', '0b'], 1s后：['1a', '0b'], ['1a', '1b']; 2s后: ['2a', '1b'], ['2a', '2b']... 只要上游数据有新的值就推送给下游；
- 可选参数：函数 — 将上游产生的最新值数组做处理后再返回给下游Observable对象；
- 合并完全独立的Observable对象；
（5）withLatestFrom()：类似combineLatest()，但是给下游推送数据只能由一个Observable对象驱动，作为参数的Observable对象只能贡献数据，不能控制产生数据的时机；
const original$ = Observable.timer(0, 1000);
const s1$ = original$.map(x => x+'a');
const s2$ = original$.map(x => x +'b');
s1$.withLatestFrom(s2$).subscribe(console.log, nul, () =>console.log('complete'));
// ['0a', '0b'], 1s后：['1a', '1b']; 2s后: ['2a', '2b']...
- 同时要从其他Observable对象获取最新数据，然后映射成新的数据流；
（6）race()：只订阅第一个吐出数据的Observable对象，其余的Observable对象将被退订；
（7）forkJoin()：接受多个Observable对象做参数，当所有Observable对象都完结了，将所有输入的Observable产生的最后一个数据合并起来传给下游Observable对象；
const s1$ = Observable.interval(1000).map(x => x+'a').take(1); //第一秒产生‘0a’，然后complete；
const s2$ = Observable.interval(1000).map(x => x+'b').take(3); // 在第三秒产生‘2b’，然后完结；
Observable.forkJoin(s1$, s2$).subscribe(console.log, null, ()=>console.log('complete')); // 只取各对象complete前产生的最后一个数据，即：['0a', '2b'], complete
网友评论