RxJS速成 (上)

What is RxJS?

RxJS是ReactiveX编程理念的JavaScript版本。ReactiveX是一种针对异步数据流的编程。简单来说，它将一切数据，包括HTTP请求，DOM事件或者普通数据等包装成流的形式，然后用强大丰富的操作符对流进行处理，使你能以同步编程的方式处理异步数据，并组合不同的操作符来轻松优雅的实现你所需要的功能

下面废话不说, 直接切入正题.

准备项目

我使用typescript来介绍rxjs. 因为我主要是在angular项目里面用ts.

全局安装typescript:

npm install -g typescript      

全局安装ts-node:

npm install -g ts-node      

建立一个文件夹learn-rxjs, 进入并执行:

npm init      

安装rxjs:

npm install rxjs --save      

RxJS的主要成员

• Observable: 一系列值的生产者
• Observer: 它是observable值的消费者
• Subscriber: 连接observer和observable
• Operator: 可以在数据流的途中对值进行转换的操作符
• Subject: 既包括Observable也包括Observer

Observable, Observer, Subscriber的角色关系:

工厂生产杂志, 邮递员去送杂志, 就相当于是Observable, 邮递员给你带来了啥? 带来了杂志, 然后(next)杂志, next杂志.....

把杂志带给了谁? 看看这对夫妇, 可能是丈夫来付账单订杂志, 他就是Subscriber. 而这本女性杂志肯定不是丈夫来看(如果他是正经丈夫的话), 而妻子没有直接去订阅杂志, 但是她看这本杂志有用(知道怎么去用它).

所以可以这样理解, 丈夫(Subscriber)把Observable和Observer联系到了一起, 就是Subscriber为Observable提供了一个Observer(丈夫订杂志, 告诉快递员把货给他媳妇就行).

Observable可以在Observer上调用三种方法(快递员跟他妻子可能会有三种情况...好像这么说不太恰当), 当Observable把数据(杂志)传递过来的时候, 这三种情况是:

• next(), 这期杂志送完了, 等待下一期吧
• error(), 送杂志的时候出现问题了, 没送到.
• complete(), 订的杂志都处理完了, 以后不送了.

下面这个图讲的就是从Observable订阅消息, 并且在Observer里面处理它们:

Observable允许:

• 订阅/取消订阅它的数据流
• 发送下一个值给Observer
• 告诉Observer发生了错误以及错误的信息
• 告诉Observer整个流结束了.

Observer可以提供:

• 一个可以处理流(stream)上的next的值的function
• 处理错误的function
• 处理流结束的function

创建Observable

• Observable.from(), 把数组或iterable对象转换成Observable
• Observable.create(), 返回一个可以在Observer上调用方法的Observable.
• Observable.fromEvent(), 把event转换成Observable.
• Observable.fromPromise(), 把Promise转换成Observable.
• Observable.range(), 在指定范围内返回一串数.

Observable.from()

observable_from.ts:

import { Observable } from "rxjs/Observable"; // 这里没有使用Rx对象而是直接使用其下面的Observable对象, 因为Rx里面很多的功能都用不上.
import 'rxjs/add/observable/from'; // 这里我需要使用from 操纵符(operator)

let persons = [
    { name: 'Dave', age: 34, salary: 2000 },
    { name: 'Nick', age: 37, salary: 32000 },
    { name: 'Howie', age: 40, salary: 26000 },
    { name: 'Brian', age: 40, salary: 30000 },
    { name: 'Kevin', age: 47, salary: 24000 },
];

let index = 1;
Observable.from(persons)
    .subscribe(
        person => {
            console.log(index++, person);
        },
        err => console.log(err),
        () => console.log("Streaming is over.")
    );      

subscribe里面有3个function, 这3个function就是Observer.

第一个function是指当前这个person到来的时候需要做什么;

第二个是错误发生的时候做什么;

第三个function就是流都走完的时候做什么.

注意, 是当执行到.subscribe()的时候, Observable才开始推送数据.

运行这个例子需要执行下面的命令:

ts-node observable_from.ts      

Observable.create()

Observable.create是Observable构造函数的一个别名而已. 它只有一个参数就是subscribe function. 

observable_creates.ts:

import { Observable } from "rxjs/Observable";

function getData() {

    let persons = [
        { name: 'Dave', age: 34, salary: 2000 },
        { name: 'Nick', age: 37, salary: 32000 },
        { name: 'Howie', age: 40, salary: 26000 },
        { name: 'Brian', age: 40, salary: 30000 },
        { name: 'Kevin', age: 47, salary: 24000 },
    ];

    return Observable.create(
        observer => { // 这部分就是subscribe function
            persons.forEach(p => observer.next(p));
            observer.complete();
        }
    );
}

getData()
    .subscribe(
        person => console.log(person.name),
        err => console.error(err),
        () => console.log("Streaming is over.")
    );      

create里面的部分是subscribe function. 这部分可以理解为, 每当有人订阅这个Observable的时候, Observable会为他提供一个Observer.

在这里面, observer使用next方法对person进行推送. 当循环结束的时候, 使用complete()方法通知Observable流结束了.

尽管getDate里面create了Observable, 但是整个数据流动并不是在这时就开始的. 在这个地方, 这只不过是个声明而已.

只有当有人去订阅这个Observable的时候, 整个数据流才会流动.

运行该文件:

RxJS Operator(操作符)

Operator是一个function, 它有一个输入, 还有一个输出. 这个function输入是Observable输出也是Observable.

在function里面, 可以做一些转换的动作

下面是几个例子:

observablePersons.filter(p => p.age > 40);      

这个filter function和数组的filter类似, 它接受另一个function(也可以叫predicate)作为参数, 这个function提供了某种标准, 通过这个标准可以判定是否当前的元素可以被送到订阅者那里.

p => p.age > 40      

这个function, 是pure function, 在functional programming(函数式编程)里面, pure function是这样定义的: 如果参数是一样的, 无论外界环境怎么变化, 它的结果肯定是一样的.

pure function不与外界打交道, 不保存到数据库, 不会存储文件, 不依赖于时间....

而这个filter function呢, 在函数式编程里面是一个high order function.

什么是High order function? 

如果一个function的参数可以是另一个function, 或者它可以返回另一个function, 那么它就是High Order function.

Marble 图

首先记住这个网址: 

http://rxmarbles.com/

有时候您可以通过文档查看operator的功能, 有时候文档不是很好理解, 这时你可以参考一下marble 图.

例如 map:

可以看到map接受一个function作为参数, 通过该function可以把每个元素按照function的逻辑进行转换.

例如 filter:

filter就是按条件过滤, 只让合格的元素通过.

例 debounceTime (恢复时间):

如果该元素后10毫秒内, 没有出现其它元素, 那么该元素就可以通过.

例 reduce:

这个也和数组的reduce是一个意思.

例子

import { Observable } from "rxjs/Observable";
import 'rxjs/add/observable/from';
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/reduce';


let persons = [
    { name: 'Dave', age: 34, salary: 2000 },
    { name: 'Nick', age: 37, salary: 32000 },
    { name: 'Howie', age: 40, salary: 26000 },
    { name: 'Brian', age: 40, salary: 30000 },
    { name: 'Kevin', age: 47, salary: 24000 },
];

Observable.from(persons)
    .map(p => p.salary)
    .reduce((total, current) => total+ current, 0)
    .subscribe(
        totalSalary => console.log(`Total salary is ${totalSalary}`),
        err => console.log(err)
    );      

这个例子非常的简单, 典型的map-reduce, 就不讲了.

结果如下:

用现实世界中炼钢生产流程的例子来解释使用Operator来进行Reactive数据流处理的过程:

原料(矿石)整个过程中会经过很多个工作站, 这里每个工作站都可以看作是RxJS的operator, 原料经过这些operator之后, 成品就被生产了出来.

每个工作站(operator)都是可以被组合使用的, 所以可以再加几个工作站也行.

错误处理

Observable是会发生错误的, 如果错误被发送到了Observer的话, 整个流就结束了.

但是做Reactive编程的话, 有一个原则: Reactive的程序应该很有弹性/韧性.

也就是说, 即使错误发生了, 程序也应该继续运行.

但是如果error function在Observer被调用了的话, 那就太晚了, 这样流就停止了.

那么如何在error到达Observer之前对其进行拦截, 以便流可以继续走下去或者说这个流停止了,然后另外一个流替它继续走下去?

错误处理的Operators:

• error() 被Observable在Observer上调用
• catch() 在subscriber里并且在oserver得到它(错误)之前拦截错误,
• retry(n) 立即重试最多n次
• retryWhen(fn) 按照参数function的预定逻辑进行重试

使用catch()进行错误处理:

observable_catch.ts:

import { Observable } from "rxjs/Observable";
import 'rxjs/add/observable/from';
import 'rxjs/add/operator/catch';
import 'rxjs/add/operator/map';

function getFromGoogle(): Observable<any> {
    return Observable.create(function subscribe(observer) {
        observer.next('https://google.com');
        observer.error({
            message: 'Google can\'t be reached.',
            status: 404,
        });
        observer.complete();
    });
}

function getFromBing(): Observable<any> {
    return Observable.create(function subscribe(observer) {
        observer.next('https://global.bing.com');
        observer.complete();
    });
}


function getFromBaidu(): Observable<any> {
    return Observable.create(function subscribe(observer) {
        observer.next('https://www.baidu.com');
        observer.complete();
    });
}

getFromGoogle()
    .catch(err => {
        console.error(`Error: ${err.status}: ${err.message}`);
        if(err.status === 404) {
            return getFromBaidu();
        } else {
            return getFromBing();
        }
    })
    .map(x => `The site is : ${x}`)
    .subscribe(
        x => console.log('Subscriber got: ' + x),
        err => console.error(err),
        () => console.log('The stream is over.')
    );      

在subscribe之前调用catch, catch里可以进行流的替换动作.

运行结果如下:

相当于:

Hot 和 Cold Observable

Cold: Observable可以为每个Subscriber创建新的数据生产者

Hot: 每个Subscriber从订阅的时候开始在同一个数据生产者那里共享其余的数据.

从原理来说是这样的: Cold内部会创建一个新的数据生产者, 而Hot则会一直使用外部的数据生产者.

举个例子:

Cold: 就相当于我在腾讯视频买体育视频会员, 可以从头看里面的足球比赛.

Hot: 就相当于看足球比赛的现场直播, 如果来晚了, 那么前面就看不到了.

Share Operator

share() 操作符允许多个订阅者共享同一个Observable. 也就是把Cold变成Hot.

例子 observable_share.ts:

import { Observable } from "rxjs/Observable";
import 'rxjs/add/observable/interval';
import 'rxjs/add/operator/take';
import 'rxjs/add/operator/share';

const numbers = Observable
    .interval(1000)
    .take(5)
    .share();

function subscribeToNumbers(name) {
    numbers.subscribe(
        x => console.log(`${name}: ${x}`)
    );
}

subscribeToNumbers('Dave');

const anotherSubscription = () => subscribeToNumbers('Nick');

setTimeout(anotherSubscription, 2500);      

这里interval是每隔1秒产生一个数据, take(5)表示取5个数, 也就是1,2,3,4,5.

然后share()就把这个observable从cold变成了hot的.

后边Dave进行了订阅.

2.5秒以后, Nick进行了订阅.

最后结果是:

下面是我的关于ASP.NET Core Web API相关技术的公众号--草根专栏: