


























原文链接:https://blog.angular-university.io/rxjs-higher-order-mapping/
有一些在日常开发中常用的RxJS的操作符是高阶操作符:switchMap,mergeMap,concatMap,以及exhaustMap。
举个例子,程序中大多数的网络请求都是通过以上某个操作符来完成的,所以为了能够写出几乎所有反应式编程,必须熟悉这些操作符的运用。
在给定的场景中,知道用哪个操作符以及为什么要用那个操作符,有时候会让我们觉得有些迷惑。我们经常很想搞清楚这些操作符是如何运作的,还有为什么它们要叫那个名字。
这些操作符可能看起来互不相干,不过我们确实需要把它们放在一起学习。因为如何选择了错误的操作符,很可能给我们的程序带来致命的问题。
有个原因:为了理解这些操作符,我们首先要理解每个操作符的内部是如何使用Observable的组合策略的。
暂且先不讨论switchMap本身,我们先搞懂什么是Observable switching;暂且不深究concatMap,我们先学习一下Observable concatenation,等。
这就是我们将在本文中要做的事情。我们将按照逻辑顺序来学习concat,merge,switch还有exhaust的策略以及各自对应的mapping操作符:concatMap,mergeMap,switchMap以及exhaustMap。
我们将使用弹珠图(marble diagram)结合实例代码来解释基本概念。
最后,你会很清楚地知道每个操作符是如何运作的、什么时候使用哪个操作符、为什么要使用那个,以及各个操作符名称的缘由。
本文内容,我们将包含如下主题:
此段原文未翻译:
Note that this post is part of our ongoing RxJs Series. So without further ado, let's get started with our RxJs mapping operators deep dive!
一开始,我们先了解一下这些操作符通常在做些什么。
正如操作符的名字所述,它们在做某种映射。不过,到底是什么被映射了呢?我们先看下RxJS映射操作符的弹珠图:

利用map操作符,我们可以传入一个输入流(值1,2,3),然后从输入流,再创建一个派生出来的映射输出流(值10,20,30)。
底部输出流的值是通过拿到输入流的值并应用了一个函数而获得的。该函数就是简单地把数值乘以10。
如此,map操作符都是用来映射输入Observable的值的。这里有个例子,关于我们如何使用map操作符来处理一个HTTP请求:
| const http$ : Observable<Course[]> = this.http.get('/api/courses'); | |
| http$ | |
| .pipe( | |
| tap(() => console.log('HTTP request executed')), | |
| map(res => Object.values(res['payload'])) | |
| ) | |
| .subscribe( | |
| courses => console.log("courses", courses) | |
| ); |
此例中,我们创建一个HTTP observable,发出一个向后端的请求,并且订阅该observable。这个Observable会发出来自后端的HTTP响应,该响应是一个JSON对象。
在此例中,这个HTTP响应在属性payload中包裹着数据,所以为了获得数据,我们运用RxJS的map操作符。映射函数将会映射JSON响应的payload,并且提取出属性payload对应的值。
现在我们已经回顾了基本的映射是如何工作的,让我们来谈谈更高阶映射(high-order mapping)。
在更高阶映射中,我们不会简单地映射一个值比如说1到另外一个值比如说10,我们将把值映射进另外一个Observable中!
结果是一个高阶Observable。它和其他可观察对象一样,只是它的值本身也是可观察对象,我们可以单独订阅。
这可能听起来有些牵强,但事实上,这种类型的mapping一直都在发生。我们来给一个关于这种类型的mapping的实例。我们举例来说,我们有一个Angular反应式表单,它会通过一个Observable一直发送合法的表单值:
| @Component({ | |
| selector: 'course-dialog', | |
| templateUrl: './course-dialog.component.html' | |
| }) | |
| export class CourseDialogComponent implements AfterViewInit { | |
| form: FormGroup; | |
| course:Course; | |
| @ViewChild('saveButton') saveButton: ElementRef; | |
| constructor( | |
| private fb: FormBuilder, | |
| private dialogRef: MatDialogRef<CourseDialogComponent>, | |
| @Inject(MAT_DIALOG_DATA) course:Course) { | |
| this.course = course; | |
| this.form = fb.group({ | |
| description: [course.description, | |
| Validators.required], | |
| category: [course.category, Validators.required], | |
| releasedAt: [moment(), Validators.required], | |
| longDescription: [course.longDescription, | |
| Validators.required] | |
| }); | |
| } | |
| } |
这个反应式表单提供一个可观察对象this.form.valueChanges,它在用户跟表单进行交互的时候发送最新的表单值。这个会作为我们的可观察对象源(source Observable)。
我们要做的是保存至少一些一直被发出的值,以来实现一个预存表单草稿的特性。这样一来,随着用户填写表单的时候,数据就被渐进式的保存起来。如此可以避免因为意外的页面重载导致整个表单数据的丢失。
为了实现保存表单草稿的功能,我们需要拿到表单值,然后创建第二个HTTP observable,该可观察对象会做后端保存,接着订阅它。
我们本可以手动实现这些,不过那样我们就会掉进嵌套订阅的反面模式中:
| this.form.valueChanges | |
| .subscribe( | |
| formValue => { | |
| const httpPost$ = | |
| this.http.put(`/api/course/${courseId}`, formValue); | |
| httpPost$.subscribe( | |
| res => ... handle successful save ... | |
| err => ... handle save error ... | |
| ); | |
| } | |
| ); |
如我们所见,这很快会造成我们的代码行成多级嵌套,这个就是我们在第一个地方使用RxJS来避免的问题。
让我们称呼新的
httpPost$可观察对象为内部可观察对象,因为它是由内部嵌套代码块生成的。
我们愿意用一种更简便的方法去做所有的过程:拿到表单值,然后映射到用于保存的Observable。这会有效地创建一个高阶可观察对象,它的每个值都对应一个保存请求。
接着我们显然会订阅每个网络请求Observable,然后能直接一步到位获取所有网络请求的响应,这样会避免任何嵌套。
如果我们有某种更高阶的RxJS映射操作符,我们就可以做到这一切!我们为什么需要四种不同的操作符呢?
为了理解这个,试想如果有多个表单值一连串快速地从valueChanges observable发送出来,并且保存操作需要耗费一些时间来完成:
在探讨以上这些用例之前,让我们回顾一下上面的嵌套订阅代码。
在嵌套订阅范例中,我们实际上以并行方式触发着保存操作,这并不是我们所想要的,因为无法强力保证后端会按照顺序处理保存请求,而且最后一个有效的表单值确实是存储在后端的。
让我们看下如何确保只在上一次保存完成后才会执行保存请求。
为了实现按顺序保存,我们将引入一个新的概念:可观察对象连接。在这个代码范例中,我们使用RxJS函数concat()连接两个可观察对象范例:
| const series1$ = of('a', 'b'); | |
| const series2$ = of('x', 'y'); | |
| const result$ = concat(series1$, series2$); | |
| result$.subscribe(console.log); |
在使用of函数创建了两个可观察对象series1$和series2$后,我们又创建了第三个可观察对象result$,它是连接series1$和series2$的结果。
这里是该程序的控制台输出,表示结果可观察对象发出的值:
我们能看到,这些值是series1$和series2$的值一起连接后的结果。但问题就在这里:只有当这些可观察对象正在完成时才会生效。
函数of()将创建一些可观察对象,这些可观察对象会把传入到of()的值发送出去,当所有的值都被发出后将完成这些可观察对象。
为了真的理解发生了什么,我们需要看一下可观察对象的弹珠图:

你有没有注意到第一个可观察对象的值b右边的竖线标记?它标记着带有值a和b的第一个可观察对象(series1$)完成的时间点。
沿着时间线,让我们一步一步地把过程分解,看看发生了什么:
series1$和series2$被传给了函数concat()concat()会订阅第一个可观察对象series1$,但不会订阅第二个可观察对象series2$(这对理解连接是至关重要的)source1$发送值a,它会立即反应到可观察对象result$的输出中source2$还未发送值,因为它还没有被订阅source1$接着发送值b,它被反应到输出中source1$然后就完成了,只有当它完成了,concat()才会订阅source2$source2$的值将开始反应到输出,直至完成source2$完成后,result$也将完成concat(),不仅仅像此例中的两个正如我们所见,可观察对象连接就是关于可观察对象的完成!我们取第一个可观察对象并使用它的值,等待它完成,然后我们使用下一次可观察对象···,直到所有可观察对象完成。
返回到我们的更高阶可观察对象映射的范例,让我们看看连接的概念是如何帮助我们的。
正如我们已经看到的,为了确保我们的表单值被按顺序保存,我们需要取得每个表单值,然后映射到一个httpPost$可观察对象。
接着我们需要订阅它,不过我们想要在订阅下一个httpPost$可观察对象之前完成保存。
为了确保顺序,我们需要将多个
httpPost$可观察对象连接到一起!
我们将订阅每个httpPost$,然后按顺序处理每个请求的结果。在最后,我们需要的是一个操作符,它将混合以下内容:
httpPost$可观察对象)concat()操作符,连接多个httpPost$可观察对象,以确保在前一个进行中的保存操作完成之前,不会创建新的HTTP保存请求我们需要的就是恰当命名的RxJS操作符:concatMap,它通过可观察对象连接来做更高阶映射的混合操作。
下面是如果我们使用concatMap操作符时的代码:
| this.form.valueChanges | |
| .pipe( | |
| concatMap(formValue => this.http.put(`/api/course/${courseId}`, | |
| formValue)) | |
| ) | |
| .subscribe( | |
| saveResult => ... handle successful save ..., | |
| err => ... handle save error ... | |
| ); |
我们能看到,使用更高阶映射操作符的第一个好处就是我们不会再有嵌套订阅了。
通过使用concatMap,现在所有的表单值都按顺序被发送到后端,在Chrome DevTools Network中显示如下:

拆分concatMap的网络日志图
能看到,一个HTTP保存请求仅在前一个保存完成后才开始。下面是concatMap操作符如何确保请求总是按顺序发生:
注意这里的代码仅仅是保存表单草稿的基本实现。你可以跟其他的操作符结合使用,比如仅保存合法的表单值,节流保存,以确保它们不会发生得太频繁。
应用可观察对象连接到一系列HTTP保存操作,看起来似乎是一个不错的方法,可以确保保存操作按照想要的顺序发生。
但是有其他场景,相反我们想要并行执行操作,不等待前一个内部可观察对象完成。
要完成这个,我们有合并可观察对象结合策略!merge,不像concat,在订阅下一个可观察对象之前不会等待任何可观察对象。
相反,merge同时订阅每个被合并的可观察对象,当多个值到达时,它将把每个源可观察对象的值输出到结果可观察对象中。
为了搞清楚合并(merging)不依赖完成,让我们合并两个永不完成的可观察对象,因为它们是间隔可观察对象(interval Observable):
| const series1$ = interval(1000).pipe(map(val => val*10)); | |
| const series2$ = interval(1000).pipe(map(val => val*100)); | |
| const result$ = merge(series1$, series2$); | |
| result$.subscribe(console.log); |
由interval()创建的可观察对象将会在一秒钟间隔内发送值0,1,2···,它永远不会完成。
注意我们应用了几个map操作符到这些interval Observable中,仅仅为了在控制台更方便地区分它们。
下面是在控制台中能看到的最开始的一些值:
| 0 | |
| 0 | |
| 10 | |
| 100 | |
| 20 | |
| 200 | |
| 30 | |
| 300 |
我们可以看到,合并后的源可观察对象的值只要被发出,就会立即显示在结果可观察对象中。如果其中一个合并后的可观察对象完成了,merge会继续持续发出其他可观察对象的值。
注意如果源可观察对象完成了,merge仍然会以相同方式工作。

我们可以看到,合并后的源可观察对象的值会立即显示在输出中。结果可观察对象在所有的合并后的可观察对象完成之后才会完成。
现在我们理解了合并策略,让我们看看它在更高阶可观察对象的映射上下文中将被如何使用。
如果我们把合并策略与更高阶可观察对象映射的概念相结合,我们就拥有了RxJS mergeMap 操作符。让我们来看看这个操作符的弹珠图:

下面是mergeMap操作符是如何工作的:
返回到我们的前一个表单草稿保存范例,很明显之所以我们需要concatMap而不是mergeMap,是因为我们不希望保存是并行发生的。
让我们看看如何意外地选择使用mergeMap的话,会发生什么:
| this.form.valueChanges | |
| .pipe( | |
| mergeMap(formValue => | |
| this.http.put(`/api/course/${courseId}`, | |
| formValue)) | |
| ) | |
| .subscribe( | |
| saveResult => ... handle successful save ..., | |
| err => ... handle save error ... | |
| ); |
现在,假设用户在跟表单交互,并开始非常快地输入数据。在这种情况下,在网路日志中,我们会看到多个保存请求在并行运行着。

我们可以看到,请求在并行地发生着,在这种情况下是一种错误!在重载情况下,有可能这些请求会乱序进行。
让我们来谈谈另外一个Observable合并机制:switching。从我们不等待任何Observable完成的方面来说,相对于连接,切换的概念更接近于合并。
但是switching跟merging不同,如果一个新的Observable开始发出值的话,在订阅新的Observable之前,我们会取消订阅前一个Observable。
Observable switching是为了确保未使用的Observable的取消订阅逻辑被触发,这样资源可以得到释放!
让我们看看switch的弹珠图:

注意那些斜线,它们并非意外!在switch策略的情况下,重要的是要在图表中表示更高阶的Observable,也就是图片中最上面一行。
更高阶的Observable发送本身就是Observable的值。
当一条对角线从高阶可观察对象的顶部那条线分叉的时候,就是一个值被switch发出并订阅的时候。
下面是在弹珠图中发生了什么:
现在我们能明白,为什么图表不得不用这样一种不寻常的方式,以对角线来画了:这是因为我们需要直观地表示出每个内部Observable何时被订阅(或被取消订阅),这发生在对角线Observable从源Observable分叉的地方。
让我们接着把switch策略应用到高阶映射中。假设我们有一个普通的输入流,它会发出值1,3和5。
我们将把每个值映射到一个Observable,就像我们在concatMap和mergeMap的例子中做的那样,并获得一个更高阶的可观察对象。
如果我们现在在被发出的内部Observable中切换,而不是连接或合并它们,我们最终会使用switchMap操作符:

下面是这个操作符是如何工作的:
所以我们可以看到,Observable切换就是确保我们触发未使用Observable的取消订阅逻辑。让我们来看switchMap的实战!
关于switchMap有一个很常见的例子:search TypeAhead(待翻译)。首先让我们定义源Observable,它的值本身会触发检索请求。
这个源Observable会发出值,这些值是用户在输入框中键入的文本:
| const searchText$: Observable<string> = | |
| fromEvent<any>(this.input.nativeElement, 'keyup') | |
| .pipe( | |
| map(event => event.target.value), | |
| startWith('') | |
| ) | |
| .subscribe(console.log); |
这个源Observable被关联到一个输入文本字段,用户在它上面键入搜索内容。当用户键入单词“Hello World”作为检索,下面是被searchText$发出的值:
| H | |
| H | |
| He | |
| Hel | |
| Hell | |
| Hello | |
| Hello | |
| Hello W | |
| Hello W | |
| Hello Wo | |
| Hello Wor | |
| Hello Worl | |
| Hello World |
注意那些重复的值,不管是由用户在两个单词之间使用空格引起的,还是因为使用Shift键让字母H和W大写。
为了防止向后端发送这些所有的值作为检索请求,让我们使用debounceTime操作符来等待用户输入直至稳定。
| const searchText$: Observable<string> = | |
| fromEvent<any>(this.input.nativeElement, 'keyup') | |
| .pipe( | |
| map(event => event.target.value), | |
| startWith(''), | |
| debounceTime(400) | |
| ) | |
| .subscribe(console.log); |
通过使用该操作符,如果用户以正常速度键入,我们只有searchText$的一个值在输出中。
跟之前相比,现在已经好很多了。现在只有稳定了至少400ms的值才会被发出!
但是,如果用户一边在思考检索内容,一边很慢地进行输入的。就是说在两个值之间会超过400ms,那么检索流会看起来像下面这样:
而且,用户可能输入一个值,按退格键(backspace),然后又输入一遍,这会造成重复的检索值。我们可以通过添加distinctUntilChanged操作符来阻止重复检索的情形发生。
但不止如此,我们需要一种方法,当新的检索开始后,取消之前的检索。
在这里我们需要做的是转换每个检索字符串到后端检索请求并订阅它。在两个连续的检索请求中应用switch策略,当新的检索被触发后,前面的检索将被取消。
这正是switchMap操作符将要做的事情!下面是我们的Typeahead逻辑用到的最终实现:
| const searchText$: Observable<string> = | |
| fromEvent<any>(this.input.nativeElement, 'keyup') | |
| .pipe( | |
| map(event => event.target.value), | |
| startWith(''), | |
| debounceTime(400), | |
| distinctUntilChanged() | |
| ); | |
| const lessons$: Observable<Lesson[]> = searchText$ | |
| .pipe( | |
| switchMap(search => this.loadLessons(search)) | |
| ) | |
| .subscribe(); | |
| function loadLessons(search:string): Observable<Lesson[]> { | |
| const params = new HttpParams().set('search', search); | |
| return this.http.get(`/api/lessons/${coursesId}`, {params}); | |
| } |
让我们来看下switchMap操作符的实践!当用户在检索栏中键入内容,然后稍稍犹豫了下,接着键入其他内容,下面是在网络日志中大体上可以看到的内容:

我们可以看到,有几个前面的检索在它们进行的地方已经被取消了。这太棒了!这可以节省服务器资源用于做其他的事情。
在typeahead的场景中,switchMap操作符是最合适的。不过也有一些其他的情况:我们要做的是在前面的值没有完成处理之前,忽略源Observable中所有新的值。
举个例子,假如我们在点击保存按钮的时候,触发向后端的保存请求。为了确保保存操作按顺序发生,我们可能首先使用concatMap操作符来实现它。
| fromEvent(this.saveButton.nativeElement, 'click') | |
| .pipe( | |
| concatMap(() => this.saveCourse(this.form.value)) | |
| ) | |
| .subscribe(); |
这个确保了保存操作按顺序处理。但如果用户多次点击了保存按钮,会发生什么呢?下面是我们在网络日志中会看到的内容:

我们能看到,每个点击触发了各自的保存:假如我们点击20次,我们会有20次保存操作!在这个例子中,我们需要多做一些事情,而不仅仅让保存操作按顺序发生。
我们也想要能够取消点击,但仅在某个保存操作已经在进行的时候。exhaust可观察对象结合策略将允许我们实现这个。
为了理解exhaust是如何工作的,让我们看一下这张弹珠图:

就像之前一样,这里在首行中我们有一个高阶Observable,它的值本身是Observable,从那个首行中分叉开来。下面是在这张弹珠图中发生的内容:
就像concat,merge还有switch,现在我们可以应用exhaust策略到高阶映射的上下文中。
现在让我们来看下exhaustMap的弹珠图。记住,跟上一个弹珠图中的首行不一样,源Observable1-3-5发出的值不是Observable。
相反,这些值可能是比如鼠标点击:

那么下面是exhaustMap弹珠图中发生的内容:
现在让我们把exhaustMap操作符应用到我们的保存按钮场景中:
| fromEvent(this.saveButton.nativeElement, 'click') | |
| .pipe( | |
| exhaustMap(() => this.saveCourse(this.form.value)) | |
| ) | |
| .subscribe(); |
如果现在我们一次性点击5次保存按钮,我们将得到以下网络日志:

可以看到,正如所料,某个保存请求还在进行中时产生的点击都被忽略了!
注意如果我们一次性持续点击比如20次,最终进行中的保存请求会结束,然后第二个请求将开始。
从高阶映射操作符的角度来说,concatMap,mergeMap,switchMap和exhaustMap的行为是相似的。
但它们在许多微妙的地方也是如此不同,以至于没有一种操作符可以安全地作为默认选择。
相反,我们可以基于使用场景简单地选择恰当的操作符:
如果你想尝试运行本文中的例子,这是包含了本文中运行代码的演练场
这个代码库包含了一个小的HTTP后端请求,它会帮助我们在更实际的场景中尝试RxJS映射操作符,它还包含了运行代码,像预保存草稿表单,a typeahead(未翻译),用Reactive风格写的关于组建的主题和样例:

我们已经看到,在反应式编程比如网络请求中做一些很常见的操作,使用RxJS高阶映射操作符是必不可少的。
为了真正理解这些操作符还有它们的名称,我们首先需要集中理解Observable的底层结合策略concat,merge,switch和exhaust。
我们也要认识到,有一个更高阶映射操作在发生,它们的值被转换成独立的Observable,这些Observable被映射操作符本身以一种隐藏的方式订阅。
选择正确的操作符就是关于选择正确的内部Observable组合策略。选择错误的操作符通常不会马上造成程序崩溃,但随着时间的推移,它可能导致一些难以排除的问题。
我希望你喜欢这篇文章!如果关于RxJS,你愿意学习更多,我们推荐看下RxJS实战课程,这些课程更详细地介绍了很多有用的模式和操作符。
同样,如果你们有任何问题或评论,请在下方留言,我会回复你们的。
如果你正开始学习Angular,看一下Angular入门教程
-- The End --
此内容由惯性聚合(RSS阅读器)自动聚合整理,仅供阅读参考。 原文来自 — 版权归原作者所有。