0%

Rxjs【combineLatest、withLatestFrom、zip】

发表于 更新于
本文字数： 7.8k 阅读时长 ≈ 7 分钟

Rxjs学习之路

1、小贴士

这篇文章是我的Angular Rxjs Series中的第六篇文章，在继续阅读本文之前，您至少应该熟悉系列中的第一篇基础文章

Rxjs6都改变了些什么？

Rxjs【Observable】

1
// 图谱
2
// ----- 代表一个Observable
3
// -----X 代表一个Observable有错误发生
4
// -----| 代表一个Observable结束
5
// (1234)| 代表一个同步Observable结束
6
7
// 以下的操作符介绍均采用rxjs6的写法！！！

2、combineLatest

combineLatest可以接收多个Observable，但最后一个参数一定是callback function，这个回调函数接收的参数个数和前边传入Observable一一对应，最后需要注意，一定至少有2个Observable送出新值的时候才会执行回调函数。

1
/**
2
 * 取得各个observable 最后送出的值，再输出成一个值
3
 * combineLatest可以接收多个observable，最后一个参数是callback function，这个callback function接收的参数数量跟合并的observable数量相同
4
 * callback 都会依照合并的observable 数量来传入参数，如果我们合并了三个observable，callback 就会有三个参数，而不管合并几个observable 都会只会回传一个值。
5
 * source:      ----0----1----2|
6
 * newest:      --0--1--2--3--4--5|
7
 *          combineLatest(newest, (x, y) => x + y);
8
 * example:     ----01--23-4--(56)--7|
9
 */
10
const source = interval(500).pipe(take(3));
11
const newest = interval(300).pipe(take(6));
12
const combineLatestObservable = source.pipe(combineLatest(newest, (x, y) => x + y));
13
this.combineLatestSubscription = combineLatestObservable.subscribe({
14
    next: (value) => { console.log('=====combineLatest操作符: ', value); },
15
    error: (err) => { console.log('=====combineLatest操作符: Error: ', err); },
16
    complete: () => { console.log('=====combineLatest操作符: complete!'); }
17
});
  • 从上边的例子，我们可以看到，newest送出0时，source此时没有送出值，因此不执行回调函数，当source送出0时，此时newest送出的最新值是之前的0，执行回调0+0=0，以此类推…

3、withLatestFrom

withLatestFrom其实和combineLatest很像，唯一不同的是他多了一个主从关系，即只有主Observable送出新值的时候，才会执行callback function，其他情况下不会触发回调。

1
/**
2
 * 和combineLatest类似，但是withLatestFrom只有在主要的observable 送出新的值时，才会执行callback，附随的observable 只是在背景下运作
3
 * withLatestFrom 会在main 送出值的时候执行callback，但请注意如果main 送出值时some 之前没有送出过任何值callback 仍然不会执行！
4
 * callback 都会依照合并的observable 数量来传入参数，如果我们合并了三个observable，callback 就会有三个参数，而不管合并几个observable 都会只会回传一个值。
5
 * main:       ----h----e----l----l----o|
6
 * some:       --0--1--0--0--0--1|
7
 *         withLatestFrom(some, (x, y) => y === 1 ? x.toUpperCase() : x);
8
 * example:    ----h----e----l----L----O|
9
 */
10
const main = from('hello').pipe(zip(interval(500), (x, y) => x));
11
const some = from([0, 1, 0, 0, 0, 1]).pipe(zip(interval(300), (x, y) => x));
12
const withLatestFromObservable = main.pipe(withLatestFrom(some, (x, y) => y === 1 ? x.toUpperCase() : x));
13
this.withLatestFromSubscription = withLatestFromObservable.subscribe({
14
    next: (value) => { console.log('=====withLatestFrom操作符: ', value); },
15
    error: (err) => { console.log('=====withLatestFrom操作符: Error: ', err); },
16
    complete: () => { console.log('=====withLatestFrom操作符: complete!'); }
17
});
  • 从上边的例子，我们可以观察到：main送出h时，some的最新值是上一次的0，0不等于1，所以原样输出main，即h。当main送出e时，some送出的新值是上一次0，0不等于1，所以原样输出main，即e。以此类推…

4、zip

zip会取每个Observable且按顺序传入callback function回调函数的参数中，简单的理解就是一一对应，成双成对。

1
/**
2
 * 取每个observable 相同顺位的元素并传入callback，也就是说每个observable 的第n 个元素会一起被传入callback
3
 * zip 会把各个observable 相同顺位送出的值传入callback
4
 * zip 必须cache 住还没处理的元素，当我们两个observable 一个很快一个很慢时，就会cache 非常多的元素，等待比较慢的那个observable。这很有可能造成记忆体相关的问题！
5
 * callback 都会依照合并的observable 数量来传入参数，如果我们合并了三个observable，callback 就会有三个参数，而不管合并几个observable 都会只会回传一个值。
6
 * source:      ----0----1----2|
7
 * newest:      --0--1--2--3--4--5|
8
 *            zip(newest, (x, y) => x + y)
9
 * exaple:      ----0----2----4|
10
 */
11
const source = interval(500).pipe(take(3));
12
const newest = interval(300).pipe(take(6));
13
const zipObservable = source.pipe(zip(newest, (x, y) => x + y));
14
this.zipSubscription = zipObservable.subscribe({
15
    next: (value) => { console.log('=====zip操作符: ', value); },
16
    error: (err) => { console.log('=====zip操作符: Error: ', err); },
17
    complete: () => { console.log('=====zip操作符: complete!'); }
18
});
完整例子
1
import { Component, OnInit, OnDestroy } from '@angular/core';
2
import { Subscription, interval, from } from 'rxjs';
3
import { take, combineLatest, zip, withLatestFrom } from 'rxjs/operators';
4
5
@Component({
6
    selector: 'app-rxjs-demo06',
7
    template: `
8
        <h3>Rxjs Demo06 To Study! -- Operators操作符(combineLatest, withLatestFrom, zip)</h3>
9
        <button (click)="combineLatestHandler()">combineLatest</button>
10
        <button class="mgLeft" (click)="withLatestFromHandler()">withLatestFrom</button>
11
        <button class="mgLeft" (click)="zipHandler()">zip</button>
12
        <app-back></app-back>
13
    `,
14
    styles: [`
15
        .mgLeft {
16
            margin-left: 20px;
17
        }
18
    `]
19
})
20
export class RxjsDemo06Component implements OnInit, OnDestroy {
21
    combineLatestSubscription: Subscription;
22
    withLatestFromSubscription: Subscription;
23
    zipSubscription: Subscription;
24
25
    constructor() { }
26
27
    ngOnInit(): void {
28
        // 图谱
29
        // ----- 代表一个Observable
30
        // -----X 代表一个Observable有错误发生
31
        // -----| 代表一个Observable结束
32
        // (1234)| 代表一个同步Observable结束
33
    }
34
35
    combineLatestHandler() {
36
        /**
37
         * 取得各个observable 最后送出的值，再输出成一个值
38
         * combineLatest可以接收多个observable，最后一个参数是callback function，这个callback function接收的参数数量跟合并的observable数量相同
39
         * callback 都会依照合并的observable 数量来传入参数，如果我们合并了三个observable，callback 就会有三个参数，而不管合并几个observable 都会只会回传一个值。
40
         * source:      ----0----1----2|
41
         * newest:      --0--1--2--3--4--5|
42
         *          combineLatest(newest, (x, y) => x + y);
43
         * example:     ----01--23-4--(56)--7|
44
         */
45
        const source = interval(500).pipe(take(3));
46
        const newest = interval(300).pipe(take(6));
47
        const combineLatestObservable = source.pipe(combineLatest(newest, (x, y) => x + y));
48
        this.combineLatestSubscription = combineLatestObservable.subscribe({
49
            next: (value) => { console.log('=====combineLatest操作符: ', value); },
50
            error: (err) => { console.log('=====combineLatest操作符: Error: ', err); },
51
            complete: () => { console.log('=====combineLatest操作符: complete!'); }
52
        });
53
    }
54
55
    withLatestFromHandler() {
56
        /**
57
         * 和combineLatest类似，但是withLatestFrom只有在主要的observable 送出新的值时，才会执行callback，附随的observable 只是在背景下运作
58
         * withLatestFrom 会在main 送出值的时候执行callback，但请注意如果main 送出值时some 之前没有送出过任何值callback 仍然不会执行！
59
         * callback 都会依照合并的observable 数量来传入参数，如果我们合并了三个observable，callback 就会有三个参数，而不管合并几个observable 都会只会回传一个值。
60
         * main:       ----h----e----l----l----o|
61
         * some:       --0--1--0--0--0--1|
62
         *         withLatestFrom(some, (x, y) => y === 1 ? x.toUpperCase() : x);
63
         * example:    ----h----e----l----L----O|
64
         */
65
        const main = from('hello').pipe(zip(interval(500), (x, y) => x));
66
        const some = from([0, 1, 0, 0, 0, 1]).pipe(zip(interval(300), (x, y) => x));
67
        const withLatestFromObservable = main.pipe(withLatestFrom(some, (x, y) => y === 1 ? x.toUpperCase() : x));
68
        this.withLatestFromSubscription = withLatestFromObservable.subscribe({
69
            next: (value) => { console.log('=====withLatestFrom操作符: ', value); },
70
            error: (err) => { console.log('=====withLatestFrom操作符: Error: ', err); },
71
            complete: () => { console.log('=====withLatestFrom操作符: complete!'); }
72
        });
73
    }
74
75
    zipHandler() {
76
        /**
77
         * 取每个observable 相同顺位的元素并传入callback，也就是说每个observable 的第n 个元素会一起被传入callback
78
         * zip 会把各个observable 相同顺位送出的值传入callback
79
         * zip 必须cache 住还没处理的元素，当我们两个observable 一个很快一个很慢时，就会cache 非常多的元素，等待比较慢的那个observable。这很有可能造成记忆体相关的问题！
80
         * callback 都会依照合并的observable 数量来传入参数，如果我们合并了三个observable，callback 就会有三个参数，而不管合并几个observable 都会只会回传一个值。
81
         * source:      ----0----1----2|
82
         * newest:      --0--1--2--3--4--5|
83
         *            zip(newest, (x, y) => x + y)
84
         * exaple:      ----0----2----4|
85
         */
86
        const source = interval(500).pipe(take(3));
87
        const newest = interval(300).pipe(take(6));
88
        const zipObservable = source.pipe(zip(newest, (x, y) => x + y));
89
        this.zipSubscription = zipObservable.subscribe({
90
            next: (value) => { console.log('=====zip操作符: ', value); },
91
            error: (err) => { console.log('=====zip操作符: Error: ', err); },
92
            complete: () => { console.log('=====zip操作符: complete!'); }
93
        });
94
    }
95
96
    ngOnDestroy() {
97
        if (this.combineLatestSubscription) {
98
            this.combineLatestSubscription.unsubscribe();
99
        }
100
        if (this.withLatestFromSubscription) {
101
            this.withLatestFromSubscription.unsubscribe();
102
        }
103
        if (this.zipSubscription) {
104
            this.zipSubscription.unsubscribe();
105
        }
106
    }
107
}

Marble Diagrams【宝珠图】

1. 这个Marble Diagrams【宝珠图】可以很灵活的表现出每个操作符的使用
2. 下面是超链接传送门

Marble Diagrams【宝珠图】

Angular Rxjs Series

  1. Rxjs6都改变了些什么？
  2. Rxjs【Observable】
  3. Rxjs【map、mapTo、filter】
  4. Rxjs【take, first, takeUntil, concatAll】
  5. Rxjs【skip, takeLast, last, concat, startWith, merge】