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进行了订阅.
最后结果是:
- 比较一下以“反射”和“表达式”执行方法的性能差异
- 人工智能芯片是什么?有什么用?
- 柯洁食言“复出”再战AI:明年4月见分晓
- 深入探讨ASP.NET MVC的筛选器
- pytorch自然语言处理之Pooling层的句子分类
- su命令cannot set groups: Operation not permitted的解决方法
- 利用腾讯云 COS 云对象存储定时远程备份网站
- 腾讯云技术公开课:零基础入门高可用云端架构设计
- 包学会之浅入浅出Vue.js:开学篇
- 包学会之浅入浅出Vue.js:升学篇
- 一个只有99行代码的JS流程框架 (一)
- 【腾讯云的1001种玩法】试用腾讯云 Windows Server 2012 R2 镜像的几点经验分享
- 一个只有99行代码的JS流程框架(二)
- 看书的时候如何调试书中简单的C+代码?
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- 「最强」Lettuce 已支持 Redis6 客户端缓存
- Go中校验一个字符串是否是有效的JSON字符串
- 在 Vue 中,如何从插槽中发出数据
- Azure DevOps+Docker+Asp.NET Core 实现CI/CD(二.创建CI持续集成管道)
- 这样的奇技淫巧,劝你不用也罢
- 我服务又双叒叕奔溃了,含泪干货分享
- Azure DevOps+Docker+Asp.NET Core 实现CI/CD(一 .简介与创建自己的代理池)
- Android数据库高手秘籍(十二),LitePal的索引功能
- 使用pm2管理go应用进程
- 用Docker搭建Redis主从复制的集群
- IDEA 非常重要的一些设置项 → 一连串的问题差点让我重新用回 Eclipse !
- ArrayList源码分析(基于jdk1.8)(二):subList陷阱补充
- Windows10中安装Docker
- Windows下Docker安装ClickHouse
- ArrayList源码分析(基于jdk1.8)(三):Arrays.asList方法带来的问题