数据流的操作
2019-07-08 本文已影响0人
励志摆脱懒癌的少女酱
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