《深刻浅出RxJS》读书笔记

时间  2019-12-13
标签 深刻浅出RxJS 读书 笔记 繁體版
原文   原文链接

rxjs的引入

// 若是以这种方式导入rxjs，那么整个库都会导入，咱们通常不可能在项目中运用到rxjs的全部功能
const Rx = require('rxjs');

解决这个问题，可使用深链deep link的方式，只导入用的上的功能ajax

import {Observable} from 'rxjs/Observable';

这样能够减小没必要要的依赖，不光能够优化打包文件的大小，还有利于代码的稳定性npm

另外目前最新的一种解决方案就是Tree Shaking, Tree Shaking只对import语句导入产生做用，对require不起做用。由于tree shaking的工做方式是对代码静态分析，import只能出如今代码的第一层，不能出如今if分支中。而require能够出如今if分支中，参数也是能够动态产生的字符串，因此只能动态执行时才知道require函数式如何执行的，这里Tree Shaking就不起做用了。数组

rxjs中Tree Shaking不起做用

实际项目中，若是不会使用不少RxJS的功能，建议仍是避免导入所有RxJS的作法，使用npm导入而后经过打包工具来组合promise

Observer的完结

为了让Observable有机会告诉Observer已经没有更多数据了，须要有另一种通讯机制。在Rxjs中，实现这种通讯机制的就是Observer的complete函数缓存

若是你无法预测你的程序会不会出现异常，那么就须要使用error参数，若是不须要能够直接给个Null做为第二个参数异步

const theObserver = {
    next: item => console.log(item),
    null,
    complete: () => console.log('No More Data')
};

什么时候完结这个Observable对象须要Observable主动调用complete()
在Observable发生error以后，再也不会调用后面的complete().由于在一个Observable对象中，要么是完结状态，要么是出错状态。一旦进入出错那么就终结了。函数

Observable 可观察的对象
Observer 观察者
联系二者的桥梁就是subscribe
在Rxjs中，发布者就是Observable，观察者就是subscribe函数，这样就能够吧观察者和发布者联系起来工具

如何取消订阅

const Observable = Rx.Observable;
const onSubscribe = observer => {
    let number = 1;
    const handle = setInterval(() => {
        observer.next(number++);
    }, 1000);
    return {
        unsubscribe: () => {
            clearInterval(handle);
        }
    };
};
const source$ = new Observable(onSubscribe);
const subscription = source$.subscribe(item => console.log(item));
setTimeout(() => {
    subscription.unsubscribe();
}, 3500);

Observable产生的事件，只有Observer经过subscribe订阅以后才会收到，在unsubscribe以后就再也不收到优化

Hot Observable和Cold Observable

若是一个Observable对象同时有多个Observer订阅，若是A在B以前订阅，那么B该不应订阅到错过的那些数据流。
若是错过就错过了那么这样的Observable成为Hot，可是若是B仍然从头开始订阅这个Observable那么这样的成为Coldui

若是每次订阅的时候， 已经有⼀个热的“⽣产者”准备好了， 那就是Hot Observable， 相反，若是每次订阅都要产⽣⼀个新的⽣产者， 新的⽣产者就像汽车引擎⼀样刚启动时确定是冷的， 因此叫Cold Observable

复杂的问题能够被分解为三个小问题

  • 如何产生事件
  • 如何响应事件
  • 什么样的发布者关联什么样的观察者，也就是什么时候调用subscribe

Observable产生的事件，只有Observer经过subscribe订阅以后才会收到，在unsubscribe以后就不会收到
Observable.create()用来建立一个Observable对象
在RXJS中，和数组的map同样，做为操做符的map也接受一个函数参数，不一样之处是，对于数组的map，是把每一个数组元素都映射为一个新的值，组合成一个新的数组

操做符分类

  • 建立类（creation）
  • 转化类（transformation）
  • 过滤类（filtering）
  • 合并类（combination）
  • 多播类（multicasting）
  • 错误处理类（error Handling）
  • 辅助⼯具类（utility）
  • 条件分⽀类（conditional&boolean）
  • 数学和合计类（mathmatical&aggregate）

静态操做符的导入路径rxjs/add/observable/
实例操做符的导入路径rxjs/add/operator/
在链式调用中，静态操做符只能出如今首位，实例操做符则能够出如今任何位置。

Tree Shaking帮不上Rxjs什么忙，由于Tree Shaking只能作静态代码检查，并非程序运行时去检测一个函数是否真的被调用、只有一个函数在任何代码中都没有引用过，才会认为这个函数不会被引用。可是RxJS任何一个操做符都是挂在Observable类或者Observable.prototype上的， 赋值给Observable或者Observable.prototype上某个属性在Tree Shaking看来就是被引⽤， 因此， 所
有的操做符， 无论真实运⾏时是否被调⽤， 都会被Tree Shaking认为是会⽤到的代码， 也就不会当作死代码删除。

退订资源的基本原则：当再也不须要某个Observable对象获取数据的时候，就要退订这个Observable对象
在对上游的数据处理中，利用try...catch...的组合捕获project调用的可能的错误，若是真的有错误，那就调用下游的error函数

const sub = this.subscribe({
    next: value => {
        try{
            observer.next(project(value))
        }catch(err) {
            observer.error(err);
        }
    },
    error: err => observer.error(err),
    complete: () => observer.complete()
});

关联Observable

  1. 给Observable打补丁

这种方式比较简单，能够直接绑定在prototype上，若是是静态属性直接绑定在类上面

  1. 使用bind绑定特定的Observable对象
// 好比咱们本身建立了一个map方法
function map(project) {
    return new Observable(observer => {
        const sub = this.subscribe({
            next: value => observer.next(project(value)),
            error: err => observer.next(error),
            complete: () => observer.complete()
        });
        return {
            unsubscribe: () => {
                sub.unsubscribe();
            }
        };
    });
}
// 这个时候咱们就能够主动使用bind改变this的指向
const result$ = map.bind(source$)(x => x * 2);
// 或者直接使用call
const result$ = map.call(source$, x => x * 2);
  1. 使用lift

lift是Observable的实例函数，它会返回一个新的Observable对象，经过传递给lift的函数参数能够赋予这个新的Observable对象特殊的功能

function map(project) {
    return this.lift(function(source$) {
        return source$.subscribe({
            next: value => {
                try{
                    this.next(project(value));
                }catch(err) {
                    this.error(err);
                }
            },
            error: err => this.error(error),
            complete: () => this.complete()
        });
    });
}
Observable.prototype.map = map;

create

Observable.create() 其实就是简单的调用了Observable的构造函数

Observable.create = function(subscribe) {
    return new Observable(subscribe);
}

of

range

range(1, 10) 从1开始吐出10个数据
range(1.5, 3) 从1.5开始吐出3个数据，每次加1

generate

generate相似一个for循环，设定一个初始值，每次递增这个值，知道知足某个条件为止
使用generate实现range功能

const range = (start, count) => {
    const max = start + count;
    return Observable.generate(
        start,
        value => value < max,
        value => value + 1,
        value => value
    );
};

全部可以使用for循环完成的操做，均可以使用generate来实现

repeat 重复数据的数据流

const source$ = Observable.of(1,2,3);
const repeated$ = source$.repeat(10);
// 将source$中的数据流重复10遍

empty()

产生一个直接完结的Observable对象

throw()

产生的Observable对象什么都不作，直接抛出错误

never()

产生的Observable对象什么也不作，既不吐出数据，也不产生错误

interval()

接受一个数值类型的参数，表明产生数据的间隔毫秒数

timer()

第一个参数能够是一个数值，表示多少毫秒以后吐出第一个数值0
若是存在第二个参数，那就会产生一个持续吐出数据的Observable对象，第二个参数就是时间间隔

// 2s后。每隔1s产生一个数值，该数值从0开始递增
const source$ = Observable.timer(2000, 1000);

from()

能够将一切转化为Observable

fromPromise()

能够将Promise对象转化为Observable对象，Promise若是成功则调用正常的成功回调，若是失败则调用失败的回调

fromEvent()

将DOM事件转化为Observable对象中的数据

// 将点击事件转化为Observable
const source$ = Observble.fromEvent(document.querySelector('#id'), 'click');

ajax()

用来将ajax的返回转化为Observable对象

repeatWhen()

接受一个函数做为参数，这个函数在上游第一次产生异常是被调用，这个函数应该返回一个Observable对象

const notifier = () => {
    return Observable.interval(1000);
};
const source$ = Observable.of(1,2,3);
const repeat$ = source$.repeatWhen(notifier);

defer()

当defer产生的Observable对象呗订阅的时候，defer的函数参数就会被调用，逾期这个函数返回另一个Observable

const observableFactory = () => Observable.of(1,2,3);
const source$ = Observable.defer(observableFacatory);

合并类操做符

很多合并类操做符都有两种形式，既提供静态操做符，又提供实例操做符。

concat

concat能够将多个Observable的数据内容一次合并

const source1$ = Observable.of(1,2,3);
const source2$ = Observable.of(4,5,6);
const concated$ = source1$.concat(source2$);
// 或者静态操做符
const concated$ = Observable.concat(source1$, source2$);

concat开始从下一个Observable抽取数据是发生在前一个Observable对象完结以后，因此参与到这个concat之中的Observable对象应该都能完结。若是一个Observable对象不完结，那排在后面的Observable对象永远没有上场的机会

// source1$不完结，永远轮不到source2$上场
const source1$ = Observable.interval(1000);
const source2$ = Observable.of(1);
const concated$ = source1$.concat(source2$);

merge

先到先得快速经过
merge一样支持静态和实例形式的操做符

const Observable = Rx.Observable;
const source1$ = Observable.timer(0, 1000).map(x => x + 'A');
const source2$ = Observable.timer(500, 1000).map(x => x + 'B');
const merged$ = Observable.merge(source1$, source2$);
merged$.subscribe(console.log, null, () => console.log('complete'));

merge第一时间会subscribe上游全部的Observable，而后才去先到先得的策略，任何一个Observable只要有数据下来，就会传给下游的Observable对象
image

merge的第一个Observable若是产生的是同步数据流，那会等第一个同步数据流产生完毕以后，再回合并下一个Observable对象，所以merge的主要适用场景仍然是异步数据流。一个比较经常使用的场景就是用于合并DOM事件

merge还有一个可选的参数concurrent，用于指定同时合并的Observable对象的个数

const source1$ = Observable.timer(0, 1000).map(x => x+'A');
const source2$ = Observable.timer(500, 1000).map(x => x+'B');
const source3$ = Observable.timer(1000, 1000).map(x => x+'C');
const merged$ = source1$.merge(source2$, source3$, 2);
merged$.subscribe(console.log, null, () => console.log('complete'));
// 0A 0B 1A 1B 2A 2B...

这里就限定了优先合并2个Observable对象。而第一二个又不会完结，因此source3$没有出头之日。

zip

zip将上游的两个Obserable合并，而且将他们中的数据一一对应。

// 基本用法
const source1$ = Observable.of(1,2,3);
const source2$ = Observable.of(4,5,6);
const zipped$ = Observable.zip(source1$, source2$);
zipped$.subscribe(console.log, null, () => console.log('completed'));
// [1,4] [2,5] [3,6] completed

当使用zip的时候，它会马上订阅上游Observable，而后开始合并数据。对于zip而言上游任何一个Observable完结，zip只要给这个完结的Observable对象吐出全部的数据找到配对的数据，那么zip就会给下游一个complete信号

const source1$ = Observable.interval(1000);
const source2$ = Observable.of('a', 'b', 'c');
// [0, 'a'] [1, 'b'] [2, 'c'] complete

可是这里也会有一个问题，若是某个上游的source1$吐出的数据很快，可是source$2吐出的数据慢，那么zip就不得不先存储source1$的数据

若是使用zip组合超过两个Observable对象，游戏规则依然同样，组合而成的Observable吐出的数据依然是数组

combineLatest

合并最后一个数据，从全部输入Observable对象中那最后一个产生的数据（最新数据），而后把这些数据组合起来传给下游。

const source1$ = Observable.timer(500, 1000);
const source2$ = Observable.timer(1000, 1000);
const result$ = source1$.combineLatest(source2$);

image

image

咱们也能够自由的定制下游数据

const source1$ = Observable.timer(500, 1000);
const source2$ = Observable.timer(1000, 1000);
const project = (a, b) => `${a} and ${b}`;
const result$ = source1$.combineLatest(source2$, project);

多重依赖的问题：

const original$ = Observable.timer(0, 1000);
const source1$ = original$.map(x => x + 'a');
const source2$ = original$.map(x => x + 'b');
const result$ = source1$.combineLatest(source2$);

withLatestFrom

功能相似于combineLatest，可是给下游推送数据只能由一个

const source1$ = Observable.timer(0, 2000).map(x => 1000 * x);
const source2$ = Observable.timer(500, 1000);
const result$ = source1$.withLatestFrom(source2$, (a,b) => a + b);
// 101 203 305 407...

race

第一个吐出数据的Observable对象就是胜者，race产生的Observable就会彻底采用Observable对象的数据，其他的输入Observable对象则会被退订而抛弃。

const source1$ = Observable.timer(0, 2000).map(x => x + 'a');
const source2$ = Observable.timer(500, 2000).map(y => y + 'b');
const winner$ = source1$.race(source2$);
winner$.subscribe(console.log);
// 1a 2a 3a...

startWith

让一个Observable对象在被订阅的时候，老是先吐出指定的若干数据

const origin$ = Observable.timer(0, 1000);
const result$ = origin$.startWith('start');
// start
// 0
// 1

startWith的操做符就是为了知足链式调用的需求

original$.map(x => x * 2).startWith('start').map(x => x + 'ok');

forkJoin

只有当全部的Observable对象都完结，肯定不会有新的数据产生的时候，forkJoin就会把全部输入的Observable对象产生的最后一个数据合并成给下游惟一的数据

const source1$ = Observable.interval(1000).map(x => x + 'a').take(1);
const source2$ = Observable.interval(1000).map(x => x + 'b').take(3);
const concated$ = Observable.forkJoin(source1$, source2$);
concated$.subscribe(console.log);
// ["0a", "2b"]

高阶Observable

所谓高阶Observable，指的就是产生数据依然是Observable的Observable

// 高阶Observable示例
const ho$ = Observable.interval(1000).take(2)
    .map(x => Observable.interval(1500).map(y => x + ':' + y));

concatAll

会对其内部的Observable对象作concat操做

const ho$ = Observable.interval(1000)
    .take(2)
    .map(x => Observable.interval(1500).map(y => x+':'+y).take(2));
const concated$ = ho$.concatAll();
// 0:0 0:1 1:0 1:1

concatAll首先会订阅上游产生的第一个内部的Observable对象，抽取其中的数据，而后只有当第一个Observable完结的时候才回去订阅第二个Observable。这样很容易产生数据积压

mergeAll

和concatAll()功能相似，可是只要上游产生了数据，mergeAll就会当即订阅

switch

switch的含义就是切换，老是切换到最新的内部Observable对象获取数据。每当switch的上游高阶Observable产生一个内部Observable对象，witch都会⽴刻订阅最新的内部Observable对象上， 若是已经订阅了以前的内部Observable对象， 就会退订那个过期的内部Observable对象， 这个“⽤上新的， 舍弃旧的”动做， 就是切换。

const ho$ = Observable.interval(1000)
    .take(2)
    .map(x => Observable.interval(1500).map(y => x+':'+y).take(2));
const result$ = ho$.switch();

exhaust

exhaust在耗尽当前内部Observable数据以前不会切换到下一个内部Observable对象。和switch同样，exhaust产生的Observable对象完结前提是最新的内部Observable对象完结并且上游高阶Observable对象完结

count

统计上游Observable对象吐出全部数据的个数

const source$ = Observable.of(1,2,3).concat(Observable.of(4,5,6));
const count$ = source$.count(); // 6

max, min

取的最小值和最大值

reduce

规约统计

const source$ = Observable.range(1, 100);
const reduced$ = source$.reduce((acc, current) => acc + current, 0);
// 参数基本和js中的一致

find和findIndex

在某些状况下，咱们但愿能够将find和findIndex结合在一块儿，咱们能够这样作

const source$ = Observable.of(3,1,4,1,5,9);
const isEven = x => x % 2 === 0;
const find$ = source$.find(isEven);
const findIndex$ = source$.findIndex(isEven);
const zipped$ =find$.zip(findIndex$);

defaultIfEmpty

defaultIfEmpty()除了检测上游Observable对象是否为空以外，还要接受一个默认值做为参数，若是上游Observable对象是空的，那就把默认值吐出来

const new$ = source$.defaultIfEmpty('this is default');

filter

过滤

first

若是first不接受参数，那么就是获取的上游的第一个数据
若是first接受函数做为参数，那么就会获取上游数据中知足函数条件的第一个数据

last

工做方式与first恰好相反，从上游数据的末尾开始寻找符合条件的元素

takeWhile

接受一个断定函数做为参数

const source$ = Observable.range(1, 100);
const takeWhile$ = source$.takeWhile(
    value => value % 2 === 0
);

takeUtil

takeUtil是一个里程碑式的过滤类操做符，由于takeUtil让咱们能够用Observable对象来控制另外一个Observable对象的数据产生

在RxJS中，建立类操做符是数据流的源头，其他全部操做符最重要的三类就是合并类、过滤类和转化类。

map

map用来改变数据流中的数据，具备一一对应的映射功能

const source$ = Rx.Observable.of(1,2,3);
// 注意这里只能使用普通函数，箭头函数中的this是绑定在执行环境上的，没法获取context中的值
const mapFunc = function(value, index) {
    return `${value} ${this.separator} ${index}`;
}
const context = {separator: ':'};
const result$ = source$.map(mapFunc, context);

result$.subscribe(
    console.log,
    null,
    () => console.log('complete')
);

mapTo

不管上游产生什么数据，传给下游的都是一样的数据

// 将result$中的数据都映射成A
const result$ = source$.mapTo('A');

pluck

pluck就是把上游数据中特定字段的值拔出来

const source$ = Rx.Observable.of(
    {name: 'RxJS', version: 'v4'},
    {name: 'React', version: 'v15'},
    {name: 'React', version: 'v16'},
    {name: 'RxJS', version: 'v5'}
);
const result$ = source$.pluck('name');
result$.subscribe(
    console.log,
    null,
    () => console.log('complete')
);
// RxJS
// React
// React
// RxJS
// complete

上面的代码中，pluck方法将对象中的键对应的值获取出来

获取DOM事件中的值

const click$ = Rx.Observable.fromEvent(document, 'click');
const result$ = click$.pluck('target', 'tagName');
// HTML

将上游数据放在数组中传给下游操做符

bufferTime

用一个参数来指定产生缓冲窗口的间隔

const source$ = Rx.Observable.timer(0, 100);
// 参数400，就会把时间划分为连续的400毫秒长度区块，上游传来的数据不会直接传给下游，而是在该时间区块的开始就新建一个数组对象推送给下游
const result$ = source$.bufferTime(400);

若是上游在短期内产生了大量的数据，那bufferTime就会有很大的内存压力，为了防止出现这种状况，bufferTime还支持第三个可选参数，用于指定每一个时间区间内缓存的最多数据个数

const result$ = source$.bufferTime(400, 200, 2);

bufferCount

根据个数来界定

bufferWhen

接受一个函数做为参数，这个参数名为closingSelector

bufferToggle

buffer

将上游数据放在Observable中传给下游的操做符

windowTime

用一个参数来指定产生缓冲窗口的间隔

windowCount

windowToggle

window

高阶map

全部的高阶map的操做符都有一个函数参数project，可是和普通map不一样，普通map只是把一个数据映射成另一个数据，高阶map的函数参数project把一个数据映射成一个Observable对象

const project = (value, index) => {
    return Observable.interval(100).take(5);
}
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程