335次阅读 | 发布于2年以前

RxJS 入门及应用

RxJS为什么被称为是函数响应式编程？

函数式编程（Functional Programming）

函数式编程的特点是允许把函数当成一个实参或返回值，主要思想是想将复杂的运算分解成一系列嵌套的函数，逐层推导，不断渐进，直至完成运算。常用的数组方法（map，filter等）就运用了函数式编程的思想。

const arr = [4,1,5,2,3];
const newArr = arr
  .sort((a,b) => a-b)
  .filter(value => value>2);
console.log(newArr);   // [3,4,5]

函数式编程还有一个最重要的特性，那就是纯净性Purity（纯函数）

纯函数是不会产生副作用的函数，其中输出完全由输入决定，也就是说无论调用多少次，调用f(x)都会得到相同的结果。

函数副作用：指当调用函数时，除了返回函数值之外，还对主调用函数产生附加的影响。例如修改全局变量（函数外的变量），修改参数或改变外部存储。

纯函数满足以下两个条件

①函数的执行过程完全由输入参数决定，不会受除参数之外的任何数据的影响。

②函数不会修改任何外部状态，比如修改全局变量或传入的参数对象。

通常如果创建一个非纯函数，在这个函数之外使用了共享变量的代码，会使应用状态混乱难以维护。

var count = 0;
var button = document.querySelector('button');
button.addEventListener('click', 
  () => console.log(`Clicked ${++count} times`)
);

而使用RxJS，可以将应用状态隔离出来，不会被外部环境影响也不会影响外部环境。

import { fromEvent, scan } from 'rxjs'; // 将事件转换成 observable 序列。
var button = document.querySelector('button');
const example = fromEvent(button, 'click').pipe(
  scan(count => count + 1, 0); //工作原理与数组的 reduce 类似,随着时间的推移进行归并
);
example.subscribe(count => console.log(`Clicked ${count} times`));

响应式编程（Reactive Programming）

wiki百科中的解释：

在计算中，响应式编程或反应式编程（Reactive programming）是一种面向数据流和变化传播的声明式编程范式。这意味着可以在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值通过数据流进行传播。

什么是数据流？

数据流（data stream）是数据在系统内传播的路径，表示在一定时间范围内发生的一系列事件。

任何东西都可以是一个 Stream：变量、用户输入、网络响应、定时器、数据结构等等。

什么是变化传播？

在数据流传播的过程中，可能会有一些事件去组合、创建、过滤这些 Streams，从一个旧的stream映射成一个新的stream。我们不需要去轮询变化，而是对事件进行监听，在执行一个事件后，会自动做出相应的响应，这就是变化传播。

前端框架与rxjs的结合

前端框架的职责（例如react）：数据与UI视图的同步，数据发生更新时，视图随之更新；

UI = f(data);

响应式编程的职责（例如rxjs）：聚焦于数据，从数据的源头开始，到数据的处理变化，再到数据流的订阅，数据的消费；

data = g(origin data)

两者关系看起并不冲突，并且在某些场景下结合使用可能会为我们带来便捷，前端框架可以作为响应式编程数据的一个消费者；

UI = f(g(origin data))

RxJS是用来干什么的？

RxJS是一个用于处理异步事件流的库，通过使用 observable 序列来编写异步和基于事件的程序，实际应用场景就是把请求封装成observerable，通过一些基本操作符（map、filter等等）将返回的数据处理并且catch错误，将异步事件作为集合来处理。RxJS实际上是将开发过程中遇到的异步（多为异步，同步也可以）操作看为一个事件流，RxJS内部封装了对一个事件流的操作符（创建、转换、组合、过滤、错误异常处理等），组合使用这些操作符来以更便利的方式来管理事件。

为什么用RxJS，摘自知乎回答：

思考一下，异步的本质是什么？

异步操作和同步操作最大的区别就是异步有时序。

我们可以把同步操作理解为：数据+函数

那么异步操作就是：数据+函数+时序

Rx就是把时序抽离成一根时间轴，在这根时间轴上进行同步操作，而异步相关的时序处理就交给Rx提供的各种operator操作符。

所以问题就很简单了，如果你的应用是一个时序密集的应用，那么使用Rx能帮你理清复杂的异步逻辑。反之，如果异步操作之间没有太多的联系，时序分散， 则不那么需要使用Rx。

RxJS中解决异步事件管理的基本概念

1. Observable（可观察对象）

将一个数据流看作一个可观察对象，表示这个数据流变化传播过程中发生的一些事件的集合。

	单个值	多个值
拉取(pull)	Function	Iterator
推送(push)	Promise	Observable

拉取和推送是两种不同的协议，用来描述数据生产者 (Producer) 与数据消费者 (Consumer) 如何通信。

1 . 拉取体系

js中每个函数function都属于拉取体系，函数来生产数据，消费者通过调用该函数的代码来从函数中获取单个返回值来对该函数进行消费，而迭代器Iterator则是消费者调用iterator.next()来获取多个返回值进行消费。

拉取的过程中，生产者是一个被动的过程，在消费者请求调用自己时才产生数据，消费者是一个主动的过程，消费者自己来决定何时调用生产者来获取收据。

2 . 推送体系

在如今的js中，Promise是最常见的推送体系，Promise作为生产者，将解析过的resolved值传给消费者注册过的一个回调函数。

推送的过程中，生产者是一个主动的过程，在生产者获取resolved值的时候，生产者可以决定何时把值推送给消费者，而消费者并不知道什么时候可以从生产者这里获取到值。在RxJS中，observable也属于推送体系，并且可以推送一个或多个值。

上面这些术语有些抽象，举个更容易理解什么是Observable

Observable就像是一个function函数

Function
function foo() {
    console.log('Hello')
    return 'world';
}

const x = foo();
console.log(x);

const y = foo();
console.log(y);

Observable
const foo = Observable.create(function (observer) {
    console.log('Hello');
    observer.next('world');
});
// .subscribe()类似于调用函数
foo.subscribe(function (x) {
    console.log(x);
});
foo.subscribe(function (y) {
    console.log(y);
});

// 控制台输出是相同的： 
'Hello'
'world'
'Hello'
'world'

Observable和function的区别是什么?

Observable可以随着时间推移返回（推送）多个值 ，这一点是函数做不到的。

Function
function foo() {
    return 'Hello';
    return 'world'; // 永远不会执行
}

const a = foo();
console.log(a);

//控制台输出
'Hello'

Observable
const foo = Observable.create(function (observer) {
    observer.next('Hello');
    observer.next('world');
});

foo.subscribe(function (x) {
    console.log(x);
});

// 控制台输出
'Hello'
'world'

// 也可以异步推送一些值
const foo = Observable.create(function (observer) {
    observer.next('Hello')；
    setTimeout(() => {
        observer.next('rxjs');
    },0)
    observer.next('world');
});

// 控制台输出
'Hello'
'world'
'rxjs'

冰墩墩占位

1.1 创建Observable

Observable可以使用Observable.create来创建，但通常我们使用创建操作符[1]来创建Observable。

1.2 订阅Observable

订阅Observable像是调用函数，并提供接收数据的回调函数。

observable.subscribe(value => {
 // do something
})

不同观察者通过subscribe调用同一observable数据不共享。

每一次调用，等于重新执行一遍函数。

1.3 执行Observable

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

Next：推送一个值，可以是任意类型；
Error：推送一个错误或者异常；
Complete：推送一个「已完成」的消息，表明不会再发送任何值；

next()方法中的值代表要推送给观察者的实际数据，可以执行多次；

error()和complete()会在Observable执行期间至多执行一次，并且只会执行其中一个；

Observable.create(observer => {
    try {
        observer.next(1);
        observer.next(2);
        observer.complete();
        observer.next(3); // 前面已经通知观察者已经完成，所以这个值不会发送
    } catch (e) {
        observer.error(e); // 捕获到异常发送一个错误
    }
})

1.4 销毁Observable执行

Observable的执行可能会是无限的，通常观察者希望在一个有限的时间里终止Observable执行，以避免浪费计算资源和内存消耗。

类似于清除定时器，var timer = setInterval(() => {},1000); clearInterval(timer);

// 调用subscribe时，观察者会被附加到新创建的Observable执行中，
// 会返回一个对象，即Subscription（订阅）
var subscription = observable.subscribe();
// Subscription表示正在进行中的执行，调用unsubscribe()来取消observable执行；
subscription.unsubscribe();

2. Observer (观察者)

Observer（观察者）是一组回调函数的集合，每一个回调函数对应Observable发送通知的类型：next、error、complete。

const observer = {
    next: () => {}, // 观察者接收到next()消息执行的回调函数
    error: () => {}, // 观察者接收到error()消息执行的回调函数
    complete: () => {}, // 接收到complete()消息执行的回调函数
}
// observer中的观察者可能是部分的，没有提供某个回调，observable还是可以执行的。
// 方法1：将observer观察者传入subscribe
observable.subscribe(observer)
// 方法2：subscribe按顺序（next，error，complete)传入三个回调函数
observable.subscribe((value) => {},(error) => {}, () => {})

3. Subscription (订阅)

Subscription是一个可清理资源的对象，代表Observable的执行。

基本用处就是使用unsubscribe来释放资源或取消Observable的执行。

4. Subject (主体)

引入一个新的概念，Cold Observable / Hot Observable。

Observable对象就是一个数据流，在一个时间范围内推送一系列数据。

在只存在一个observer的情况下很简单，但是对于存在多个observer的场景，会变得复杂。

假设一个场景：

两个observer观察者A和B订阅同一个Observable对象，但他们不是同时订阅，第一个观察者A订阅N秒后，第二个观察者B才订阅这个Observable对象。并且在这N秒期间,Observable已经推送了一些数据，那么第二个观察者B应不应该收到已经被推送给第一个观察者A的那些数据呢？

Selection 1 ：已经推送给观察者A的值就不给B了，B只从订阅那一时间点接收Observable推送的数据就行了。

Selection 2：已经推送给观察者A的值还是要给B，B订阅时从头开始获取Observable推送的数据。

RxJS考虑到这两种不同的场景，让Observable支持这两种不同的需求，Selection 1这样的Observable就是Hot Observable，而Selection 2这样的Observable就是Cold Observable。

RxJS Subject是一种特殊类型的Observable，允许将值多播给多个观察者（每个已订阅的观察者从订阅时间点开始接收当前Observable推送的值，非独立），而普通的Observable是单播的（每个已订阅的观察者是独立执行Observable的）。

对于多个订阅Subject的观察者，subscribe不会重新从头发送值，他只是将观察者注册到观察者列表中，后续有新值发送的时候，将值多播给观察者列表中的所有观察者。

RxJS的四种不同类型Subject

Observable	Subject	BehaviorSubject	AsyncSubject	ReplaySubject
每次从源头开始将值推送给观察者	将值多播给已订阅的该Subject的观察者列表	把最后一个值（当前值）发送给观察者（需要一个初始值）	执行的最后一个值发给观察者	可以把之前错过的值发给观察者

4.1.1 BehaviorSubject

BS有一个“当前值”的概念，它保存了发送给观察者的最后一个值（当前值），当有新的观察者订阅时，会立即接收到“当前值”；

而如果用Subject，在观察者订阅时，之前已发送的值不会再发给观察者包括最近的一个值，后续再有值发送的时候，新注册的观察者才会接收到新的值。

var 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);
// 输出：
observerA: 0 //line3 ：A订阅时立即收到当前值（初始值）0
observerA: 1 //line7 : BS推送新的值1，订阅者A接收到值1
observerA: 2 //line8 : BS推送新的值2，订阅者A接收到值2
observerB: 2 //line 10 : B订阅时立即收到变化后的当前值2
observerA: 3 //line 14：BS推送新的值3，订阅者A和B一起收到值3
observerB: 3

4.1.2 AsyncSubject

AS只有当Observable执行完成时【执行complete()】，才会将执行的最后一个值发送给观察者

var subject = new Rx.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();
// 输出：
// line 17 执行complete()后两个订阅者A和B才会收到最后的一个值(5)
observerA: 5
observerB: 5

4.1.3 ReplaySubject

RS类似BS，它可以发送旧值给新的观察者，还可以记录Observable的执行的一部分，将Observable执行过程中的多个值回放给新的观察者。

var 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);
// 输出
observerA: 1 // line 7: RS推送值1，订阅者A收到值1
observerA: 2 // line 8: RS推送值2，订阅者A收到值2
observerA: 3 // Line 9: RS推送值3，订阅者A收到值3
observerA: 4 // line 10: RS推送值4，订阅者A收到值4
observerB: 2 // line 12: 新的订阅者订阅RS
observerB: 3 // 订阅时按顺序收到了RS缓冲的三个值
observerB: 4
observerA: 5 // line 16：RS推送值5，观察者A和B收到值5
observerB: 5

Tips
//RS除了可以指定缓冲数量，还可以指定时间（单位毫秒）来确定多久之前的值要记录
var subject = new ReplaySubject(3,500) 记录3个值，500ms前。

5. Operators (操作符)

操作符是允许复杂的异步代码以声明式的方式进行轻松组合的基础代码单元。

操作符本质就是一个纯函数，当操作符被调用时，不会改变已经存在的Observable实例，会基于当前Observable创建一个新的Observable。

一个Observable对象代表的是一个数据流，实际场景中，产生Observable对象并不是每次都通过直接调用Observable构造函数来创造数据流对象。于现实中复杂的问题，并不会创造一个数据流之后就直接通过 subscribe接上一个Observer，往往需要对这个数据流做一系列处理，然后才交给Observer。就像一个管道，数据从管道的一段流入，途径管道各个环节，当数据到达Observer的时候，已经被管道操作过，有的数据已经被中途过滤抛弃掉了，有的数据已经被改变了原来的形态，而且最后的数据可能来自多个数据源，最后Observer只需要处理能够走到终点的数据，而这个数据管道就是pipe。

而对于每一个操作符，链接的就是上游(upstream)和下游 (downstream)

Marble diagrams （弹珠图）

为了能够解释流是如何变化的，文字通常不足以能够描述清楚，我们常常使用弹珠图来对流的时序变化（操作符的运行方式）进行描述。

一个弹珠图：

参考资料

https://cn.rx.js.org/manual/overview.html

https://rxjs-cn.github.io/learn-rxjs-operators/

参考资料

[1]创建操作符: https://rxjs-cn.github.io/learn-rxjs-operators/operators/creation/