Rxjs 操做符快速入门

前言

好的程序员懂得如何从重复的工做中逃脱：

- 操做DOM时，发现了Jquery。

- 操做JS时，发现了lodash。

- 操做事件时，发现了Rx。
复制代码

Rxjs自己的 概念 并不复杂，简单点说就是对观察者模式的封装，观察者模式在前端领域大行其道，不论是使用框架仍是原生JS，你必定都体验过。

在我看来，Rxjs的强大和难点主要体现对其近120个操做符的灵活运用。

惋惜官网中对这些操做符的介绍晦涩难懂，这就致使了不少人明明理解Rxjs的概念，却苦于不懂的使用操做符而黯然离场。

本文总结自《深刻浅出Rxjs》一书，旨在于用最简洁、通俗易懂的方式来讲明Rxjs经常使用操做符的做用。学习的同时，也能够作为平时快速查阅的索引列表。

阅读提醒

• 流的概念

• 订阅：调用 subscribe
• 吐出：调用 next
• 完成：调用 complete

须要注意流的完成和订阅时间，某些操做符必须等待流完成以后才会触发。

其实根据操做符的功能咱们也能够大体推断出结果：若是一个操做符须要拿到全部数据作操做、判断，那必定是须要等到流完成以后才能进行。

建立流操做符

建立流操做符最为流的起点，不存在复杂难懂的地方，这里只作简单的归类，具体使用查阅官网便可，再也不赘述。

同步流

• create：new Observable
• of
• range
• repeat
• empty
• never
• throw
• generate

异步流

• interval/timer
• form：string/number/数组/类数组/promise/generator
• formPromise
• formEvent
• formEventPattern
• ajax
• repeatWhen
• defer：订阅时再建立

合并类操做符

订阅多条流，将接收到的数据向下吐出。

concat

首尾链接

• 依次订阅：前一个流完成，再订阅以后的流。

• 当流所有完成时concat流结束。

concat(source1$, source2$)
复制代码

merge

先到先得

• 订阅全部流，任意流吐出数据后，merge流就会吐出数据。

• 对异步数据才有意义。

• 当流所有完成时merge流结束。

merge(source1$, source2$)
复制代码

zip

一对一合并（像拉链同样）

• 订阅全部流，等待全部的流都触发了i次，将第i次的数据合并成数组向下传递。

• 其中一个流完成以后，等待另外一个流的同等数量数据到来后完成zip流。

• 当咱们使用zip时，指望第一次接受到的数据是全部的流第一次发出的数据，第二次接受的是全部的流第二次发出的数据。

zip(source1$, source2$)
复制代码

combineLatest

合并全部流的最后一个数据

• 订阅全部流，任意流触发时，获取其余全部流的最后值合并发出。

• 由于要获取其余流的最后值，因此在刚开始时，必须等待全部流都吐出了值才能开始向下传递数据。

• 全部的流都完成后，combineLatest流才会完成。

combineLatest(source1$, source2$)
复制代码

withLatestFrom

合并全部流的最后一个数据，功能同combineLatest，区别在于：

• combineLatest：当全部流准备完毕后（都有了最后值），任意流触发数据都会致使向下吐出数据。

• withLatestFrom：当全部流准备完毕后（都有了最后值），只有调用withLatestFrom的流吐出数据才会向下吐出数据，其余流触发时仅记录最后值。

source1$.pipe(withLatesFrom(source2$, source3$))
复制代码

race

胜者通吃

• 订阅全部的流，当第一个流触发后，退订其余流。

race(source1$, source2$)
复制代码

startWith

在流的前面填充数据

source1$.pipe(startWith(1))
复制代码

forkJoin

合并全部流的最后一个数据

• 订阅全部流，等待全部流所有完成后，取出全部流的最后值向下发送。

forkJoin(source1$, source2$)
复制代码

辅助类操做符

count

当前流完成以后，统计流一共发出了多少个数据。

source$.pipe(count())
复制代码

mix/max

当前流完成以后，计算 最小值/最大值。

source$.pipe(max())
复制代码

reduce

同数组用法，当前流完成以后，将接受的全部数据依次传入计算。

source$.pipe(reduce(() => {}, 0))
复制代码

布尔类操做符

every

同数组，须要注意的是：若是条件都为true，也要等到流完成才会吐出结果。

缘由也很简单，若是流没有完成，那怎么保证后面的数据条件也为true呢。

source$.pipe(every(() => true/false))
复制代码

find、findIndex

同数组，注意点同every

source$.pipe(find(() => true/false))
复制代码

isEmpty

判断流是否是一个数据都没有吐出就完成了。

source$.pipe(isEmpty())
复制代码

defaultIfEmpty

若是流知足isEmpty，吐出默认值。

source$.pipe(defaultIfEmpty(1))
复制代码

过滤类操做符

filter

同数组

source$.pipe(filter(() => true/false))
复制代码

first

取第一个知足条件的数据，若是不传入条件，就取第一个

source$.pipe(first(() => true/false))
复制代码

last

取第一个知足条件的数据，若是不传入条件，就取最后一个，流完成才会触发。

source$.pipe(last(() => true/false))
复制代码

take

拿够前 N 个就完成

source$.pipe(take(N))
复制代码

takeLast

拿够后N个就结束，由于是后几个因此只有流完成了才会将数据一次发出。

source$.pipe(takeLast(N))
复制代码

takeWhile

给我传判断函数，何时结束你来定

source$.pipe(takeWhile(() => true/false))
复制代码

takeUntil

给我一个流(A)，何时这个流(A)吐出数据了，我就完成

source$.pipe(takeUntil(timer(1000)))
复制代码

skip

跳过前 N 个数据

source$.pipe(skip(N))
复制代码

skipWhile

给我传函数，跳过前几个你来定

source$.pipe(skipWhile(() => true/false))
复制代码

skipUntil

给我一个流(A)，何时这个流(A)吐出数据了，我就不跳了

source$.pipe(skipUntil(timer(1000)))
复制代码

转化类操做符

map

• 接受上游传入的值，返回一个其余的值给下游。（若是你还返回上游的值，那就没有任何意义了）

source$.pipe(map(() => {}))
复制代码

mapTo

• 将传入的值给下游。

source$.pipe(mapTo('a'))
复制代码

pluck

• 提取上游吐出对象的某个key，传给下游。

source$.pipe(pluck('v'))
复制代码

有损回压控制

对防抖、节流不了解的请自行查阅相关说明。

throttle

传入一个流(A)，对上游数据进行节流，直到流(A)吐出数据时结束节流向下传递数据，而后重复此过程

source$.pipe(throttle(interval(1000)))
复制代码

throttleTime

根据时间(ms)节流

source$.pipe(throttleTime(1000))
复制代码

debounce

传入一个流(A)，对上游数据进行防抖，直到流(A)吐出数据时结束防抖向下传递数据，而后重复此过程

source$.pipe(debounce(interval(1000)))
复制代码

debounceTime

根据时间(ms)防抖

source$.pipe(debounceTime(1000))
复制代码

audit

audit 同 throttle，区别在于：

• throttle：将节流期间接受的第一个数据发出
• audit：将节流期间接受的最后一个数据发出

source$.pipe(audit(interval(1000)))
复制代码

auditTime

同上，再也不赘述

source$.pipe(auditTime(1000))
复制代码

sample

• 正常的流，上游触发，下游就会收到数据。

• 使用了sample以后的流，会将上游发出的最新一个数据缓存，而后按照本身的节奏从缓存中取。

• 换句话说，无论上游发出数据的速度是快是慢。sample都无论，他就按照本身的节奏从缓存中取数，若是缓存中有就向下游吐出。若是没有就不作动做。

传入一个流(A)，对上游数据吐出的最新数据进行缓存，直到流(A)吐出数据时从缓存中取出数据向下传递，而后重复此过程

source$.pipe(sample(interval(1000)))
复制代码

sampleTime

根据时间(ms)取数

source$.pipe(sampleTime(1000))
复制代码

distinct

• distinct前缀表示去重操做

全部元素去重，返回当前流中历来没有出现过的数据。

传入函数时，根据函数的返回值分配惟一key。

source$.pipe(distinct())
Observable.of({ age: 4, name: 'Foo'}).pipe(distinct((p) => p.name))
复制代码

distinctUntilChanged

相邻元素去重，只返回与上一个数据不一样的数据。

传入函数时，根据函数的返回值分配惟一key。

source$.pipe(distinctUntilChanged())
复制代码

distinctUntilKeyChanged

• distinctUntilChanged的简化版，帮你实现了取对象key的逻辑。

source$.pipe(distinctUntilKeyChanged('id'))
复制代码

ignoreElements

忽略上游的全部数据，当上游完成时，ignoreElements也会完成。（我不关心你作了什么，只要告诉我完没完成就行）

source$.pipe(ignoreElements())
复制代码

elementAt

只获取上游数据发出的第N个数据。

第二个参数至关于默认值：当上游没发出第N个数据就结束时，发出这个参数给下游。

source$.pipe(elementAt(4, null))
复制代码

single

• 检查上游的全部数据，若是知足条件的数据只有一个，就向下发送这个数据。不然向下传递异常。

source$.pipe(single(() => true/false))
复制代码

无损回压控制

• buffer前缀：将值缓存到数组中，吐出给下游。
• window前缀：将值缓存到一个流中，吐出给下游。

bufferTime、windowTime

缓存上游吐出的数据，到指定时间后吐出，而后重复。

source$.pipe(bufferTime(1000))
复制代码

bufferCount、windowCount

缓存上游吐出的数据，到指定个数后吐出，而后重复。

第二个参数用来控制每隔几个数据开启一次缓存区，不传时可能更符合咱们的认知。

source$.pipe(bufferCount(10))
复制代码

bufferWhen、windowWhen

传入一个返回流(A)的工厂函数

流程以下：

1. 触发订阅时，调用工厂函数拿到流(A)，开始缓存
2. 等待流(A)发出数据时，将缓存的值向下吐出
3. 从新调用工厂函数，拿到一个新的流(A)，开启缓存，循环往复。

randomSeconds = () => timer(Math.random() * 10000 | 0)
source$.pipe(bufferWhen(randomSeconds))
复制代码

bufferToggle、windowToggle

第一个参数为开启缓存流(O)，第二个参数为返回关闭缓存流(C)的工厂函数

流程以下：

1. 当开启流(O)吐出数据时，调用工厂函数获取关闭流(C)，开始缓存
2. 等待关闭流(C)吐出数据后，将缓存的值向下吐出
3. 等待开启流(O)吐出数据，而后重复步骤1

source$.pipe(bufferToggle(interval(1000), () => randomSeconds))
复制代码

buffer、window

传入一个关闭流(C)，区别与bufferWhen：传入的是流，而不是返回流的工厂函数。

触发订阅时，开始缓存，当关闭流(C)吐出数据时，将缓存的值向下传递并从新开始缓存。

source$.pipe(buffer(interval(1000)))
复制代码

累计数据

scan

scan和reduce的区别在于：

• reduce：只有当流完成后才会触发
• scan：每一次流接受到数据后都会触发

区别于其余流，scan拥有了保存、记忆状态的能力。

source$.pipe(scan(() => {}, 0))
复制代码

mergeScan

同scan，可是返回的不是数据而是一个流。

• 当上游吐出数据时，调用规约函数获得并订阅流(A)，将流(A)返回的数据向下游传递，并缓存流(A)返回的最后一个数据。当上游再次吐出数据时，将缓存的最后一个数据传给规约函数，循环往复。

source$.pipe(mergeScan(() => interval(1000)))
复制代码

错误处理

catch

捕获错误

source$.pipe(catch(err => of('I', 'II', 'III', 'IV', 'V')))
复制代码

retry

传入数字 N，遇到错误时，从新订阅上游，重试 N 次结束。

source$.pipe(retry(3))
复制代码

retryWhen

传入流(A)，遇到错误时，订阅流(A)，流(A)每吐出一次数据，就重试一次。流完成，retrywfhen也完成。

source$.pipe(retryWhen((err) => interval(1000)))
复制代码

finally

source$.pipe(finally())
复制代码

多播操做符

multicast

接收返回Subject的工厂函数，返回一个hot observable（HO）

当连接开始时，订阅上游获取数据，调用工厂函数拿到Subject，上游吐出的数据经过Subject进行多播。

• 返回的HO拥有connect、refCount方法。
• 调用connect才会真正开始订阅顶流并发出数据。
• 调用refCount则会根据subscribe数量自动进行connect和unsubscribe操做。
• 多播操做符的老大，较为底层的设计，平常使用很少。
• 后面的多播操做符都是基于此操做符实现。

source$.pipe(multicast(() => new Subject()))
复制代码

publish

• 封装了multicast操做符须要传入Subject工厂函数的操做，其余保持一致。

source$.pipe(publish())
复制代码

share

基于publish的封装，返回调用refCount后的结果（看代码）

source$.pipe(share())
// 等同于
source$.pipe(publish().refCount())
复制代码

publishLast

当上游完成后，多播上游的最后一个数据并完成当前流。

source$.pipe(publishLast())
复制代码

publishReplay

传入缓存数量 N ，缓存上游最新的 N 个数据，当有新的订阅时，将缓存吐出。

• 上游只会被订阅一次。

source$.pipe(publishReplay(1))
复制代码

publishBehavior

缓存上游吐出的最新数据，当有新的订阅时，将最新值吐出。若是被订阅时上游从未吐出过数据，就吐出传入的默认值。

source$.pipe(publishBehavior(0))
复制代码

高阶合并类操做符

• 高阶操做符不是高级操做符
• 当一个流吐出的不是数据，而是一个流时，这就是一个高阶流，就像若是一个函数的返回值仍是一个函数的话，咱们就称之为高阶函数
• 高阶操做符就是操做高阶流的操做符

以下代码示例，顶层的流吐出的并非普通的数据，而是两个会产生数据的流，那么此时下游在接受时，就须要对上游吐出的流进行订阅获取数据，以下：

of(of(1, 2, 3), of(4, 5, 6))
	.subscribe(
		ob => ob.subscribe((num) => {
			console.log(num)
		})
	)
复制代码

上面的代码只是简单的将数据从流中取出，若是我想对吐出的流运用前面讲的操做符应该怎么办？

cache = []
of(of(1, 2, 3), of(4, 5, 6))
	.subscribe({
		next: ob => cache.push(ob),
		complete: {
			concat(...cache).subscribe(console.log)
			zip(...cache).subscribe(console.log)
		}
	})
复制代码

先无论上述实现是否合理，咱们已经能够对上游吐出的流运用操做符了，可是这样实现未免也太过麻烦，因此Rxjs为咱们封装了相关的操做符来帮咱们实现上述的功能。

总结一下：高阶操做符操做的是流，普通操做符操做的是数据。

concatAll

对应concat，缓存高阶流吐出的每个流，依次订阅，当全部流所有完成，concatAll随之完成。

source$.pipe(concatAll())
复制代码

mergeAll

对应merge，订阅高阶流吐出的每个流，任意流吐出数据，mergeAll随之吐出数据。

source$.pipe(mergeAll())
复制代码

zipAll

对应zip，订阅高阶流吐出的每个流，合并这些流吐出的相同索引的数据向下传递。

source$.pipe(zipAll())
复制代码

combineAll

对应combineLatest，订阅高阶流吐出的每个流，合并全部流的最后值向下传递。

source$.pipe(combineAll())
复制代码

高阶切换类操做符

switch

切换流 - 喜新厌旧

高阶流每吐出一个流时，就会退订上一个吐出的流，订阅最新吐出的流。

source$.pipe(switch())
复制代码

exhaust

切换流 - 长相厮守

当高阶流吐出一个流时，订阅它。在这个流没有完成以前，忽略这期间高阶流吐出的全部的流。当这个流完成以后，等待订阅高阶流吐出的下一个流订阅，重复。

source$.pipe(exhaust())
复制代码

高阶Map操做符

看完例子，即知定义。

例子

实现以下功能：

• mousedown事件触发后，监听mousemove事件

普通实现

mousedown$ = formEvent(document, 'mousedown')
mousemove$ = formEvent(document, 'mousemove')

mousedown$.pipe(
	map(() => mousemove$),
	mergeAll()
)
复制代码

1. 当mousedown事件触发后，使用map操做符，将向下吐出的数据转换成mousemove事件流。
2. 因为返回的是流而非数据，因此须要使用mergeAll操做符帮咱们将流中的数据展开。
3. 这样咱们最终接受到的就是mousemove的event事件对象了。

注：因为只有一个事件流，因此使用上面介绍的任意高阶合并操做符都是同样的效果。

高阶Map实现

mousedown$.pipe(
  mergeMap(() => mousemove$)
)
复制代码

不难看出，所谓高阶map，就是

concatMap 	= map + concatAll
mergeMap 		= map + mergeAll
switchMap 	= map + switch
exhaustMap 	= map + exhaust
concatMapTo = mapTo + concatAll
mergeMapTo 	= mapTo + mergeAll
switchMapTo = mapTo + switch
复制代码

expand

相似于mergeMap，可是，全部传递给下游的数据，同时也会传递给本身，因此expand是一个递归操做符。

source$.pipe(expand(x => x === 8 ? EMPTY : x * 2))
复制代码

数据分组

groupBy

输出流，将上游传递进来的数据，根据key值分类，为每个分类建立一个流传递给下游。

key值由第一个函数参数来控制。

source$.pipe(groupBy(i => i % 2))
复制代码

partition

groupBy的简化版，传入判断条件，知足条件的放入第一个流中，不知足的放入第二个流中。

简单说：

• groupBy根据key的分类，可能会向下传递N条流。
• partition只会向下传递两条流：知足条件的和不知足条件的。

source$.pipe(partition())
复制代码

结语

以上就是本文的所有内容了，但愿你看了会有收获。

若是有不理解的部分，能够在评论区提出，你们一块儿成长进步。

祝你们早日拿下Rxjs这块难啃的骨头。

参考资料

• 官方文档
• 《深刻浅出Rxjs》