RxJS——创建数据流

介绍：

在RxJS的世界中，一切都以数据流为中心，在代码中，数据流以Observable类的实例对象形式存在，创建Observable对象就是数据流处理的开始。

所谓创建类操作符，就是在不依赖于其他Observable对象，可以凭空或者根据其他数据源能够创造出一个Observable对象的方法。创建类操作符并不是不需要任何输入，大多数创建型的操作符都接受输入参数，有的还需要其他的数据源，比如浏览器的DOM结构或者WebSocket。重要的是，创建类操作符往往不会从其他Observable对象获取数据源，在数据管道中，创建类操作符是RxJS中数据流的源头，只有获得数据流之后，才可以发挥RxJS其他操作符的强大功能。

对于创建数据流应该明确区分同步数据流和异步数据流的创建。对于同步数据流，关心的只是产生什么样的数据，已经产生数据的顺序关系，数据之间没有时间间隔，所以不需要考虑异步的情况。对于异步数据流，除了要考虑产生什么样的数据，还要考虑产生数据之间的间隔，也就是产生数据的节奏。

因创建类操作的特性，创建类操作符大部分都是静态操作符。

在日常工作中，我们应该尽量使用创建类操作符，避免直接利用Observable的构造函数来创造Observable对象，RxJS提供的创建类操作符覆盖了几乎所有的数据流创建模式，没有必要重复造轮子。

下面的代码均是以rxjs6版本写的。

同步数据流：

1. create：不需要导入任何其他模板就可以直接使用。

import { Observable } from 'rxjs';
Observable.create = function(subscribe){ return new Observable(subscribe)}

create这个操作符并没有给我们带来任何神奇的功能，使用它和直接使用Observable构造函数没什么区别，所以，绝大部分情况用不上它。

2. of: 列举数据

import { of } from 'rxjs';
const source$ = of(1,2,3)
source$.subscribe(console.log,null,() => console.log('complete')
)

of产生的是Cold Observable，对于每一个Observer都会重复吐出同样的一组数据，所以可以反复使用。

在现实项目中，适合使用of的场景是已知不多的几个数据，需要把这些数据用Observable对象来封装，而且也不需要这些数据的处理要有时间间隔。

3. range: 指定范围

import { range } from 'rxjs';
const source$ = range(1,100);

range第一个参数是数字序列开始的数字，第二个参数是数字序列的长度。从初始化开始，每次递增1，初始化数字可以是整数，也可以是小数。

和of一样，range以同步的方式吐出数据。

三个极简的操作符：empty、never 和 throw

empty、never 和 throw是比较有意思的三个操作符，它们产生的Observable对象十分简单，简单得看起来似乎没有什么价值，不过，当需要默认立刻完结、立刻出错或者永不完结的数据流对象时，这三个操作符可以直接使用。

4. empty：产生一个直接完结的Observable对象，没有参数，不产生任何数据，直接完结。

import { EMPTY} from 'rxjs';
const source$ = EMPTY

5. never: 创建一个不向观察者发出任何项的 Observable。产生的Observable对象什么动作都没有，既不吐数据，也不完结，也不产生错误。

import { NEVER } from 'rxjs';
function info() {console.log('Will not be called');
}
// 发出7, 然后不发出任何值(也不发出完成通知)
const result = NEVER().startWith(7);
result.subscribe(x => console.log(x), info, info);

6. throw: 创建一个不发送数据给观察者并且立马发出错误通知的 Observable。产生的Observable对象什么都不做，直接出错，抛出的错误就是throw的参数。

import { throw } from 'rxjs';
// 先发出数字7，然后发出错误通知
const result = throw(new Error('oops!')).startWith(7);
result.subscribe(x => console.log(x), e => console.error(e));

这三个操作符通常用在Observable对象的最后一步。

异步数据流：

1. interval和timer：定时产生数据

在JavaScript中要模拟异步的处理，通常就是用JavaScript自带的两个函数setInterval和setTimeout,通过指定时间，让一些指令在一段时间之后再执行。可以说，在RxJS中，interval和timer这两个操作符的地位就等同于原生JavaScript的setInterval和setTimeout。地位等同，但功能并不完全一样。

import { interval } from 'rxjs';
const source1$ = interval(1000)

上面该代码在1秒钟的时刻吐出数据0,在2秒钟的时刻吐出数据1，在3秒钟的时刻吐出数据2……。该数据流不会完结，因为interval不会主动调用下游的complete，要想停止这个数据流，必须调用退订的操作。

interval产生的异步数据流总是从0开始递增，如需要从1开始递增，代码如下：

import { interval } from 'rxjs';
import { map } from 'rxjs/operators'
const source$ = Observable.interval(1000)
const result$ = source$.map(x=> x + 1)

要解决复杂的问题，应该用多个操作符的组合，而不是让一个操作符的功能无限膨胀。

由上面的描述可知，interval就是RxJS世界中的setInterval，区别只是setInterval定时调用一个函数，而interval返回的是Observable对象定时产生的一个数据。

timer是RxJS世界中的setTimeout。timer的第一个参数可以是一个数值，也可以是一个Date类型的对象。

第一个参数是数值，代表毫秒数，产生的Observable对象在指定毫秒之后会吐出一个数据0，然后立刻完结,代码如下：

import { timer } from 'rxjs';
const source$ = timer(1000)

通过传递一个Date对象给timer来实现，代码如下：

import { timer } from 'rxjs';
const now = new Date()
const later = new Date(now.getTime() + 1000)
const source$ = timer(later)

使用时间参数还是Date对象作为参数，应该根据具体情况而定，如果明确延时产生数据的时间间隔，那就应该用数值作为参数，如果明确一个时间点点，那就用Date对象作为参数是最佳选择。

timer还支持第二个参数，如果使用第二个参数，那就会产生一个持续吐出数据的Observable对象，类似interval的数据流。第二个参数指定的是各数据之间的时间间隔，从被订阅到产生第一个数据0的时间间隔，依然由第一个参数决定。

import { timer } from 'rxjs';
const source$ = timer(2000,1000)

上面的该代码在source$被订阅之后，在2秒钟的时刻吐出0，然后在3秒钟的时刻吐出1，在4秒钟的时刻吐出2……，永不完结的流。

timer支持产生持续的数据序列，这和setTimeout就不一样，setTimeout只能支持一次异步操作，从这个角度看，timer的功能就是setTimeout的超集。

如果timer的第一个参数和第二个参数一样，那就和interval的功能完全一样了，代码如下：

import { interval, timer } from 'rxjs';
const source1$ = interval(1000)
const source2$ = timer(1000, 1000)

2. from：可把一切转化为Observable

from可能是创建类操作符中包容性最强的一个，因为它接受的参数只要“像”Observable就行，然后根据参数中的数据产生一个真正的Observable对象。

“像”Observable的对象很多，一个数组就像Observable，一个不是数组但是“像”数组的对象也算，一个字符串也很像Observable，一个JavaScript中的generator也很像Observable，一个Promise对象也很像，所以from可以把任何对象都转化为Observable对象。

import { from, of } from 'rxjs';
const source$ = from([1,2,3])const another$ = of(1,2,3)
const source1$ = from(another$)

在上面的代码中，source$的数据流与another$完全一样，既然我们有了another$这个Observable，为什么还要转化为source$呢。

from还可以接受Promise对象，行为和fromPromise完全一样。

3. fromPromise：异步处理的交接

如果from的参数是Promise对象，那么这个Promise成功结束，from产生的Observable对象就会吐出Promise成功的结果，并且立刻结束。当Promise对象以失败而告终，from产生的Observable对象也会立刻产生失败事件。

import { from } from 'rxjs';
const promiseResolve = Promise.resolve('good')
const resolveSource$ = from(promiseResolve)
resolveSource$.subscribe(console.log,error => console.log('catch',error),() => console.log('complete')
)
// 运行结果
good
completeconst promiseReject = Promise.reject('bad')
const rejectSource$ = from(promiseReject)
rejectSource$.subscribe(console.log,error => console.log('catch',error),() => console.log('complete')
)
// 运行结果
catch bad

4. fromEvent: 获取用户在网页中的操作事件，把DOM中的事件转化为Observable对象中的数据，产生的是Hot Observable。

fromEvent的第一个参数是一个事件源，在浏览器中，最常见的事件源就是特定的DOM元素，第二个参数是事件的名称，对应DOM事件就是click，mousemove，resize, scroll等。

在网页应用中的用法，代码如下：

<div><button id="clickMe">Click Me</button><div id="text">0</div>
</div>// RxJS 实现
import { fromEvent } from 'rxjs';
let clickCount = 0;
const event$ = fromEvent(document.querySelector('click'))
event$.subscrive(() => {document.querySelector('#text').innerText = ++clickCount;    }
)

另一种应用，如给网页背景增加水印，获取网页的大小，代码如下：

import { fromEvent } from 'rxjs';
const resize$ = fromEvent(window, 'resize')
const scroll$ = fromEvent(window, 'scroll')
// 通过merge操作符继续逻辑处理
merge(resize$, scroll$).pipe();

fromEvent是DOM和RxJS世界的桥梁，产生Observable对象之后，就可以完全按照RxJS的规则来处理数据。除了可以从DOM中获取数据，还可以从Node.js的events中获取数据，代码如下：

import { fromEvent } from 'rxjs';
import EventEmitter from 'events';
const emitter = new EventEmitter()
const source$ = fromEvent(emitter, 'msg')
source$.subscribe(console.log,error => console.log('catch',error),() => console.log('complete')
)
emitter.emit('msg', 1)
emitter.emit('msg', 2)
emitter.emit('msgs', 'bad')
emitter.emit('msg', 3)
// 运行结果
1
2
3

在上面的代码中，emitter发送的事件名称为msg，事件对象是数字，可以认为msg就相当于DOM事件中的click事件名称，事件对象就是具体某个click事件。可以看到，emitter还发出了一个名为msgs的事件，这个事件并没有计入source$，因为fromEvent的第二个参数明确指定只接受msg类型的事件。

参考数据：《深入浅出RxJS》