学习 RXJS 系列（一）——从几个设计模式开始聊起

用户1097444

发布于 2022-06-29 15:03:00

1.6K0

发布于 2022-06-29 15:03:00

一、RXJS 是什么

RXJS 是 Reactive Extensions for JavaScript 的缩写，起源于 Reactive Extensions，是一个基于可观测数据流 Stream 结合观察者模式和迭代器模式的一种异步编程的应用库。RxJS 是 Reactive Extensions 在 JavaScript 上的实现。

二、前置知识

响应式编程

响应式编程（Reactive Programming）是一种基于事件的模型，它是一种面向数据流和变化传播的编程范式。这意味着可以在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值通过数据流进行传播。例如，对于 a=b+c 这个表达式的处理，在命令式编程中，会先计算 b+c 的结果，再把此结果赋值给变量 a，因此 b，c 两值的变化不会对变量 a 产生影响。但在响应式编程中，变量 a 的值会随时跟随 b，c 的变化而变化。

响应式编程的思路大概如下：你可以用包括 Click 和 Hover 事件在内的任何东西创建 Data stream。任何东西都可以是一个 Stream：变量、用户输入、属性、Cache、数据结构等等。

流

概括来说，流的本质是一个按时间顺序排列的进行中事件的序列集合。我们可以对一个或多个流进行过滤、转换等操作。需要注意的是，流是不可改变的，所以对流进行操作后会在原来的基础上返回一个新的流。

观察者模式

观察者模式（有时又被称为模型（Model）- 视图（View）模式、源 - 收听者 (Listener) 模式或从属者模式）是软件设计模式的一种。在此种模式中，一个目标物件管理所有相依于它的观察者物件，并且在它本身的状态改变时主动发出通知。这通常透过呼叫各观察者所提供的方法来实现。此种模式通常被用来实现事件处理系统。

观察者模式（Observer）完美的将观察者和被观察的对象分离开。举个例子，用户界面可以作为一个观察者，业务数据是被观察者，用户界面观察业务数据的变化，发现数据变化后，就显示在界面上。看到上面这个描述的场景是不是觉得似曾相识？Vue 的工作原理不就是这样的吗，将数据与视图双向绑定，通过响应式编程的思想动态更新订阅的观察者列表。

迭代器模式

迭代器模式（Iterator Pattern）是一种非常常用的设计模式。这种模式用于顺序访问集合对象的元素，不需要知道集合对象的底层表示。迭代器模式属于行为型模式。

JavaScript 中原有表示 “集合” 的数据结构主要是 “数组 (Array)” 和 “对象 (Object)”，ES6 又新增了 Map 和 Set，共四种数据集合，浏览器端还有 NodeList 类数组结构。ES6 中也有 Iterator 迭代器的介绍，为 “集合” 型数据寻求统一的遍历接口正是 ES6 的 Iterator 诞生的背景。

三、基本概念介绍

Observable

Observable 表示一个可调用的未来值或事件的集合，他能被多个 observer 订阅，每个订阅关系相互独立、互不影响。这里可以举个简单的例子，假如你订阅了报纸，只要报纸每次有新的内容出来就会送到（更新）你手上，这个场景中报纸就是 Observable，而你就是一个观察者（observer）。

我们看看在 RXJS 中怎么创建一个 Observable：

const Rx = require('rxjs/Rx');

const newObservable = Rx.Observable.create(observer => {
  observer.next('message1');
});

这里通过调用 Observable.create 创建了一个 Observable，这个方法接受一个函数作为参数，这个函数叫做 producer 函数，用来生成 Observable 的值。这个函数的入参是 observer，在函数内部通过调用 observer.next() 便可生成有一系列值的一个 Observable。

Observer

Observer 是一个回调函数的集合，也就是一个包含几个回调函数的对象。它知道如何去监听由 Observable 提供的值。Observer 在信号流中是一个观察者（哨兵）的角色，它负责观察任务执行的状态并向流中发射信号。

我们简单描述下一个 Observer 长什么样子：

const observer = {
    next: function(value) {
        console.log(value);
    },
    error: function(error) {
        console.log(error);
    },
    complete: function() {
        console.log('complete');
    }
}

RXJS 中 Observer 的回调函数是可选的，我们定义 Observer 时可以不定义 next、error 或者 complete，这并不会对 Observer 的执行造成影响。

我们来看下如何定义一个 Observer：

const newObservable = Rx.Observable.create((observer) => {
    observer.next('message1');
})

newObservable.subscribe((text) =>console.log(text));

这里通过 subscribe 方法让一个 observer 订阅一个 Observable。你可能对 subscribe 的参数有些疑惑，这里我们可以看看 subscribe 的函数定义，了解是如何与上面我们提到的 next、error 和 complete 关联起来的：

subscribe(next?: (value: T) =>void, error?: (error: any) =>void, complete?: () =>void): Subscription;

从入参来看，从左至右依次是 next、error，complete，并且是可选的，我们可以自己选择性的传入相关回调，因为他们都是可选的。

Subscription

Subscription 表示 Observable 的执行，我们可以调用该对象的 unsubscribe 方法清理掉 Observable 的执行，这个方法不需要任何参数，只是用来清理由 Subscription 占用的资源。

const myObservable = Rx.Observable.create(observer => {
  observer.next('foo');
  setTimeout(() => observer.next('bar'), 1000);
});
const subscription = myObservable.subscribe(x =>console.log(x));

subscription.unsubscribe();

我们可以看到，Observable 的执行需要调用 subscribe 方法来触发，如果在 Observable 执行的时候我们调用了 unsubscribe 方法，就会取消正在进行中的 Observable 的执行。

Subject

Subject 对象可以当成是一个中间代理，它位于 Observable 和 Observer 中间，相对于 Observable 来说它是一个 Observer，接收 Observable 发出的数据；相对于 Observer 它又是一个 Observable，对订阅了它的 observer 发送数据。

需要注意的是，Subject 会对订阅了它的 observers 进行多播，这里就涉及到一个单播与多播的概念了，我们分析一下这两个概念：

单播：单播的意思是，每个普通的 Observables 实例都只能被一个观察者订阅，当它被其他观察者订阅的时候会产生一个新的实例。也就是普通 Observables 被不同的观察者订阅的时候，会有多个实例，不管观察者是从何时开始订阅，每个实例都是从头开始把值发给对应的观察者。

多播：前面说到，每个普通的 Observables 实例都只能被一个观察者订阅，但是如果通过 Subject 来代理 Observable 实例的话就能够被多个 observer 所订阅，且无论有没有 observer 订阅，都会发送数据。也就是说无论 observer 什么时候订阅都只会接收到实时的数据。