Operators 操作符完全解析 - 《RxJS》

Operator 概念
- 实现一个 Operator
- 实例操作符 - 静态操作符
创建操作符
多播操作符
错误处理操作符
工具操作符
条件和布尔操作符
数学和聚合操作符

Operator 概念

采用函数式编程风格的纯函数 (pure function)，使用像 map、filter、concat、flatMap 等这样的操作符来处理集合。也正因为他的纯函数定义，所以我们可以知道调用任意的操作符时都不会改变已存在的Observable实例，而是会在原有的基础上返回一个新的Observable。

尽管 RxJS 的根基是 Observable，但最有用的还是它的操作符。操作符是允许复杂的异步代码以声明式的方式进行轻松组合的基础代码单元。

Operators 操作符完全解析 - 图1

Observable 图

实现一个 Operator

假设我们不使用RxJS提供的过滤操作符，那么让你自己实现又该怎么做呢？

function filter(source, callback) {    return Rx.Observable.create(((observer) => {        source.subscribe(            (v) => callback(v) && observer.next(v),            (err) => observer.error(err),            (complete) => observer.complete(complete)        );    }))}const source = Rx.Observable.interval(1000).take(3);filter(source, (value) => value < 2).subscribe((value) => console.log(value));// 0// 1

这样就实现了一个简单的filter操作符，是不是很简洁，其实主要的做法还是像上面所说，基于传入的Observable，返回一个新的Observable。

代码中首先创建了一个Observable，接着用一个新的观察者订阅传入的源，并调用回调函数判断是否这个值需要继续下发，如果为false，则直接跳过，根据我们传入的源与过滤函数来看，源对象最终会发送三个数 0、1、2，打印结果为 0、1，2 被过滤了。

当然我们也可以将其放置到Rx.Observable.prototype上以便以我们可以采用this的方式获取源：

Rx.Observable.prototype.filter = function (callback) {    return Rx.Observable.create(((observer) => {        this.subscribe(            (v) => callback(v) && observer.next(v),            (err) => observer.error(err),            (complete) => observer.complete(complete)        );    }))}Rx.Observable.interval(1000).take(3).filter((value) => value < 2).subscribe((value) => console.log(value));// 0// 1

这样是不会就更加简洁了，就像我们使用原生数组的filter方法一样。

要说这两种方式的区别，其实也比较好理解，一个是放在prototype中，能够被实例化的对象直接调用，另一个是定义了一个新的函数，可以用来导出给调用者使用（其实也可以直接挂载到Observable的静态属性上）。

看到这里估计会有读者已经猜到笔者接下来说讲解什么了。

实例操作符 - 静态操作符

实例操作符：通常是能被实例化的对象直接调用的操作符。我们一般更多会使用实例操作符多一点，比如filter、map、concat等等。使用实例操作符可以更快乐的使用this，而省去一个参数，还能维持链式调用。
静态操作符：Observable是一个class类，我们可以直接把操作符挂载到他的静态属性上，好处在于无需实例化即可调用，缺点在于就无法再使用this的方式进行目标对象调用了，而是需要把目标对象传入。

如果添加一个实例化属性上面已经有示例了，这里就不做过多赘述了。

将上述的filter例子改造一下，将其挂载到静态属性上：

Rx.Observable.filter = (source, callback) => {    return Rx.Observable.create(((observer) => {        source.subscribe(            (v) => callback(v) && observer.next(v),            (err) => observer.error(err),            (complete) => observer.complete(complete)        );    }))}