从命令式到响应式（四）

摘要：使用的操作符这条从左到右的横线代表经过操作符转换后的输出流。返回值通过判定函数检测的值组成的流。返回值持续发出输入流上的值，直到通知流上发出值为止。

上期介绍过了rxjs中的三大件，Observable，subscription，subject，但是在开发过程我们最常接触到的东西非操作符莫属。比如上期代码中曾出现过的from就是一个操作符。rxjs中的操作符大致上可以分为几类，创建类，组合类，转换类，过滤类，条件类，聚合类，错误处理类，多播类及工具类，其中前四类是数据处理时使用频率非常高的，在本节及下一节中将介绍其中一些使用频率非常高的操作符。rxjs一共提供了120个左右操作符，合理的使用这些操作符会使我们获取愉快的编码体验。

首先需要分清的是操作符是属于实例方法还是静态方法，实例方法的实例当然指的是Observable类的实例，通常情况会在数据转换的过程中使用；而静态方法当然指的是Observable类的静态方法，只能通过Observable类来调用，大部分创建类型的操作符都是静态方法，在rxjs5中区别非常明显，例如：

import "rxjs/Observable";
import "rxjs/add/operator/map";

// 这里的interval 是静态方法，而map就是实例方法。
Observable.interval(1000).map(num => num * num);

在rxjs6中还可能这样写：

import { interval } from "rxjs/observable/interval";
import { map } from "rxjs/operators/map";

// 引入的位置是不一样的。
interval(1000)
    .pipe(
        map(num => num * num)
    );

最直观的描述操作符的行为的方式就是弹珠图，在官网上重要的操作符基本上都给出了相应的弹珠图。从现在开始为了表达的简洁，我们把可观测序列称之为流，弹珠图各部分的含义如下：

// 这条从左到右的横线代表随时间的推移，输入流的执行过程。
// 横线上的值代表从流上发射出的值
// 横线尾部的竖线代表complete通知执行的时间点，表示这条流已经成功的执行完成。
----------4------6--------------a-------8-------------|---->

            multipleByTen // 使用的操作符

// 这条从左到右的横线代表经过操作符转换后的输出流。
// 横线尾部的X代表在这个时间点上流发生了错误，至此之后不应该再有 Next 通知或 Complete 通知从流上发出。
---------40-----60--------------X--------------------------->

前面说过操作符会把我们的数据进行转换，在响应式编程中，我们应该尽量保持数据在流中进行转换，而不是时刻想着去subscribe一条流，取出数据，再转换数据。尤其在angular中，能不手动的subscribe的流，一定要力求不主动订阅，最典型的就是页面上需要显示的数据，我们完全可以交给async管道来进行订阅。OK，啰嗦了一大堆，下面主角登场。

from

静态方法

将数组、类数组对象、promise、部署了遍历器接口的对象或类 Observable 对象转换成Observable，它
几乎可以将任何东西都转换成流，并且将原数据上的值依次推送到流上，字符串被当成由字母组成的数组进行转换。

            from([1,2,3])

    1--------------2--------------3|

示例

from([1,2,3,4,5]).subscribe(v => console.log(v));

function* generatorDoubles(seed) {
    var i = seed;

    while(true) {
        yield i;

        i = i*2
    }
}

const iterator = generatorDoubles(3);
from(iterator).take(5).subscribe(v => console.log(v));

of

静态方法

创建一个流，把传入此函数的参数从左到右依次推送到流上，然后发出结束通知。

            of(1,2,3);

    1-------------2--------------3|

示例

of(10,20,30).subscribe(v => console.log(v));

timer

静态方法

创建一个输出流，在指定的延迟时间到达后开始发射值，在指定的间隔时间到达后发射递增过的值。类似于interval，但是这个操作符允许指定流开始发射值的时间，

            timer(3000, 1000);

    ------0--1--2--3--4--5--------->

第一个参数代表等待时间，第二个参数代表时间间隔，这些值是一些数字常量。等待时间可以是一个毫秒数，也可以是一个日期对象。如果没有指定时间周期，输出流上只会发射0，反之，它会发出一个无限的数列。

示例

Rx.Observable.timer(3000, 1000)
    .subscribe(v => console.log(v));

Rx.Observable.timer(5000)
    .subscribe(v => console.log(v));

filter

实例方法

创建一个流，它的值是使用判定函数对输入流发出的值进行过滤后的值。

    --0--1--2--3--4--5----|-->

            filter(v => v % 2 === 0);

    --0-----2-----4-------|>

和数组的filter方法行为一样，从输入流上获取值，使用判定函数对值进行过滤，只有符合过滤条件的值才会在输出流上发出。

返回值 Observable 通过判定函数检测的值组成的流。

示例

from([1,2,3,4,5,6])
    .filter(num => num %2 === 0)
    .subscribe(v => console.log(v));

first

实例方法

发送输入流上的第一个值，或者第一个符合某些条件的值。

    ---------a-------b------c---------d-->

            first

    ---------a|

在不传入任何参数时，这个操作符仅发出输入流上的第一个值，然后立即发出结束通知。如果传入一个判定函数，则发出第一个通过判定函数检测的值。它还可以接受一个结果控制函数来转化输出的值，或一个在输入流没有发出符合条件的值情况下使用的默认值。如果没有提供默认值，并且在输入流上也没有找到符合条件的值时，输出流将会抛出错误。

返回值 Observable 第一个符合条件的值。

异常 EmptyError 在结束通知发出前如果没有发出过有效值，将会发送一个错误通知给观察者。

示例

from([2,3,4])
    .first()
    .subscribe(v => console.log(v));

from([2,3,4])
    .first(num => num === 5)
    .subscribe(v => console.log(v)); // EmptyError;

skip

实例方法

返回一个跳过指定数量的值的流。

---a---b---c---d---e---|->

            skip(3);

---------------d---e---|>

返回值 Observable 跳过了一定数量值的流。

示例

from([1,2,3,4,5,6])
    .skip(3)
    .subscribe(v => console.log(v));

take

实例方法

从第一个值开始发出指定数量的值，然后发出结束通知。

    ---a------b------c------d-----e---|-->

            take(3);

    ---a------b------c|

输出流仅仅发出了输入流上从第一个值开始的n个值。如果输入流上值的个数小于n，那么所有的值都会被发出。值发射完成后，不管输入流有没有发出结束通知，输出流都会立即发出结束通知。

返回值 Observable 发出输入流上从第一个值开始的n个值，或者输入流发出值的个数小于n时发出所有的值的流。

异常 ArgumentOutOfRangeError 在给此操作符传入负数时给观察者发出的错误。

示例

interval(1000)
    .take(5)
    .subscribe(v => console.log(v));

takeUntil

实例方法

在通知流发出通知之前，持续发射输入流上的值。在通知流发出值之前，输出流完全就是输入流的镜像。此操作符会一直监视传入的通知流，如果通知流发出了值或结束通知，输出流就会停止发射输入流上的值，并发出完成通知。

返回值 Observable 持续发出输入流上的值，直到通知流上发出值为止。

示例

Rx.Observable.interval(1000)
    .takeUntil(Rx.Observable.fromEvent(document, "click"))
    .subscribe(v => console.log(v));

学习操作符时，我们还要关注的一点是，这个操作符是否会发出结束通知，一方面订阅发出结束通知的流时，在库的底层会帮助我们释放资源可以省去手动取消订阅，比如 angular 中 http 服务上的方法，另一方面结束通知可能会影响接下来你使用的操作符，典型的如reduce 和 scan，在一个不发出结束通知的流上使用reduce时你将永远不会得到结果。