RxJS——异步数据流的响应式编程库（适合新手入门）

文章目录

RxJS概述
Redux VS RxJS
RxJS核心概念解析
热观察和冷观察
merge/combine合流
RXJS6 的变化

RxJS概述

RxJS 全称 Reactive Extensions for JavaScript

RxJS 结合了函数式编程、观察者模式（例如 DOM EventListener）、迭代器模式（例如 ES6 Iterater）

RxJS 官方是这样说的: Think of RxJS as Lodash for events. 把 RxJS 想像成针对 events 的 lodash

RxJS 本质是个工具库，处理的是事件，这里的 events，可以称之为流

那么流是指什么呢？举个例子，代码中每 1s 输出一个数字，用户每一次对元素的点击，就像是在时间这个维度上，产生了一个数据集。这个数据集不像数组那样，它不是一开始都存在的，而是随着时间的流逝，数据一个一个被输出。这种异步行为产生的数据，就可以被称之为一个流。在 RxJS 中，称之为 ovservalbe（抛开英文，本质其实就是一个数据的集合，只是这些数据不一定是一开始就设定好的，而是随着时间而不断产生的）

而 RxJS，就是为了处理这种流而产生的工具，比如流与流的合并、流的截断、延迟、消抖等

Redux VS RxJS

RxJS核心概念解析

Observable

它的本质其实就是一个随时间不断产生数据的一个集合，称之为流更容易理解。而其对象存在一个 subscribe 方法，调用该方法后，才会启动这个流（也就是数据才会开始产生），这里需要注意的是多次启动的每一个流都是独立的，互不干扰。

Observer

从行为上来看，无非就是定义了如何处理上述流产生的数据，称之为流的处理方法。

Subscribtion

它的本质就是暂存了一个启动后的流，之前提到，每一个启动后的流都是相互独立的，而这个启动后的流，就存储在 subscription 中，提供了 unsubscribe，来停止这个流。

简单理解了这三个名词 observable, observer, subscription 后，从数据的角度来思考：

observable 定义了要生成一个什么样的数据，其 subscribe 方法，接收一个 observer（定义了接收到数据如何处理），并开始产生数据。该方法的返回值 subscription, 存储了这个已经开启的流，同时具有 unscbscribe 方法，可以将这个流停止。

整理成这个公式：

Subscription = Observable.subscribe (observer)

observable: 随着时间产生的数据集合，可以理解为流，其 subscribe 方法可以启动该流

observer: 决定如何处理数据

subscription: 存储已经启动过的流，其 unsubscribe 方法可以停止该流

Subject

它是一个代理对象，既是一个 Observable 又是一个 Observer，它可以同时接受 Observable 发射出的数据，也可以向订阅了它的 observer 发射数据，同时，Subject 会对内部的 observers 清单进行多播 (multicast)

Subjects 是将任意 Observable 执行共享给多个观察者的唯一方式

Subject 有三个变体：

BehaviorSubject是一种在有新的订阅时会额外发出最近一次改变的值的 Subject，需要传入一个参数即初始值（如果没改变则发送初始值）
ReplaySubject会保存所有值，然后回放给新的订阅者，需要传入一个参数即初始值，用于控制重放值的数量（默认重放所有）
AsyncSubject只有当 Observable 执行完成时 (执行 complete())，才会将执行的最后一个值发送给观察者。如果因异常而终止，AsyncSubject 将不会释放任何数据，但是会向 Observer 传递一个异常通知。

Scheduler

用来控制并发并且是中央集权的调度员，允许在发生计算时进行协调，例如 setTimeout 或 requestAnimationFrame 或其他。

调度器是一种数据结构。它知道如何根据优先级或其他标准来存储任务和将任务进行排序。

调度器是执行上下文。它表示在何时何地执行任务 (举例来说，立即的，或另一种回调函数机制 (比如 setTimeout 或 process.nextTick)，或动画帧)。

调度器有一个 (虚拟的) 时钟。调度器功能通过它的 getter 方法 now() 提供了 “时间” 的概念。在具体调度器上安排的任务将严格遵循该时钟所表示的时间。

Scheduler调度器能做四种调度：

queue：将每个下一个任务放在队列中，而不是立即执行。queue 延迟使用调度程序时，其行为与 async 调度程序相同。当没有延迟使用时，它将同步安排给定的任务 - 在安排好任务后立即执行。但是，当递归调用时（即在已调度的任务内部），将使用队列调度程序调度另一个任务，而不是立即执行，该任务将被放入队列并等待当前任务完成。
asap：内部基于 Promise 实现（Node 端采用 process.nextTick），会使用可用的最快的异步传输机制，如果不支持 Promise 或 process.nextTick 或者 Web Worker 的 MessageChannel 也可能会调用 setTimeout 方式进行调度。
async：与 asap 方式很像，只不过内部采用 setInterval 进行调度，大多用于基于时间的操作符。
animationFrame：内部基于 requestAnimationFrame 来实现调度，所以执行的时机将与 window.requestAnimationFrame 保持一致，适用于需要频繁渲染或操作动画的场景。

Operators

RxJS操作符非常多，在此只介绍几个常用的：

create

create将 onSubscription 函数转化为一个实际的 Observable 。每当有人订阅该 Observable 的时候，onSubscription 函数会接收 Observer 实例作为唯一参数行。onSubscription 应该调用观察者对象的 next, error 和 complete 方法。

const source = Rx.Observable.create(((observer: any) => {observer.next(1);observer.next(2);setTimeout(() => {observer.next(3);}, 1000)
}))// 方式一
source.subscribe({next(val) {console.log('A：' + val);}}
);
// 方式二
source.subscribe((val) => console.log('B：' + val));// A：1
// A：2
// B：1
// B：2
//- 1s后:
// A：3
// B：3

from

从一个数组、类数组对象、Promise、迭代器对象或者类 Observable 对象创建一个 Observable。该方法就有点像 js 中的 Array.from 方法（可以从一个类数组或者可迭代对象创建一个新的数组），只不过在 RxJS 中是转成一个 Observable 给使用者使用。
```
const source = Rx.Observable.from([10, 20, 30]);
source.subscribe(v => console.log(v));// 10
// 20
// 30
```
of

与 from 的能力差不多，只不过在使用的时候是传入一个一个参数来调用的，有点类似于 js 中的 concat 方法。同样也会返回一个 Observable，它会依次将你传入的参数合并并将数据以同步的方式发出。
```
const source = Rx.Observable.of(1, 2, 3);
source.subscribe(v => console.log(v));// 1
// 2
// 3
```

debounceTime

功能与 debounce 防抖函数差不多，只有在特定的一段时间经过后并且没有发出另一个源值，才从源 Observable 中发出一个值。

假设一个数据源每隔一秒发送一个数，而我们使用了 debounceTime 操作符，并设置了延时时间，那么在数据源发送一个新数据之后，如果在延时时间内数据源又发送了一个新数据，这个新的数据就会被先缓存住不会发送，等待发送完数据之后并等待延时时间结束才会发送给订阅者。不仅如此，在延时时间未到的时候并且已有一个值在缓冲区，这个时候又收到一个新值，那么缓冲区就会把老的数据抛弃放入新的，然后重新等待延时时间到达然后将其发送。

const source = Rx.Observable.interval(1000).take(3);
const result = source.debounceTime(2000);
result.subscribe(x => console.log(x));// 程序启动之后的前三秒没有数据打印，等到五秒到了之后，打印出一个2，接着就没有再打印了
// 数据源会每秒依次发送三个数 0、1、2，由于我们设定了延时时间为 2 秒，那么也就是说，我们在数据发送完成之前都是不可能看到数据的，因为发送源的发送频率为 1 秒，延时时间却有两秒，也就是除非发送完，否则不可能满足发送源等待两秒再发送新数据，每次发完新数据之后要等两秒之后才会有打印，所以不论我们该数据源发送多少个数，最终订阅者收到的只有最后一个数。

take

只发出源 Observable 最初发出的 N 个值 (N = count)

这个操作符在前面出现了很多次了，还挺常见的，用于控制只获取特定数目的值，跟 interval 这种会持续发送数据的配合起来就能自主控制要多少个值了。

skip

返回一个 Observable，该 Observable 跳过源 Observable 发出的前 N 个值 (N = count)。

假设这个数据源发送 6 个值，可以使用 skip 操作符来跳过前多少个。

const source = Rx.Observable.from([1, 2, 3, 2, 4, 3]);
const result = source.skip(2);
result.subscribe(x => console.log(x));// 打印结果为：3、2、4、3，跳过了前面两个数。

concat

concat和 concatAll效果是一样的，区别在于 concat 要传递参数，参数必须是 Observable 。

concat 将多个 observable 串接起来，前一个完成好了再执行下一个。

const source1 = interval(1000).pipe(take(3));
const source2 = of(3);
const source3 = of (4,5);
const example = source1.pipe(concat(source2,source3))
example.subscribe({next: value => {console.log(value);},error: err => {console.log("Error: " + err);},complete: () => {console.log("complete");}
});// 0
// 1
// 2
// 3
// 4
// 5
// complete

热观察和冷观察

在 RxJS 中，有热观察和冷观察的概念。其中的区别：

Hot Observable：可以理解为现场直播，我们进场的时候只能看到即时的内容
Cold Observable：可以理解为点播（电影），我们打开的时候会从头播放

RxJS 中 Observable 默认为冷观察，而通过 publish() 和 connect() 可以将冷的 Observable 转变成热的

let publisher$ = Rx.Observable.interval(1000).take(5).publish();publisher$.subscribe(data => console.log('subscriber from first minute',data),err => console.log(err),() => console.log('completed')
)setTimeout(() => {publisher$.subscribe(data => console.log('subscriber from 2nd minute', data),err => console.log(err),() => console.log('completed'))
}, 3000)publisher$.connect();
// 第一个订阅者输出的是0,1,2,3,4，而第二个输出的是3,4，此处为热观察

热观察和冷观察根据具体的场景可能会有不同的需要，而 Observable 提供的缓存能力也能解决不少业务场景。

例如，如果我们想要在拉群后，自动同步之前的聊天记录，通过冷观察就可以做到。

merge/combine合流

一般来说，合流有两种方式：

// merge
--1----2-----3--------4---
----a-----b----c---d------merge
--1-a--2--b--3-c---d--4---// combine
--1----2-----3--------4---
----a-----b-----c--d------combine
--1a-2a-2b-3b-3c-3d-4d--

merge 的合流方式可以用在聊天室、多人协作、公众号订阅就可以通过这样的方式合流，最终按照顺序地展示出对应的操作记录。

在 Excel 中，通过函数计算了 A1 和 B2 两个格子的相加。这种情况下可以使用 combine 合流：

const streamA1 = Rx.Observable.fromEvent(inputA1, "input"); // 监听 A1 单元格的 input 事件
const streamB2 = Rx.Observable.fromEvent(inputB2, "input"); // 监听 B2 单元格的 input 事件const subscribe = combineLatest(streamA1, streamB2).subscribe((valueA1, valueB2) => {// 从 streamA1 和 streamB2 中获取最新发出的值return valueA1 + valueB2;
});
// 获取函数计算结果
observable.subscribe((x) => console.log(x));

在一个较大型的前端应用中，通常会拆分成渲染层、数据层、网络层、其他服务等多个功能模块。

虽然服务按照功能结构进行拆分了，但依然会存在服务间调用导致依赖关系复杂、事件触发和监听满天飞等情况。这种情况下，只能通过全局搜索关键字来找到上下游数据流、信息流，通过节点和关键字搜索才能大概理清楚某个数据来源哪里。

如果使用了响应式编程，我们可以通过各种合流的方式、订阅分流的方式，来将应用中的数据流动串在一起。这样，我们可以很清晰地当前节点上的数据来自于哪里，是用户的操作还是来自网络请求。

RXJS6 的变化

RXJS6 改变了包的结构，主要变化在 import 方式和 operator 以及使用 pipe ()

Imports 方式改变

从 rxjs 中类似像导入 observable subject 等的不再进一步导入，而是止于 rxjs, rxjs6 在包的结构上进行了改变

operator 的改变

总而言之：类似于创建之类的用的 API 都是从 rxjs 引入的，类似于 map 之类的操作都是从 rxjs/operators 引入的

pipeable observable

被重新命名的 API

在这里只是简单介绍 RxJS ，帮助新手快速了解 RxJS ，更详细的 RxJS 学习可以查看官方文档