RxJS快速入门

摘要

RxJS 是一个库，它通过使用 observable 序列来编写异步和基于事件的程序。它提供了一个核心类型 Observable，附属类型 (Observer、 Schedulers、 Subjects) 和受 [Array#extras] 启发的操作符 (map、filter、reduce、every, 等等)，这些数组操作符可以把异步事件作为集合来处理。

内容导航

RxJS是什么

RxJS 是一个库，它通过使用 observable 序列来编写异步和基于事件的程序。它提供了一个核心类型 Observable，附属类型 (Observer、 Schedulers、 Subjects) 和受 [Array#extras] 启发的操作符 (map、filter、reduce、every, 等等)，这些数组操作符可以把异步事件作为集合来处理。

可以把 RxJS 当做是用来处理事件的 Lodash 。

ReactiveX 结合了 观察者模式、迭代器模式 和 使用集合的函数式编程，以满足以一种理想方式来管理事件序列所需要的一切。

RxJS的主要成员

• Observable (可观察对象): 表示一个概念，这个概念是一个可调用的未来值或事件的集合。
• Observer (观察者): 一个回调函数的集合，它知道如何去监听由 Observable 提供的值。
• Subscription (订阅): 表示 Observable 的执行，主要用于取消 Observable 的执行。
• Operators (操作符): 采用函数式编程风格的纯函数 (pure function)，使用像 map、filter、concat、flatMap 等这样的操作符来处理集合。
• Subject (主体): 相当于 EventEmitter，并且是将值或事件多路推送给多个 Observer 的唯一方式。
• Schedulers (调度器): 用来控制并发并且是中央集权的调度员，允许我们在发生计算时进行协调，例如 setTimeout 或 requestAnimationFrame 或其他。

Observable (可观察对象)

RxJS 是基于观察者模式和迭代器模式以函数式编程思维来实现的。RxJS 中含有两个基本概念：Observables 与 Observer。Observables 作为被观察者，是一个值或事件的流集合；而 Observer 则作为观察者，根据 Observables 进行处理。Observables 是多个值的惰性推送集合。

• of():用于创建简单的Observable，该Observable只发出给定的参数，在发送完这些参数后发出完成通知.
• from():从一个数组、类数组对象、promise、迭代器对象或者类Observable对象创建一个Observable.
• fromEvent(),:把event转换成Observable.
• range():在指定起始值返回指定数量数字.
• interval():基于给定时间间隔发出数字序列。返回一个发出无限自增的序列整数，可以选择固定的时间间隔进行发送。
• timer():创建一个Observable，该Observable在初始延时之后开始发送并且在每个时间周期后发出自增的数字

创建 Observable

import { Observable,of,from,fromEvent,range,interval } from 'rxjs'; import { map } from 'rxjs/operators';  	const Observable1 = new Observable(subscriber => {     try{           subscriber.next(1);           subscriber.next(2);           subscriber.next(3);           setTimeout(() => {             subscriber.next(4);             subscriber.complete();           }, 1000);         } catch (err) {         	subscriber.error(err);	//传递一个错误对象，如果捕捉到异常的话。     	}     });     const Observable2 = from([       { name: 'Dave', age: 34, salary: 2000 },       { name: 'Nick', age: 37, salary: 32000 },       { name: 'Howie', age: 40, salary: 26000 },       { name: 'Brian', age: 40, salary: 30000 },       { name: 'Kevin', age: 47, salary: 24000 },     ]); 	const Observable3 = of("Dave","Nick");//把所有参数组合到数组,逐个提供给消费者 	const Observable4 = range(1,10); 	const Observable5 = interval(3000);//从零开始每3000毫秒自增并提供给消费者 	const Observable6 = timer(3000,1000);//等待3000毫秒后,从零开始每1000毫秒自增并提供给消费者 

订阅 Observables

因为 Observable 执行可能会是无限的，并且观察者通常希望能在有限的时间内中止执行，所以我们需要一个 API 来取消执行。因为每个执行都是其对应观察者专属的，一旦观察者完成接收值，它必须要一种方法来停止执行，以避免浪费计算能力或内存资源。

当调用了 observable.subscribe ，观察者会被附加到新创建的 Observable 执行中。这个调用还返回一个对象，即 Subscription (订阅)：

import { Observable,of,from,fromEvent,range,interval } from 'rxjs'; 	const observable1 = range(1,10);     observable1.subscribe(           num => {             console.log(num);           },           err => console.log(err),           () => console.log("Streaming is over.")         ); 

执行 Observables

Observable 执行可以传递三种类型的值：

• "Next" 通知： 发送一个值，比如数字、字符串、对象，等等。
• "Error" 通知： 发送一个 JavaScript 错误 或 异常。
• "Complete" 通知： 不再发送任何值。

"Next" 通知是最重要，也是最常见的类型：它们表示传递给观察者的实际数据。"Error" 和 "Complete" 通知可能只会在 Observable 执行期间发生一次，并且只会执行其中的一个。

import { Observable,of,from,fromEvent,range,interval } from 'rxjs'; import { map } from 'rxjs/operators';  	const observable = new Observable(subscriber => {     try{           subscriber.next(1);           subscriber.next(2);           subscriber.next(3);           subscriber.complete();           subscriber.next(4); // 因为违反规约，所以不会发送         } catch (err) {         	subscriber.error(err);	//传递一个错误对象，如果捕捉到异常的话。     	}     }); 

清理 Observable 执行

因为 Observable 执行可能会是无限的，并且观察者通常希望能在有限的时间内中止执行，所以我们需要一个 API 来取消执行。因为每个执行都是其对应观察者专属的，一旦观察者完成接收值，它必须要一种方法来停止执行，以避免浪费计算能力或内存资源

当你订阅了 Observable，你会得到一个 Subscription ，它表示进行中的执行。只要调用 unsubscribe() 方法就可以取消执行。

import { Observable,of,from,fromEvent,range,interval } from 'rxjs';     const observable = new Observable(subscriber => {       let intervalID = setInterval(() => {         subscriber.next('hi');       }, 1000);       // 提供取消和清理 interval 资源的方法       return function unsubscribe() {         clearInterval(intervalID);       };     });     let subscription = observable.subscribe(x => console.log(x));     subscription.unsubscribe(); 

Observer (观察者)

观察者是由 Observable 发送的值的消费者。观察者只是一组回调函数的集合，每个回调函数对应一种 Observable 发送的通知类型：next、error 和 complete 。下面的示例是一个典型的观察者对象：

观察者只是有三个回调函数的对象，每个回调函数对应一种 Observable 发送的通知类型。

observable.subscribe(     next: x => console.log('Observer got a next value: ' + x),     error: err => console.error('Observer got an error: ' + err),     complete: () => console.log('Observer got a complete notification') ); 

Subscription (订阅)

Subscription 是表示可清理资源的对象，通常是 Observable 的执行。Subscription 有一个重要的方法，即 unsubscribe，它不需要任何参数，只是用来清理由 Subscription 占用的资源。在上一个版本的 RxJS 中，Subscription 叫做 "Disposable" (可清理对象)。

Subscription 基本上只有一个 unsubscribe() 函数，这个函数用来释放资源或去取消 Observable 执行。

import { Observable,of,from,fromEvent,range,interval } from 'rxjs'; 	var observable1 = interval(1000);     var subscription1 = observable1.subscribe(x => console.log(x));     // 稍后：     // 这会取消正在进行中的 Observable 执行     // Observable 执行是通过使用观察者调用 subscribe 方法启动的     subscription1.unsubscribe();      var observable2 = interval(400);     var observable3 = interval(300);     var subscription2 = observable2.subscribe(x => console.log('first: ' + x));     var childSubscription = observable3.subscribe(x => console.log('second: ' + x));     subscription2.add(childSubscription);     setTimeout(() => {       // subscription 和 childSubscription 都会取消订阅       subscription2.unsubscribe();     }, 1000); 

Subject (主体)

RxJS Subject 是一种特殊类型的 Observable，它允许将值多播给多个观察者，所以 Subject 是多播的，而普通的 Observables 是单播的(每个已订阅的观察者都拥有 Observable 的独立执行)。

Subject 像是 Observable，但是可以多播给多个观察者。Subject 还像是 EventEmitters，维护着多个监听器的注册表。

还有一些特殊类型的 Subject：BehaviorSubject、ReplaySubject 和 AsyncSubject。

每个 Subject 都是 Observable 。 - 对于 Subject，你可以提供一个观察者并使用 subscribe 方法，就可以开始正常接收值。从观察者的角度而言，它无法判断 Observable 执行是来自普通的 Observable 还是 Subject 。

在 Subject 的内部，subscribe 不会调用发送值的新执行。它只是将给定的观察者注册到观察者列表中，类似于其他库或语言中的 addListener 的工作方式。

每个 Subject 都是观察者。 - Subject 是一个有如下方法的对象： next(v)、error(e) 和 complete() 。要给 Subject 提供新值，只要调用 next(theValue)，它会将值多播给已注册监听该 Subject 的观察者们。

import { Subject,from } from 'rxjs'; 	//我们为 Subject 添加了两个观察者，然后给 Subject 提供一些值  	var subject1 = new Subject();     subject1.subscribe({       next: (v) => console.log('observerA: ' + v)     });     subject1.subscribe({       next: (v) => console.log('observerB: ' + v)     });     subject1.next(1);     subject1.next(2);     //因为 Subject 是观察者，这也就在意味着你可以把 Subject 作为参数传给任何 Observable 的 subscribe 方法     var subject2 =new Subject();     subject2.subscribe({       next: (v) => console.log('observerA: ' + v)     });     subject2.subscribe({       next: (v) => console.log('observerB: ' + v)     });     var observable = from([1, 2, 3]);     observable.subscribe(subject2); // 你可以提供一个 Subject 进行订阅 

多播的 Observables

“多播 Observable” 通过 Subject 来发送通知，这个 Subject 可能有多个订阅者，然而普通的 “单播 Observable” 只发送通知给单个观察者。

多播 Observable 在底层是通过使用 Subject 使得多个观察者可以看见同一个 Observable 执行。

在底层，这就是 multicast 操作符的工作原理：观察者订阅一个基础的 Subject，然后 Subject 订阅源 Observable 。

import { Subject } from 'rxjs/internal/Subject'; import { take, multicast } from 'rxjs/operators'; 	     const source = timer(1000, 2500).pipe(take(5));     const subject = new Subject();     subject.subscribe({       next: (v) => console.log('observerC: ' + v)     });     subject.subscribe({       next: (v) => console.log('observerD: ' + v)     });     const multicasted = source.pipe(multicast(subject));     multicasted.subscribe({       next: (v) => console.log('observerA: ' + v)     });     multicasted.subscribe({       next: (v) => console.log('observerB: ' + v)     }); 	source.subscribe(subject); 

BehaviorSubject

Subject 的其中一个变体就是 BehaviorSubject，它有一个“当前值”的概念。它保存了发送给消费者的最新值。并且当有新的观察者订阅时，会立即从 BehaviorSubject 那接收到“当前值”。

BehaviorSubjects 适合用来表示“随时间推移的值”。举例来说，生日的流是一个 Subject，但年龄的流应该是一个 BehaviorSubject 。

import { BehaviorSubject } from 'rxjs'; 	//BehaviorSubject 使用值0进行初始化，当第一个观察者订阅时会得到0。第二个观察者订阅时会得到值2，尽管它是在值2发送之后订阅的。 	const subject = new BehaviorSubject(0); // 0是初始值     subject.subscribe({       next: (v) => console.log('observerA: ' + v)     });      subject.next(1);     subject.next(2);      subject.subscribe({       next: (v) => console.log('observerB: ' + v)     });      subject.next(3); 

ReplaySubject

ReplaySubject 类似于 BehaviorSubject，它可以发送旧值给新的订阅者，但它还可以记录 Observable 执行的一部分。

ReplaySubject 记录 Observable 执行中的多个值并将其回放给新的订阅者。

除了缓冲数量，你还可以指定 window time (以毫秒为单位)来确定多久之前的值可以记录。

import { ReplaySubject } from 'rxjs'; 	const subject = new ReplaySubject(3); // 为新的订阅者缓冲最后3个值     subject.subscribe({       next: (v) => console.log('observerA: ' + v)     });     subject.next(1);     subject.next(2);     subject.next(3);     subject.next(4);     subject.subscribe({       next: (v) => console.log('observerB: ' + v)     });     subject.next(5);  	//我们缓存数量100，但 window time 参数只设置了120毫秒     const subject = new ReplaySubject(100, 120 /* windowTime */);      subject.subscribe({       next: (v) => console.log('observerA: ' + v)     });     let i = 1;     setInterval(() => subject.next(i++), 200);     setTimeout(() => {       subject.subscribe({         next: (v) => console.log('observerB: ' + v)       });     }, 1000);	 

AsyncSubject

AsyncSubject 是另一个 Subject 变体，只有当 Observable 执行完成时(执行 complete())，它才会将执行的最后一个值发送给观察者。

AsyncSubject 和 last() 操作符类似，因为它也是等待 complete 通知，以发送一个单个值。

import { AsyncSubject } from 'rxjs';     const subject = new AsyncSubject();      subject.subscribe({       next: (v) => console.log('observerA: ' + v)     });      subject.next(1);     subject.next(2);     subject.next(3);     subject.next(4);      subject.subscribe({       next: (v) => console.log('observerB: ' + v)     });      subject.next(5);     subject.complete(); 

Scheduler (调度器)

调度器控制着何时启动 subscription 和何时发送通知。它由三部分组成：

• 调度器是一种数据结构。 它知道如何根据优先级或其他标准来存储任务和将任务进行排序。
• 调度器是执行上下文。 它表示在何时何地执行任务(举例来说，立即的，或另一种回调函数机制(比如 setTimeout 或 process.nextTick)，或动画帧)。
• 调度器有一个(虚拟的)时钟。 调度器功能通过它的 getter 方法 now() 提供了“时间”的概念。在具体调度器上安排的任务将严格遵循该时钟所表示的时间。

调度器可以让你规定 Observable 在什么样的执行上下文中发送通知给它的观察者。

import { asyncScheduler, Observable } from 'rxjs'; 	//我们使用普通的 Observable ，它同步地发出值`1`、`2`、`3`，并使用操作符 `observeOn` 来指定 `async` 调度器发送这些值。 	const observable = new Observable(subscriber => {       subscriber.next(1);       subscriber.next(2);       subscriber.next(3);       subscriber.complete();     })       .pipe(         observeOn(asyncScheduler)       );      console.log('just before subscribe');     observable.subscribe({       next: x => console.log('got value ' + x),       error: err => console.error('something wrong occurred: ' + err),       complete: () => console.log('done'),     });     console.log('just after subscribe');     //你会发现"just after subscribe"在"got value..."之前就出现了     //just before subscribe     //just after subscribe     //got value 1     //got value 2     //got value 3     //done      

调度器类型

async 调度器是 RxJS 提供的内置调度器中的一个。可以通过使用 Scheduler 对象的静态属性创建并返回其中的每种类型的调度器。

调度器	目的
null	不传递任何调度器的话，会以同步递归的方式发送通知。用于定时操作或尾递归操作。
queueScheduler	当前事件帧中的队列调度(蹦床调度器)。用于迭代操作。
asapScheduler	微任务的队列调度，它使用可用的最快速的传输机制，比如 Node.js 的 process.nextTick() 或 Web Worker 的 MessageChannel 或 setTimeout 或其他。用于异步转换。
asyncScheduler	使用 setInterval 的调度。用于基于时间的操作符。
animationFrameScheduler	计划将在下一次浏览器内容重新绘制之前发生的任务。 可用于创建流畅的浏览器动画。

Pipeable(操作符)

操作符就是函数,管道操作符本质上是一个纯函数，它将一个Observable作为输入并生成另一个Observable作为输出。订阅输出Observable也将订阅输入Observable。 操作符有两种：

管道操作符是一个将Observable作为其输入并返回另一个Observable的函数。这是一个纯粹的操作：以前的Observable保持不变。

1. 管道操作符是可以使用语法observableInstance.pipe(operator())传递给Observable的类型。 这些包括filter()和mergeMap()。 调用时，它们不会更改现有的Observable实例。 相反，它们返回一个新的Observable，其订阅逻辑基于第一个Observable。

2. 创建运算符是另一种运算符，可以称为独立函数来创建新的Observable。例如：of(1,2,3)创建一个observable ，该对象将依次发射1、2和3。创建运算符将在后面的部分中详细讨论。

obs.pipe(   op1(),   op2(),   op3(),   op3(), ) 

常用的操作符

finalize<T>(callback: () => void): MonoTypeOperatorFunction<T>:

返回原始Observable，但在Observable完成或发生错误终止时将调用指定的函数。

创建操作符

• ajax
• bindCallback
• bindNodeCallback
• defer
• empty
• from
• fromEvent
• fromEventPattern
• generate
• interval
• of
• range
• throwError
• timer
• iif

连接创建操作符

These are Observable creation operators that also have join functionality -- emitting values of multiple source Observables.

• combineLatest
• concat
• forkJoin
• merge
• partition
• race
• zip

转换操作符

• buffer
• bufferCount
• bufferTime
• bufferToggle
• bufferWhen
• concatMap
• concatMapTo
• exhaust
• exhaustMap
• expand
• groupBy
• map
• mapTo
• mergeMap
• mergeMapTo
• mergeScan
• pairwise
• partition
• pluck
• scan
• switchMap
• switchMapTo
• window
• windowCount
• windowTime
• windowToggle
• windowWhen

过滤操作符

• audit
• auditTime
• debounce
• debounceTime
• distinct
• distinctKey
• distinctUntilChanged
• distinctUntilKeyChanged
• elementAt
• filter
• first
• ignoreElements
• last
• sample
• sampleTime
• single
• skip
• skipLast
• skipUntil
• skipWhile
• take
• takeLast
• takeUntil
• takeWhile
• throttle
• throttleTime

组合操作符

Also see the Join Creation Operators section above.

• combineAll
• concatAll
• exhaust
• mergeAll
• startWith
• withLatestFrom

多播操作符

• multicast
• publish
• publishBehavior
• publishLast
• publishReplay
• share

错误处理操作符

• catchError
• retry
• retryWhen

工具操作符

• tap
• delay
• delayWhen
• dematerialize
• materialize
• observeOn
• subscribeOn
• timeInterval
• timestamp
• timeout
• timeoutWith
• toArray

条件和布尔操作符

• defaultIfEmpty
• every
• find
• findIndex
• isEmpty

数学和聚合操作符

• count
• max
• min
• reduce