rxjs5.X系列 —— Combination/Multicasting系列 api 笔记

时间：2022-12-02 16:10:08

欢迎指导与讨论 :）

　　前言

　　　　本文是笔者翻译 RxJS 5.X 官网各类operation操作系列的的第三篇 —— Combination组合与Multicasting广播。如有错漏，希望大家指出提醒O(∩_∩)O。更详细的资料尽在rxjs官网 【http://reactivex.io/rxjs/manual/overview.htm】与带有demo例子的网站【http://xgrommx.github.io/rx-book/content/observable】。

　　　　本文有关Combination操作的内容有：combineAll、combineLastest、concat、cancatAll、exhaust、forkJoin、merger、mergeAll、race、startWith、switch、withLastestFrom、zip、zipAll

　　　　有关Multicasting操作的内容有：cache、multicast、publish、publishBehavior、publishLast、publishReplay、share

　　一、combineAll

　　　　将高阶Observable转化为一阶。当外层Observable结束时，对每个内层Observable使用combineLastest，并最终以数组项的形式返回每个内层Observable的最新值。

// 点击三次后，外层Observable结束

// 然后对三个内层Observable使用combineLastest

Rx.Observable.fromEvent( document, 'click')

    .map( ev => Rx.Observable.of( 1, 2, 3 ))

    .take( 3 )

    .combineAll( )

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

// 输出 [ 3, 3, 1 ], [ 3, 3, 2 ], [ 3, 3, 3]

　　二、combineLastest

　　　　每当其中一个Observable发射值时，都会以数组的形式发射两个Observable的最新值。

Rx.Observable.interval( 1000 )

    .combineLatest( Rx.Observable.timer( 1000 , 2000 ))

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

// 输出 [ 0, 0 ], [ 1, 0 ], [ 2, 1 ], [ 3, 2 ]...

　　三、concat

　　　　链式拼接两个Observable的结果

Rx.Observable.of( 1, 2, 3 )

    .concat( 'a', 'b', 'c' )

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

// 输出：1,2,3,a,b,c

　　四、cancatAll

　　　　将高阶Observable转化为一阶。并将结果以链式拼接的形式进行发射。

Rx.Observable.fromEvent( document, 'click')

    .mapTo( Rx.Observable.interval( 1000 ).take( 3 ))

    .concatAll( )

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

// 点击三次，则输出：0 1 2, 0 1 2,  0 1 2

　　五、exhaust

　　　　将高阶Observable转化为一阶。并将结果以链式拼接的形式进行发射。但是，当前一个内层Observable的值还没有发射完毕时，不会接受下一个内层Observable，并不会把它的值拼接到外层Observable中。

Rx.Observable.fromEvent( document, 'click')

    .mapTo( Rx.Observable.interval( 1000 ).take( 3 ))

    .exhaust( )

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

// 在值被发射时，点击不会产生新的Observable并进行值的拼接

　　六、forkJoin 

　　　　将多个Observable进行并行计算，并数组的形式返回它们各自的最新值

Rx.Observable.forkJoin(

        Rx.Observable.of(42),

        Rx.Observable.range(0, 10),

        Rx.Observable.of(1,2,3)

    )

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

// 输出: [ 42, 10, 3 ]

　　七、merger

　　　　将两个Observable进行合并，每当其中一个Observable发射值时，都会被外层Observer所收到。

Rx.Observable.fromEvent( document, 'click' )

    .merge( Rx.Observable.interval( 1000 ))

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

// 不点击的情况下，每秒输出i， i为从零到n

// 点击一下，马上输出 $mouseEvent

　　八、mergeAll

　　　　将多个Observable进行合并，每当其中一个Observable发射值时，都会被外层Observer所收到。

Rx.Observable.fromEvent( document, 'click' )

    .mapTo( Rx.Observable.interval( 1000 ))

    .mergeAll( )

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

// 每次点击，新增一个计时器，并把它合并到同一个流中

　　九、race —— 暂无

　　十、startWith

　　　　在源Observable开头插入一个指定的值

Rx.Observable.interval( 1000 )

    .startWith( 123 )

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

// 输出：123, 1, 2, 3,....

　　十一、switch

　　　　将高阶Observable转化为一阶Observable。当生成新的内层Observable时，外层流会丢弃上一个内层Observable，并发射新的内层Observable的一系列的值，并重复上述过程。既有，重新开始的意思。

Rx.Observable.fromEvent( document, 'click' )

    .mapTo( Rx.Observable.interval( 1000 ))

    .switch( )

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

// 每次点击会重新输出：0,1,2,3,4,5....

　　十一、withLastestFrom

　　　　每当源Observable发射新的值时，会以数组的形式，把源Observable和另一个Observable的最新值进行组合并发射。

Rx.Observable.fromEvent( document, 'click' )

    .withLatestFrom( Rx.Observable.interval( 1000 ))

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

// 每次点击输出: [ MouseEvent, x ], x是计时器的最新值

　　十二、zip

　　　　组合多个Observable，并生成一个新的Observable，其值能够通过每个Observable的值，和自定义函数进行定义。

let age$ = Rx.Observable.of<number>(27, 25, 29);

let name$ = Rx.Observable.of<string>('Foo', 'Bar', 'Beer');

let isDev$ = Rx.Observable.of<boolean>(true, true, false);

Rx.Observable

    .zip(age$, name$, isDev$,

         (age: number, name: string, isDev: boolean) => ({ age, name, isDev }))

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

// 输出

// {age: 27. name: 'Foo', isDev: true }

// {age: 25. name: 'Bar', isDev: true }

// {age: 29. name: 'Bear', isDev: false}

　　十三、zipAll —— 暂无

　　十四、cache —— 暂无

　　十五、multicast

　　　　返回一个ConnectableObservable。每一个订阅了同一个Observable的observer，实际上是拥有不同的、独立的Observable的执行( 原文：each subscribed Observer owns an independent execution of the Observable )，而Subject是多播的。

var source = Rx.Observable.create((o)=>{

    o.next(1);o.next(2);

});

var subject = new Rx.Subject();

var multicasted = source.multicast(subject);

// 原理是`subject.subscribe({...})`:返回的multicasted是一个connectableObservable

multicasted.subscribe({next: (v) => console.log('observerA: ' + v)});

multicasted.subscribe({next: (v) => console.log('observerB: ' + v)});

// 原理是 `source.subscribe(subject)`:

multicasted.connect();

//  observerA: 1 observerB: 1

//  observerA: 2 observerB: 2

　　十六、publish

　　　　返回一个ConnectableObservable。可进行广播

var a$ = Rx.Observable.interval( 1000 )

    .publish( )

a$.subscribe({next: (v) => console.log( v )});

a$.subscribe({next: (v) => console.log( v )});

a$.connect();

// 同时输出 1 1, 2 2, 3 3...

　　十七、publishBehavior —— 暂无

　　十八、publishLast 

　　　　   返回一个ConnectableObservable。且只广播该Observable的最后一个值

var a$ = Rx.Observable.of( 1, 3, 4, 5)

    .publishLast( )

a$.subscribe({next: (v) => console.log('observerA: ' + v)});

a$.subscribe({next: (v) => console.log('observerB: ' + v)});

a$.connect( )

// 输出

// observerA: 5

// observerB: 5

　　十九、publishReplay

　　　　   返回一个ConnectableObservable。且当第二个及以后才订阅这个ConnectableObservable时，只会受到最新的n个值，这个n由我们以参数的形式提供

var a$ = Rx.Observable.of(1,2,3,4,5)

    .publishReplay( 3 )

    .refCount( )

a$.subscribe({next: (v) => console.log('observerA: ' + v)});

a$.subscribe({next: (v) => console.log('observerB: ' + v)});

a$.subscribe({next: (v) => console.log('observerC: ' + v)});

// 输出

// observerA: 1 observerA: 2 observerA: 3 observerA: 4 observerA: 5

// observerB: 3 observerB: 4 observerB: 5

// observerC: 3 observerC: 4 observerC: 5

　　二十、share

　　　　　返回一个可被共享的Observable。它是 .publish( ).refCount( )的另一种写法

var a$ = Rx.Observable.interval( 1000 )

    .share( )

a$.subscribe({next: (v) => console.log( v )});

a$.subscribe({next: (v) => console.log( v )});

// 同时输出 1 1, 2 2, 3 3...

相关文章