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进行了订阅.

最后结果是:

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Python中文社区

网易云音乐歌曲评论爬虫(附源码)

用过网易云音乐听歌的朋友都知道,网易云音乐每首歌曲后面都有很多评论,热门歌曲的评论更是接近百万或者是超过百万条.现在我就来分享一下如何爬取网易云音乐歌曲的全部评...

1861
来自专栏小樱的经验随笔

CTF---密码学入门第二题 我喜欢培根

我喜欢培根分值:20 来源: Ph0enix 难度:中 参与人数:3449人 Get Flag:1410人 答题人数:1653人 解题通过率:85% key...

3304
来自专栏一个会写诗的程序员的博客

《Kotlin 反应式编程》使用 RxKotlin 实现一个极简的 http DSL ( Reactive Programming Using Rx Kotlin )《Kotlin 反应式编程》使用

我们现在已经基本知道 Kotlin 中 DSL 的样子了。但是这些 DSL 都是怎样实现的呢?本节我们就通过实现一个极简的http DSL来学习创建 DSL 背...

2032
来自专栏小工匠技术圈

【小工匠聊密码学】--消息摘要--SHA3算法

1843
来自专栏深度学习之tensorflow实战篇

计算机常用算法对照表整理

常用对照: NLP CRF算法: 中文名称条件随机场算法,外文名称conditional random field algorithm,是一种数学算法,是2...

4565
来自专栏大数据文摘

如何利用维基百科的数据可视化当代音乐史

1907
来自专栏后端技术探索

业务id转密文短链的一种实现思路

买家通过电商app下单后,会受到一条短信,短信内容中包括改订单详情页面的h5地址连接,因为是出现在短信中,所以对连接有要求:1.尽量短;2.安全性考虑,订单在数...

953
来自专栏懂啵的蟒络空间

哈希现金(Hashcash)与“工作量证明”

“哈希现金(Hashcash)是一种用于防止垃圾电子邮件和拒绝服务攻击的工作量证明系统,最近以其在比特币(以及其他加密货币)挖矿算法中的应用而闻名,由Adam ...

50510
来自专栏java学习

面试题64(有1千万条有重复的短信,以文本文件的形式保存,一行一条,也有重复。请用5 分钟时间找出重复出现最多的前10 条短信)

1·有1千万条有重复的短信,以文本文件的形式保存,一行一条,也有重复。请用5 分钟时间找出重复出现最多的前10 条短信。? 正确解析如下... 解析: 对于本题...

2988
来自专栏数据小魔方

教你如何优雅的用R语言调用有道翻译

最近刚发现了个有趣的包,一个R语言发烧友开发了R语言与有道在线翻译的接口,可能这位大神也是一个受够了每天打开网页狂敲键盘查词的罪,索性自己动手,从此丰衣足食。 ...

2273

扫码关注云+社区