JS rxjs与 Observable

前言

Observable 生产的数据，应该提供开发人员足够的自由度，对这些数据进行各种处理，比如 map / transform 等等。这就是 Rxjs Operator 大展身手的地方。

运算符是对 Observable 进行操作并返回 Observable 的函数。 这允许我们链接这些运算符。 链中的每个运算符都会修改由前一个运算符的运算产生的 Observable。

链中的算子不会同时工作，而是按顺序运行，每个算子都在链中前一个算子生成的 Observable 上工作。

订阅是使用 subscribe 运算符完成的。 订阅操作符允许观察者连接 Observable。 观察者要从 Observable 获取数据或错误，首先必须订阅该 Observable。

Rxjs 里的 Operator 可以分为两大类。

1. 可以加入到 pipe 链式操作的 Operator

可管道运算符是可以链接在一起的运算符。 这些是纯函数，它们将可观察对象作为输入并提供可观察对象作为输出。

伪代码如下：

observeable.pipe(
  operator1(),
  operator2(),
  operator3(),
  operator3(),
)

operator1 将接收 observable，对其执行操作并发出 observable。 从 operator1 发出的 observable 被传递给 operator2 作为输入（通过其余的运算符以此类推）。

filter、mergeMap 和 forkJoin 是可管道操作符的一些示例。

2. 创建型 Operator

创建操作符是创建新 Observable 的独立函数。最典型的创建操作符就是 of:

import { of } from 'rxjs';
const observable = of(1, 2, 3);

变量 observable 是一个 Observable 类型的实例，它将发出 1、2 和 3（按顺序）。

create, of 和 from 是创建型操作符的典型例子。

3. 冷热 Observable 的区别

Code Observable 在观察者订阅它之前不会开始发出值。

相反的，Hot Observable 可以随时开始发出值，订阅者可以随时开始观察发出的值。 但是，订阅者可能会错过订阅时间之前发出的任何值。

看个例子：

import { Observable } from 'rxjs';
const observable = Observable.create((observer: any) => {
  observer.next('Hello World!');
});
observable.subscribe((message) => console.log(message)); // Hello World!

这个 Observable 发射的值就是 Hello World. 它会调用订阅者的 next 方法进行发射。

Observable.create 的输入参数是一个订阅者 subscriber. 这是一个函数，函数的输入参数是 Observer，类型为 any. 函数体为执行该输入参数的 next 方法。

Observable 的构造函数很简单，就把 create 传入的函数，设置到 Observable 内部属性 _subscribe 里： 

然后针对 Observable.create 返回的 Observable 实例，调用 subscribe 方法进行订阅。

订阅者的逻辑通过 subscribe 方法传入，是一个函数： 

toSubscriber 方法会创建一个订阅者 subscriber 实例： 

subscriber 的构造函数里，就是简单的为 destination 字段赋值：

这里把 subscribe 函数传入的应用逻辑，赋给 safeSubscriber 的 _next 属性： 

然后执行 _subscribe 方法： 

这个 _subscribe 方法就是 subscribe 函数调用第一步，把 subscribe 传入的应用逻辑赋给 _subscribe 属性的那个函数。

注意，subscribe 函数调用，首先会触发 Observable 的 emit value： 

this._next 触发 destination.next:

这里调用 SafeSubscriber 的 _next 属性指向的方法，即 subscribe 传入的函数实现：

总结一下，Observable.create 和 Observable.subscribe 这套调用, 首先会触发 Observable 实例调用 next 方法，发射调用创建操作符时指定的待发射值。接着会触发应用开发人员传入 subscribe 调用的业务逻辑，消费这个发射的值。

这段代码执行的先后顺序如下：