rxjs的世界

时间 2019-11-20

标签 rxjs 世界繁體版

原文原文链接

rxjs学习了几个月了，看了大量的东西，在理解Observable的本文借鉴的是渔夫的故事，原文，知识的主线以《深刻浅出rxjs》为主，动图借鉴了rxjs中文社区翻译的文章和国外的一个动图网站react

正文：git

在思惟的维度上加入时间考量es6

1、函数响应式编程

Rxjs使用了一种不一样于传统的编程模式----函数响应式编程github

1.1 函数化编程

函数化编程对函数的使用有一些特殊的要求ajax

声明式
纯函数
数据不可变性保持原有数据不变，让新的数据发生变化

为何最近函数式编程崛起数据库

从硬件发展角度，函数式编程的性能问题已经不是问题
语言演进，在一个面向对象的语言中添加了函数化编程的概
函数化编程可让数据流动更明显

1.2 响应式编程

对面数据流，能够天然的处理编程

响应式编程里，最有名的框架Reaactive Extension(ReactiveX 简称 Rx) 响应式的扩展json

Rx是一套经过可监听流来作异步编程的API
最初由微软提出，有各类语言的实现，等于为那些语言新增了一些功能扩展
Rx诞生主要是为了解决异步的问题api

1.3 函数响应式编程的优点

数据流抽象了不少现实问题
擅长处理异步操做
把复杂问题分解成简单问题的组合

2、基本概念

observable  可观察量；感受到的事物   [əb'zɜːvəbl]
observer    n. 观察者 [əb'zɜːvə]
subsecribe    vi. 订阅   [səb'skraɪb]
subscription   n. 捐献；订阅；订金；签署    [səb'skrɪpʃ(ə)n]

Observable 就是“能够被观察的对象”既“可被观察者”
Obserer 就是“观察者”
subscribe 就是二者之间的桥梁数组

2.1 河流的故事

(1）Rx.Observable 是一条河流。
(2）source 做为一条在河流中捕鱼船上的竹筒。鱼（data）能够一个一个的钻到竹筒中（source)

var source = Rx.Observable.create(subscriber)

(3) subscriber 是位捕鱼的渔人，是位好心人,主要任务是把捕获的鱼（data)扔向岸边的饥民

var subscriber = function(observer) {
    var fishes = fetch('http://www.oa.com/api'); // 捕获到鱼
    observer.next(fishes.fish1); // 把捕获的第一条鱼扔向岸边的饥民
    observer.next(fishes.fish2); // 把捕获的第二条鱼扔向岸边的饥民
}

(4)observer 做为岸边上饥民。由于来自天南地北，方言不一样，因此描述本身在获取到鱼后的吃法表述时语法不一样，但其实实质都是同样的，有鱼了（value=> {})怎么办,没鱼了（error => {})怎么办，当天的鱼扔完了（complete => {}）怎么办。

方式一：

observer = (
    value => { console.log(value); },
    error => { console.log('Error: ', error); },
    () => { console.log('complete') }
)
source.subscribe(observer)

方式二：

observer = function(value) {
    console.log(value);
}
source.subscribe(observer); 
// 这根捕鱼的竹筒不少饥民都翘首以待（subscribe),因此竹筒（source)会被新来的饥民订阅（subscribe).
固然，饥民不订阅天然渔人就不会把竹筒(source)中捕获的鱼扔给他。

方式三：

observer = {
    next: function(value) {
        console.log(value);
    },
    error: function(error) {
        console.log('Error: ', error)
    },
    complete: function() {
        console.log('complete')
    }
}
source.subscribe(observer);

//subscribe 河流source知道河流的两边有哪些百姓须要救济，
因此会帮助他subscribe渔人扔出的鱼，这样他就会收到鱼了
source.subscribe(observer);

(5)subscription 为哪一个饥民订阅了哪一个竹筒的清单。能够从清单上划去，那么这个饥民就再不会受到渔人扔出的鱼了

subscription = source.subscribe(observer1);
subscription.unsubscribe(); // 从清单上划去饥民observer1的订阅信息，由于observer1已经不是饥民了，不须要救济了。

五个角色连接起来就是rxjs的实现过程

var 渔人 = function (饥民) {
    var fishes = fetch('server/api'); // 捕获到必定数量的鱼
    饥民.next(fishes.fish1); // 接下来把鱼1扔给饥民
    饥民.next(fishes.fish1); // 接下来把鱼1扔给饥民
} 
var 饥民1 = { // 饥民要想好不一样种状况下的应对方法，不能在没有捕到鱼的时候就饿死。
    next:function (fish) {
        // 有鱼扔过来了，把fish煮了吃掉。
    }，
    error: function(error) {
       // 捕获的鱼有毒，不能吃，因此要想其余办法填饱肚子，能够选择吃野菜什么的，
    }，
    complete: function() {
        // 当天的鱼扔完了，那么能够回家了
    }
}
var 竹筒 = 河流.create(渔人)； // 河流中来了一名渔人，那么他必定会在河流中放下捕鱼的竹筒。
清单 = 竹筒.subscribe(饥民1) // 竹筒被饥民1关注后，就能够收到渔人扔出的鱼了。
setTimeout(() => {
        清单.unsubscribe();  // 一年后，饥民摆脱困境，再也不须要救济，就退订这个竹筒了。把机会让给别人。
}, 1年);

对应到真正的rxjs语法，咱们再来一遍

var subscriber = function(observer) { // 建立了一位渔人
    observer.next('fish1');
    observer.next('fish2');
    observer.complete();
}
var observer1 = { // 来了一位饥民1
    next: function(value) {
        console.log(`我接到鱼${value}啦，不会挨饿咯`);
    },
    error: function(error) {
        console.log(`哎，捕到的鱼由于${error}缘由不能吃`)
    },
    complete: function() {
        console.log('今天的鱼发完了')
    }
}
var source = Rx.Observable.create(subscriber); // 河流中来了一名渔人，他在河流中放下捕鱼的竹筒。
subscription = source.subscribe(observer1); // 竹筒被饥民1关注后，饥民1能够收到渔人扔出的鱼了。
setTimeout(()=> {
    subscription.unsubscribe(); // 3秒后饥民退订了竹筒，给其余饥民机会。
}, 3000);
打印出的结果以下：
// "我接到鱼fish1唠"
// "我接到鱼fish2唠"
// "今天的鱼发完了"

下面是对捕鱼的三个阶段所碰到问题的解决方案

（1）竹筒中如何才能产生鱼
（2）竹筒中有鱼了，怎么向外取
（3）取出来后，鱼被扔向岸边的过程当中发生了什么。

因此操做符的使用也是有前后顺序的。

2.2 术语解读

观察者模式

如何生成事件，这是发布者的责任，在Rxjs中是Observalbe对象的工做
如何响应事件，这是观察者的责任，在Rxjs中由subscribe的参数决定
什么样的发布者关联什么样的观察者，也就是什么时候调用subscribe

迭代器模式

2.3 Hot Observable 和 Cold Observable

在一个Observable对象有两个Obserer对象来订阅，并且这两个并非同时订阅的。
就会形成两个状况：

错过就错过了 ----Hot Observable
不能错过，错过的也得补发 ----Cold Observable

2.4 操做符

在现实的复杂问题，并不会创造一个数据流以后就直接经过subscribe街上一个Observer，每每须要对那个数据流作一系列的处理，而后才交给Observer。数据从管道的一段流入，途径管道各个环节，当数据到达Observer的时候，已经被管道操做过。

就像上面故事同样，拿到鱼，以后可一作成鱼汤，而后加上米饭，最后在给饥饿的人。

在Rxjs中有一系列用于产生Observable函数，那些函数有的凭空创造Observable对象，有的根据外部数据源产生Observable对象，更多的是根据其余的Observable中的数据来产生新的Obsercable对象，也就是把上游数据转化为下游的数据。那些函数统称为操做符。

2.5 弹珠图

弹珠图，能够形象且具体的方式来描述数据流

那个网址能够生成：https://rxviz.com/

3、创造数据流

3.1 建立同步数据流

create: 直接建立
of: 列举数据

当鱼是现成的，可是是散装的时候，好比昨天还存了几条在船上，用of装到竹筒中

var source$ = Observable.of(1, 2, 3);

range: 指定范围

var source$ = Observable.range(1, 100);

genetate: 循环建立

const source$ = Observable.genetate(
    value => vlalue < 10,
    value => value + 2,
    value => value * value
)

repeat: 重复数据的数据流
是一个实例操做符

const source$ = Observable.of(1, 2, 3);
const reapeated$ = source$.repeat(10);
=>1,2,3,1,2,3 ... 重复了1，2，3 10次

empty: 直接完结的Observable
never: 一直待着，不作任何事
throw：抛出错误

var source = Rx.Observable.empty(); // 一条鱼都没有捕捉到的状况,
直接触发observer中complete的执行
结果为 // complete!
var source = Rx.Observable.never();  // 渔人累了，无论是捕到鱼仍是捕
不到鱼都没有力气向岸边上的饥民发出告知了。
结果为 // complete永远都不会触发
var source = Rx.Observable.throw('ill'); // 当渔人生病了，或者要去会个老朋友，会向岸边的饥民（observer)用竹筒呐喊一声告知，
这样饥民就想别的办法（触发error方法）解决当天的食物问题。

3.2 建立异步数据流

定时产生interval和timer

对应js里面的setInterval和setTimeout

const source$ = Observable.interval(1000); // 从0开始没隔1秒输出一个0，1，2

const source$ = Observable.timer(1000); // 1s以后产生0，而后结束
const source$ = Observable.timer(2000, 1000);// 2秒以后0， 而后没隔1s产生1,2,3

form: 可把一切转化为Observable，把数组、字符串、promise、Observable
fromEvent(document.body,’click’); 转化dom事件
formEventPattern 自定义的事件
ajax

Rx.Observable.fromEvent(document.querySelector('#getStart'), 'click')
    .subscribe(
        v => { 
            Rx.Observable.ajax('https://api.github.com/repos/ReactiveX/rxjs', {
                responseType: 'json'
            }).subscribe(value => {
                const startCount = value.response.stargazers_count;
                document.querySelector('#text').innerText = startCount;
            })
        }
    );

repeatWhen: 过一段时间在repeat
defer: 只有在订阅的时候还会执行，节省内存

const observableFactory = () => Observable.ajax(ajaxUrl);
const source$ = Observable.defer(observableFactory)

4、合并数据流

4.1 基本的合并

concat: 收尾相连

相似数组的concat方法，可是只有第一个结束还会取第二个，若是第一个是无限的就到不了第二个
支持静态和动态两种方式

它按顺序订阅每一个输入流并发出其中全部的值，同一时间只会存在一个订阅。只有当前输入流完成的状况下才会去订阅下一个输入流并将其值传递给结果流。

当全部输入流完成时，结果流就会完成，如何任意输入流报错，那么结果流就会报错。若是某个输入流没有完成的话，那么结果流便不会完成，这意味着某些流永远都不会被订阅。

若是值发出的顺序很重要，而且你想要传给操做符的第一个输入流先发出值的话，那么请使用 concat 操做符。举个例子，有两个流，一个从缓存中获取值，一个从远程服务器获取值。若是你想要将二者组合起来并确保缓存中的值先发出的话，就可使用 concat 。

merge：先到先得快速经过

跟cancat不一样的地方是merge会同时订阅全部的上游Observable，而后对上游数据采起先到先得的策略。

zip: 拉链式组合
zip须要配对出现

const source1$ = Rx.Observable.of([1, 2, 3]);
const source2$ = Rx.Observable.of(['a', 'b', 'c']);

const zipped$ = Rx.Observable.zip(source1$, source2$);

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

combineLatest: 合并最后一个数据

当任何一个上游Observable产生数据时，从全部输入Observable对象中拿最后一次产生的数据(最新数据)，而后把那些数据组合起来传给下游。

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

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

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 500, 3, 'partial');
combineLatest(a, b).subscribe(fullObserver('latest'));

withLatestFrom

withLatestFrom的功能相似于combineLatest,可是给下游推送数据只能由一个上游Observable对象驱动

concat/merge/zip/combineLatest都是支持静态操做符合实例操做符两种方式，并且做为输入的Observable对象地位都是对等的；可是withLatestFrom只有实例操做符，并且全部输入Observable的地位不一样，调用withLastestFrom的那个Observable对象起到主导数据生产节奏的做用，做为参数的Observable对象只能贡献数据，不能控制生产数据的时机。

当有一个主线流，同时还须要其余流的最新值时，可使用此操做符。��withLatestFrom 与 combineLatest 有些相似，不一样之处在于 combineLatest 是当任意输入流发出值时，结果流都发出新的值，而 withLatestFrom 是只有当主线流发出值时，结果流才发出新的值。

如同 combineLatest ，withLatestFrom 会一直等待每一个输入流都至少发出一个值，当主线流完成时，结果流有可能在完成时从未发出过值。若是主线流不完成的话，那么结果流永远不会完成，若是任意输入流报错的话，结果流也将报错。

在下面的动图中，能够看到 withLatestFrom 操做符组合了两个流 A 和 B ，B 是主线流。每次 B 发出新的值时，结果流都会使用 A 中的最新值来发出组合值:

const a = stream('a', 3000, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

b.pipe(withLatestFrom(a)).subscribe(fullObserver('latest'));

const { timer } = Rx;
const { map, withLatestFrom } = RxOperators;
// 没隔2秒生产一个数据，经过map的映射，实践产生的数据0、100、200
const source$ = timer(0, 2000).pipe(map(x => 100 * x));
// 从500毫秒，没隔1秒生产一个从0开始的递增数字序列
const secondSource$ = timer(500, 1000);

source$.pipe(
  withLatestFrom(secondSource$, (a, b) => a+b)
);

没个2秒钟输出一行
101
203
305
407

时间	source$	secondSource$	输出
0	0	空
500	空	0
1500	空	1
2000	100	1	101
2500	空	2
3500	空	3
4000	200	3	203

race 胜者通吃

race(): Observable
它自己并对流进行任何组合，而是选择第一个产生值的流。一旦第一个流发出值后，其余的流就会被取消订阅，彻底忽略掉。

当被选中的流完成时，结果流也随之完成，若是被选中的流报错，那么结果流也将报错。一样，若是被选中的流不完成，那么结果流便永远不会完成。

若是有多个提供值的流时此操做符会很是有用，举个例子，某个产品的服务器遍及世界各地，但因为网络条件，延迟是不可预测的，而且差别巨大。使用 race 的话，能够向多个数据源发送一样的请求，随后消费首个响应请求的结果

使用首先发出值的 observable

const $example = Rx.Observable.race(
    Rx.Observable.interval(1500),
    Rx.Observable.interval(2000),
    Rx.Observable.interval(1200),
    Rx.Observable.interval(1000).mapTo('1s won!')
)

const subscribe = $example.subscribe(val => console.log(val))
// 输出: "1s won!"..."1s won!"...etc

startWith

函数签名: startWith(an: Values): Observable 发出给定的第一个值

在开头补充一些数据

// 每1秒发出值
const source = Rx.Observable.interval(1000);
// 以 -3, -2, -1 开始
const example = source.startWith(-3, -2, -1);
// 输出: -3, -2, -1, 0, 1, 2....
const subscribe = example.subscribe(val => console.log(val));

forkJoin Rx 版的 Promise.all()

forkJoin(...args, selector : function): Observable 当全部 observables 完成时，发出每一个 observable 的最新值

只有当全部的Observable对象都完结，确认不会有新的数据产生的时候，forkJoin就会把全部输入Observable对象产生的最后一个数据合并成给下游惟一的数据
它可能与 Promise.all 的使用方式相似。

// RxJS v6+
import { mergeMap } from 'rxjs/operators';
import { forkJoin, of } from 'rxjs';

const myPromise = val =>
  new Promise(resolve =>
    setTimeout(() => resolve(`Promise Resolved: ${val}`), 5000)
  );

const source = of([1, 2, 3, 4, 5]);
// 发出数组的所有5个结果
const example = source.pipe(mergeMap(q => forkJoin(...q.map(myPromise))));
/*
  输出:
  [
   "Promise Resolved: 1",
   "Promise Resolved: 2",
   "Promise Resolved: 3",
   "Promise Resolved: 4",
   "Promise Resolved: 5"
  ]
*/
const subscribe = example.subscribe(val => console.log(val));

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

forkJoin(a, b).subscribe(fullObserver('forkJoin'));

9.pairwise

// RxJS v6+
import { pairwise, take } from 'rxjs/operators';
import { interval } from 'rxjs';

// 返回: [0,1], [1,2], [2,3], [3,4], [4,5]
interval(1000)
  .pipe(
    pairwise(),
    take(5)
  )
  .subscribe(console.log);

将前一个值和当前值做为数组发出

4.2 高阶Observable 高级数据流

高阶Observable就是生产的数据依然是Observable的Observable，以前介绍的Observable就是一阶高阶组件

const { interval } = Rx;
const { map, take  } = RxOperators;

interval(1000)
  .pipe(
        take(2),
        map(x => interval(1500).pipe(map(y => x + ":" + y),take(2))))

一、操做高阶Observable的合并类操做符

处理高阶Observable的合并操做符，就是在原来操做符后面添加All

.concatAll
.mergeAll
.combineALL

二、concatAll

此操做符将合并全部内部流发出的值，合并方式就如同 concat 操做符，是按顺序链接。

在下面的动图中，能够看到高阶流 H ，它会生成两个内部流 A 和 B 。 concatAll 操做符首先从流 A 中取值，而后再从流 B 中取值，并将全部值传递到结果流中

const a = stream(‘a’, 200, 3);
const b = stream(‘b’, 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(concatAll()).subscribe(fullObserver(‘concatAll’));

concatAll首先会订阅上游产生的第一个内部Observable对象
抽取其中的数据，而后，只有当第一个Observable对象完结的时候，才会去订阅第二个内部Obserbale对象。后面的Obsevable对象时懒执行的。

可是如何concatAll消耗内部Observable的速度永远追不上产生内部Observable对象的速度。如何一直那样就会形成内存积压，就是内存泄漏

三、mergeAll

合并全部内部流发出的值，合并方式就如同 merge 操做符，是并发的。

mergeAll只要发现上游产生一个内部Observable就会马上订阅，并从中抽取

在下面的动图中，能够看到高阶流 H ，它会生成两个内部流 A 和 B 。 mergeAll 操做符将合并这两个流中的值，每当发出值时值便会传递到结果流中。

const a = stream(‘a’, 200, 3);
const b = stream(‘b’, 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(mergeAll()).subscribe(fullObserver(‘mergeAll’));

四、combineAll

当源 observable 完成时，对收集的 observables 使用 combineLatest

// RxJS v6+
import { take, map, combineAll } from 'rxjs/operators';
import { interval } from 'rxjs';

// 每秒发出值，并只取前2个
const source = interval(1000).pipe(take(2));
// 将 source 发出的每一个值映射成取前5个值的 interval observable
const example = source.pipe(
  map(val => interval(1000).pipe(map(i => `Result (${val}): ${i}`), take(5)))
);
/*
  soure 中的2个值会被映射成2个(内部的) interval observables，
  这2个内部 observables 每秒使用 combineLatest 策略来 combineAll，
  每当任意一个内部 observable 发出值，就会发出每一个内部 observable 的最新值。
*/
const combined = example.pipe(combineAll());
/*
  输出:
  ["Result (0): 0", "Result (1): 0"]
  ["Result (0): 1", "Result (1): 0"]
  ["Result (0): 1", "Result (1): 1"]
  ["Result (0): 2", "Result (1): 1"]
  ["Result (0): 2", "Result (1): 2"]
  ["Result (0): 3", "Result (1): 2"]
  ["Result (0): 3", "Result (1): 3"]
  ["Result (0): 4", "Result (1): 3"]
  ["Result (0): 4", "Result (1): 4"]
*/
const subscribe = combined.subscribe(val => console.log(val));

4.3 进化的高阶Observable处理

针对上游数据可能产生的积压状况，不少场景并不须要无损的数据流链接，能够舍弃一些数据，至于怎么舍弃，就涉及另外两个合并类操做符，分别是switch和exhaust，这两个操做符是concatAll的进化版

五、SwitchAll 切换输入Obserable

有时候从全部内部流中接收值并不是是咱们想要的效果。在某些场景下，咱们可能只对最新的内部流中的值感兴趣。一个比较好的例子就是搜索。当用户输入关键字时，就向服务器发送请求，由于请求是异步的，因此返回的请求结果是一个 observable 。在请求结果返回以前，若是用户更新了搜索框中的关键字会发生什么状况？第二个请求将会发出，如今已经有两个请求发送给服务器了。可是，第一次搜索的结果用户已经再也不关心了。更有甚者，若是第一次的搜索结果要是晚于第二次的搜索结果的话 (译者注: 好比服务器是分布式的，两次请求请求的不是同一个节点)，那么用户看到的结果将是第一次的，这会让用户感到困扰。咱们不想让这种事情发生，这也正是 switchAll 操做符的用武之地。它只会订阅最新的内部流并忽略(译者注: 忽略 = 取消订阅)前一个内部流。

每当SwitchAll的上游高阶Observable产生一个内部Observable对象，SwitchAll都会马上订阅最新的内部Observable对象上，如歌已经订阅了以前非内部Observable对象，就会退订那个过期的内部Observable对象，这种用上新的，舍弃旧的，就是切换。

能够看到高阶流 H ，它会生成两个内部流 A 和 B 。switchAll 操做符首先从流 A 中取值，当发出流 B 的时候，会取消对流 A 的订阅，而后从流 B 中取值，并将值传递到结果流中。

const a = stream(‘a’, 200, 3);
const b = stream(‘b’, 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(switchAll()).subscribe(fullObserver(‘switchAll’));

exhaust 含义耗尽

在耗尽当前内部Observable的数据以前不会切换到下一个内部Observable对象，
前一个还没完结，新的有产生了，switch是选择新的，exhaust是选择旧的。

5、辅助类操做

5.1 数学类操做符

count: 统计数据个数
max: 最大值
min: 最小值
reduce: 很数组同样

const source$ = Rx.Observable.range(1, 100).reduce((acc, current) => acc + current, 0).subscribe(
    v => {
        console.log('Value', v)
    },
    e => {
        console.log('Error', e)
    },
    () => {
        console.log('Completed')
    }
);

Value 5050
Completed

5.2 条件布尔类操做符

every 所有条件都正确，返回true
find 返回知足条件的值
findIndex 返回知足条件的值的索引
isEmpty 用于检查一个上游Observable对象是否是空的
defaultEnpty 判断上游是不是空，能够穿默认值，若是是空，返回默认值

const findIndex$ = Rx.Observable.of(3,1,4,1,5,9).findIndex(x => x % 2 === 0);

6、过滤数据流

过滤类操做符最基本的功能就是对一个给定的数据流中每一个数据判断是否知足某个条件，若是知足条件就能够传递给下游，不然就抛弃。

filter: 过滤

first: 返回第一个元素

last: 返回最后一个元素

take: 拿操做符

take只支持一个参数count，也就是限定拿上游的Observable的数据数量。

takeLast 拿最后的几个

takeWhile 接受一个断定函数做为参数，断定函数有两个参数，分别表明上游的数据和对应的序号，takeWhile

const source$ = Rx.Observable.range(1, 100).takeWhile(
    value => value % 2 === 0
).subscribe(
    v => {
        console.log('Value', v)
    },
    e => {
        console.log('Error', e)
    },
    () => {
        console.log('Completed')
    }
);
1 4 9 16 Completed

takeUnit
能够用Observable对象来控制另外一个Observable对象的数据产生。

takeUnit的神奇特色就是其参数是另外一个Observable对象notifier，由那个notifier来控制何时结束从上游Oservable拿数据。

const source$ = Rx.Observable.interval(1000);
const notifier$ = Rx.Observable.timer(2500);
const takeUnit$ = source$.takeUnit(notifier$);

skip 跳过

跳过前n个以后全拿

const source$ = Observable.interval(1000);
const skip$ = source$.skip(3);

在等待了3秒以后，skip$会吐出三、四、5...每隔一秒吐出一个递增的证整数

skipWhile和skipUnit 和take反向

6.2 回压控制

debounceTime

舍弃掉在两次输出之间小于指定时间的发出值,诸如预先知道用户的输入频率的场景下很受欢迎

const input = document.getElementById('example');
// 对于每次键盘敲击，都将映射成当前输入值
const example = Rx.Observable.fromEvent(input,'keyup').map( 
    i => i.currentTarget.value
);
// 在两次键盘敲击之间等待0.5秒方才发出当前值，
// 并丢弃这0.5秒内的全部其余值
const debouncedInput = example.debounceTime(500);
// 输出值
const subscribe = debouncedInput.subscribe(val =>{  
    console.log(`Debounced Input: ${val}`);
});

throttleTime

当指定的持续时间通过后发出最新值

每5秒接收最新值:

// 每1秒发出值
const source = Rx.Observable.interval(1000);
/*节流5秒节流结束前发出的最后一个值将从源 observable 中发出*/
const example = source.throttleTime(5000);
// 输出: 0...6...12
const subscribe = example.subscribe(val => console.log(val));

debounce

根据一个选择器函数，舍弃掉在两次输出之间小于指定时间的发出值,尽管没有debounceTime使用普遍，但当 debounce 的频率是变量时,debounce是很重要的

// 发出四个字符串
const example = Rx.Observable.of('WAIT','ONE','SECOND','Last will display');
/*只有在最后一次发送后再通过一秒钟，才会发出值，并抛弃在此以前的全部其余值*/
const debouncedExample = example.debounce(()=> Rx.Observable.timer(1000));
/*在这个示例中，全部的值都将被忽略，除了最后一个输出: 'Last will display'*

const subscribe = debouncedExample.subscribe(val => console.log(val));

4.throttle

仅当由提供的函数所肯定的持续时间已通过去时才发出值

// 每1秒发出值
const source = Rx.Observable.interval(1000);
// 节流2秒后才发出最新值
const example = source.throttle(val => Rx.Observable.interval(2000));
// 输出: 0...3...6...9
berconst subscribe = example.subscribe(val => console.log(val));

sample和sampleTime

当提供的 observable 发出时从源 observable中取样

// 每1秒发出值
const source = Rx.Observable.interval(1000);
// 每2秒对源 observable 最新发出的值进行取样
const example = source.sample(Rx.Observable.interval(2000));
// 输出: 2..4..6..8..
const subscribe = example.subscribe(val => console.log(val));

auditTime 和 audio

throttle把第一个暑假传给下游，audio是把最后一个暑假传给下游

6.3 其余过滤方式

ignoreElements
忽略全部通知，除了 complete 和 error
elementAt
把上游数据当数组，只获取指定下标的那一个数据
single
发出经过表达式的单一项

// 发出 (1,2,3,4,5)
const source = Rx.Observable.from([1,2,3,4,5]);
// 发出匹配断言函数的一项
const example = source.single(val => val ===4);
// 输出: 4
const subscribe = example.subscribe(val => console.log(val));

7、异常错误处理

try/catch只能在同步代码中使用

catch

优雅地处理 observable 序列中的错误

捕获 observable 中的错误

//emit error
const source = Rx.Observable.throw('This is an error!');
//gracefully handle error, returning observable with error message
const example = source.catch(val => Rx.Observable.of(`I caught: ${val}`));
//output: 'I caught: This is an error'
const subscribe = example.subscribe(val => console.log(val));

捕获拒绝的 promise

//create promise that immediately rejects
const myBadPromise = () => new Promise((resolve, reject) => reject('Rejected!'));
//emit single value after 1 second
const source = Rx.Observable.timer(1000);
//catch rejected promise, returning observable containing error message
const example = source.flatMap(() => Rx.Observable
                                       .fromPromise(myBadPromise())
                                       .catch(error => Rx.Observable.of(`Bad Promise: ${error}`))
                                    );
//output: 'Bad Promise: Rejected'
const subscribe = example.subscribe(val => console.log(val));

retry

若是发生错误，以指定次数重试 observable序列

出错的话能够重试2次

//emit value every 1s
const source = Rx.Observable.interval(1000);
const example = source
  .flatMap(val => {
    //throw error for demonstration
    if(val > 5){
      return Rx.Observable.throw('Error!');
    }
    return Rx.Observable.of(val);
  })
  //retry 2 times on error
  .retry(2);
/*
  output: 
  0..1..2..3..4..5..
  0..1..2..3..4..5..
  0..1..2..3..4..5..
  "Error!: Retried 2 times then quit!"
*/
const subscribe = example
  .subscribe({
     next: val => console.log(val),
     error: val => console.log(`${val}: Retried 2 times then quit!`)
});

retryWhen

当发生错误时，基于自定义的标准来重试observable 序列

//emit value every 1s
const source = Rx.Observable.interval(1000);
const example = source
  .map(val => {
    if(val > 5){
     //error will be picked up by retryWhen
     throw val;
    }
    return val;
  })
  .retryWhen(errors => errors
               //log error message
               .do(val => console.log(`Value ${val} was too high!`))
               //restart in 5 seconds
               .delayWhen(val => Rx.Observable.timer(val * 1000))
            );
/*
  output: 
  0
  1
  2
  3
  4
  5
  "Value 6 was too high!"
  --Wait 5 seconds then repeat
*/
const subscribe = example.subscribe(console.log);

finally
报不报错，都会执行

8、转化数据流

8.1 映射数据

map 对源 observable 的每一个值应用投射函数
mapTo 将每一个发出值映射成常量

// 每2秒发出值
const source = interval(2000);
// 将全部发出值映射成同一个值concatMapTo
const example = source.pipe(mapTo('HELLO WORLD!'));
// 输出: 'HELLO WORLD!'...'HELLO WORLD!'...'HELLO WORLD!'...
const subscribe = example.subscribe(val => console.log(val));

pluck 选择属性来发出

const source = from([{ name: 'Joe', age: 30 }, { name: 'Sarah', age: 35 }]);
// 提取 name 属性
const example = source.pipe(pluck('name'));
// 输出: "Joe", "Sarah"
const subscribe = example.subscribe(val => console.log(val));

8.2 缓存窗口：无损回压控制

windowTime和bufferTime

根据时间来缓存上游的数据，基本用法就是一个参数来指定产生缓冲窗口的间隔

const source$ = Rx.Observable.timer(0, 100);
const result$v= source$.windowTime(400);

windowTime的参数是400，也就会把时间划分为连续的400毫秒长度区块，在每一个时间区块中，上游传下来的数据不会直接送给下游.

bufferTime产生的是普通的Observable对象，其中数据是数组的形式，bufferTime会把时间区块内的数据缓冲，在时间区块结束的时候把全部的缓存数据放在一个数组里传给下游。

还可使用第二参数，等于指定每一个时间区块开始的时间间隔

windowCount和bufferCount
根据个数来界定区间的windowCount和bufferCount

const source$ = Observable.timer(0, 100);
const soutce$ = Observable.windowCount(4);

效果是同样的

如何第二个参数比第一个参数大，就会丢弃一些数据

windowWhen和bufferWhen
用Observable对象来控制Observable对象的生成。when就是接受一个函数做为参数，那个参数为closingSelector，返回一个Observable对象，用于控制上游的数据分割，每当返回的Observable对象产生数据或者完结时，windowWhen就认为是一个缓冲区块的结束，从新开启一个缓冲窗口。

const source$ = Observable.timer(0,100);
const closingSelector = () => {
    return Observable.timer(400);
}
const result$ = source$.windowWhen(closingSelector);

不经常使用

windowToggle和bufferToggle
togggle的含义就是两个状态至今的切换，windoeToggle和bufferToggle也是利用Observable来控制缓冲窗口的开和关。

须要两个参数，opening$是一个Observable对象，每当opening$产生一个数据，表明一个缓冲窗口的开始，同时，第二个参数closingSelector也会被调用，用来得到缓冲窗口结束的通知。

const source$ = Observable.timer(0, 100);
const openings$ = Observable.timer(0, 400);

const closingSelector = value => {
    return value % 2 === 0 ? Observable.timer(200): Observable.timer(100);
}
const result$ = source$.windowToggle(openings$, closingSelector);

window和buffer
只支持一个参数，每当notifer$产生一个数据。及时前一个缓存窗口的结束，也是后一个缓存的开始

const source$ = Observable.timer(0, 100);
const notifer$ = Observable.timer(400, 400);
const result$ = source$.window(notifer$);

8.3 高阶的map

concatMap
mergeMap
switchMap
exhaustMap

concatMap = map + concatAll
mergeMap = map + mergeAll
switchMap = map + switch
exhaustMap = map + exhaust

全部xxxxMap名称模式的操做符，都是一个map加上一个“砸平”操做的组合。

有趣的事是映射操做符 concatMap、 mergeMap 和 switchMap 的使用频率要远远高于它们所对应的处理高阶 observable 的操做符 concatAll、 mergeAll 和 switchAll 。可是，若是你细想一下，它们几乎是同样的。全部 *Map 的操做符实际上都是经过两个步骤来生成高阶 observables 的，先映射成高阶 observables ，再经过相对应的组合逻辑来处理高阶 observables 所生成的内部流。

咱们先来看下以前的 meregeAll 操做符的代码示例:

const a = stream('a', 200, 3);
const b = stream('b', 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(mergeAll()).subscribe(fullObserver('mergeAll'));

map 操做符生成了高阶 observables ，而后 mergeAll 操做符将这些内部流的值进行合并，使用 mergeMap 能够轻松替换掉 map 和 mergeAll ，就像这样:

const a = stream('a', 200, 3);
const b = stream('b', 200, 3);
const h = interval(100).pipe(take(2), mergeMap(i => [a, b][i]));

h.subscribe(fullObserver('mergeMap'));

两段代码的结果是彻底相同的

concatMapTo
mergeMapTo
switchMap
expand 相似mergeMap可是会返回给本身，会形成数据爆炸增加

8.4 数据分组

数据分组和合并是相反的，数据分组是把一个数据流拆分为多个数据流

groupBy

输出是一个高阶Obserable对象，每一个内部Obserable对象包含上游产生的知足某个条件的数据
能够当作是一个分发器，对于上游推送下来的任何数据，检查这个数据的key值，若是这个key值是第一次出现，就产生一个新的内部Observable对象，同时这个数据就是内部Observable对象的第一个数据；如何key治已经出现过，就直接把那个数据塞给对应的内部Observable对象。

const intervalStream$ = Observable.interval(1000);
const groupByStream$ = intervalStreanm$.groupBy(
    x => x % 2
)

partition

接受一个断定函数做为参数，对上游的每一个数据进行断定，知足条件的放一个Obserable对象，不知足的放一个，一份为二

8.5 累计数据

scan

const source$ = Observable.interval(100);
const result$ = source$.scan((accumulation,value) => {
    return accumulation + value
})

mergeScan
跟上面的相似可是返回的是一个observable对象，可是不怎么用

9、多播

多波就是一个observable能够有多个subscribe者

9.1 Hot和Cold数据流差别

// 冷的
var cold = new Observable((observer) => {
  var producer = new Producer();
  // observer 会监听 producer
});
// 热的
var producer = new Producer();
var hot = new Observable((observer) => {
  // observer 会监听 producer
});

经过上面的代码发现，冷的 Observables 在内部建立生产者，热的 Observables 在外部建立生产者,由于cold在内部建立，因此屡次订阅就会每次都从新建立，而hot在外部，屡次订阅都是公用一个。因此能够产生多播。在rxjs中能够直接产生Hot Observable：.formPromise .fromEvent .fromEventPattern 那些操做符数据库都是来自外部，真正的数据源和有没有Observver没有任何关系。真正的多播，一定是不论有多少Pbservable来subscribe，推给Observer的都是同样的数据源，知足那种条件的，就是Hot Observable，由于Hot Observable 中的内容建立和订阅者无关。

9.2 Subject

如何把cold变成hot的就须要subject

Observer - 拥有 next、error 和 complete 方法。
Observable - 拥有 Observable 的全部操做符，而且你能够订阅它。
Subject 能够在源 Observable 和多个观察者之间充当桥梁或代理，使得多个观察者能够共享同一个 Observable 执行。

const interval$ = Rx.Observable.interval(1000);
const subject = new Rx.Subject();
interval$.subscribe(subject);
subject.subscribe(val => console.log(`First observer ${val}`));

setTimeout(() => {
  subject.subscribe(val => console.log(`Second observer ${val}`))
}, 2000);

9.2 makeHot操做符

subject并非一个操做符，能够本身创造一个：

Rx.Observable.prototype.makeHot = function () {
  const cold$ = this;
  const subject = new Rx.Subject();
  cold$.subscribe(subject);
  return subject;
}

const makeTick$ = Rx.Observable.interval(1000).take(3).makeHot();
makeTick$.subscribe(val => console.log(`First observer ${val}`))

setTimeout(() => {
    makeTick$.subscribe(val => console.log(`Second observer ${val}`))
}, 2000)

First observer 0
First observer 1
Second observer 1
First observer 2
Second observer 2

可是上面的代码有一个漏洞，返回的结果，能够直接调用next，error等方法，从而影响上游的数据。

Rx.Observable.prototype.makeHot = function () {
  const cold$ = this;
  const subject = new Rx.Subject();
  cold$.subscribe(subject);
  return Rx.Oservable.create((observable) => Rx.Subject.subscribe(observer));
}

可是Subject是不能重复使用的。同时若是上游有多少数据，使用合并操做符进行合并，在传给下游。

9.3 基础的多播

multicast
share
publish

multicast是多播操做符的老大，是最底层的实现，因此不怎么用。

const hotSource$ = coldSource$.multicast(new Subject());

返回的是一个Observable对象，是Observable子类ConnecttableObservable的实例对象。

ConnecttableObservable就是“能够被链接的”Observable，那中Observable对象包含一个connect函数，那个函数的做用是触发multicast用Subject对象去订阅上游的Observable，换句话，就是若是不调用connect函数，那个ConnecttableObservable对象就不会从上游Observable哪里得到任何数据。

const makeTick$ = Rx.Observable.interval(1000).take(3).multicast(new Rx.Subject());
makeTick$.subscribe(val => console.log(`First observer ${val}`))

setTimeout(() => {
    makeTick$.subscribe(val => console.log(`Second observer ${val}`))
}, 2000)

上面不会运行,加上connect()还会运行

const makeTick$ = Rx.Observable.interval(1000).take(3).multicast(new Rx.Subject());
makeTick$.subscribe(val => console.log(`First observer ${val}`))

setTimeout(() => {
    makeTick$.subscribe(val => console.log(`Second observer ${val}`))
}, 2000)

makeTick$.connect()

connect是用来控制多播的时机的，可是手动会比较麻烦，因此，ConnecttableObservable实现可refConunt函数

9.4 refConunt

添加还取消经过个数来本身识别

const makeTick$ = Rx.Observable.interval(1000).take(3).multicast(new Rx.Subject()).refCount();
makeTick$.subscribe(val => console.log(`First observer ${val}`))

setTimeout(() => {
    makeTick$.subscribe(val => console.log(`Second observer ${val}`))
}, 2000)

9.5 工厂函数

可是若是后面的弄成5000，第一个已经完结，第二个就不会再被订阅

须要传入的工厂函数

const subjectFactory = () => {
  console.log('enter subjectFactory');
  return new Rx.Subject();
}

const makeTick$ = Rx.Observable.interval(1000)
    .take(3)
    .multicast(subjectFactory)
    .refCount()
makeTick$.subscribe(val => console.log(`First observer ${val}`))

setTimeout(() => {
    makeTick$.subscribe(val => console.log(`Second observer ${val}`))
}, 5000)

enter subjectFactory
First observer 0
First observer 1
First observer 2
enter subjectFactory
Second observer 0
Second observer 1
Second observer 2

es6的写法

const subjectFactory = () => new Subject();

9.6 秘密参数selector

当使用了第二参数是，就不会返回ConnecttableObservable，而是使用selecter参数。换一句话说，只要指定了selector参数，就指定了multicast返回的Observable对象的生成方法。详解...

9.6 publish

publish的实现

function publish(selector) {
    if (selector) {
        return this.multicast(() => new Subject(), selector);
    } else {
        return this.multicast(new Subject();
    }
}

使用

const makeTick$ = Rx.Observable.interval(1000)
    .take(3)
    .publish()
    .refCount()
makeTick$.subscribe(val => console.log(`First observer ${val}`))

setTimeout(() => {
    makeTick$.subscribe(val => console.log(`Second observer ${val}`))
}, 5000)

First observer 0
First observer 1
First observer 2

function shareSubjectFactory() {
    return new Subject();
}

function share() {
    return multicast.call(this, shareSubjectFactory).refCount()
}

简化

Observable.prototype.share = function share() {
    this.multicast(() => new Subject()).refCount();
}

使用

const makeTick$ = Rx.Observable.interval(1000)
    .take(3)
  .share()
makeTick$.subscribe(val => console.log(`First observer ${val}`))

setTimeout(() => {
    makeTick$.subscribe(val => console.log(`Second observer ${val}`))
}, 5000)

First observer 0
First observer 1
First observer 2

Second observer 0
Second observer 1
Second observer 2

9.7高阶多播功能

publishLast
publishReplay
publishBehavior
对应着三个Subject子类
AsyncSubject
ReplaySubject
BehaviorSubject

publishLast和AsyncSubject

function publishLast() {
    return multicast.call(this=, new AsyncSubject())
}

AsyncSubject不会吧上游的Cold Observable的全部数据都转手给下游，它只会记录最后一个数据，当上游Cold Observable完结的时候，才把最后一个数据传递给Observer。同时是可重用的，不论下发添加的是什么数据，返回都是同样的最后一个数据。

const makeTick$ = Rx.Observable.interval(1000)
    .take(3)
  .publishLast()
  .refCount()
  
makeTick$.subscribe(val => console.log(`First observer ${val}`))

setTimeout(() => {
    makeTick$.subscribe(val => console.log(`Second observer ${val}`))
}, 5000)

First observer 2
Second observer 2

1号Observer在4秒的时候得到publishLast所产生的Observable吐出的第一个也是最后一个数据2

2号Observer在5秒是添加，它会马上得到第一个也是最后一个数据2

publishReplay和AsyncSubject

重播

实现

function publishReplay(
    bufferSize = Number.POSITIVE_INFINITY,
    windowTime = Number.POSITIVE_INFINITY
) {
    return multicast.call(this, new ReplaySubject(bufferSize, windowTime))
}

两个参数表明缓存区的大小，通常只会使用第一参数，指定缓存的个数，如何不指定，就是上游来的，多少下游就缓存多少。容易内存溢出。

const makeTick$ = Rx.Observable.interval(1000)
  .take(3)
  .do(x => console.log('source', x))
    .publishReplay()
    .refCount()

makeTick$.subscribe(val => console.log(`First observer ${val}`))

setTimeout(() => {
    makeTick$.subscribe(val => console.log(`Second observer ${val}`))
}, 5000)

source 0
First observer 0
source 1
First observer 1
source 2
First observer 2

Second observer 0
Second observer 1
Second observer 2

2号依然得到了数据，可是没有从新subscribe上游的，而是publishReplay缓存了，而后回放的。

可是要注意须要给publishReplay一个合理的参数，限制缓存的大小。

publishBehavior和 BehaviorSubject

行为，就是添加一个默认的行为，上游尚未吐出数据时，就会马上得到一个默认数据。

function publishBehavior(value) {
    return multicast.call(this, new BehaviorSubject(value));
}

使用

const makeTick$ = Rx.Observable.interval(1000)
  .take(3)
  .do(x => console.log('source', x))
    .publishBehavior(-1)
    .refCount()

makeTick$.subscribe(val => console.log(`First observer ${val}`))

setTimeout(() => {
    makeTick$.subscribe(val => console.log(`Second observer ${val}`))
}, 5000)

First observer -1

source 0
First observer 0
source 1
First observer 1
source 2
First observer 2

10、scheduler

Scheduler能够做为创造类和合并类操做符的函数使用，此外，rxjs还提供了observeOn和subsribeOn两个操做符，用于在数据管道任何位置插入给定Scheduler。

Scheduler能够翻译成“调度器”，用于控制Rxjs数据流中数据信息的推送节奏。
在Rxjs 中，提供了下列Scheduler实例

undefined/null 就是不指定，表明同步执行
asap 尽快执行的Scheduler
async 利用setInterval实现的Scheduler，基于时间吐出数据的场景
queue 利用队列实现的Scheduler，用于迭代一个大的集合的场景
animationFrame 用于动画场景的scheduler

Rxjs默认选择Scheduler的原则是：尽可能减小并发运行。因此，对于range，就比选择undefined了；对于很大的数据，就选择queue；对于时间相关的操做符好比interval，就选择async

实现原理

asap会尽可能是使用Micro Task，而async利用的是Macro Task
queue，如何调用它的schedule函数式参数delay是0，就是同步的方式，如何delay参数大于0，那queue表现其实就和async同样

console.log('beforr schedule');
Rx.Scheduler.async.schedule(() => console.log('async'))
Rx.Scheduler.asap.schedule(() => console.log('asap'))
Rx.Scheduler.queue.schedule(() => console.log('queue'))
console.log('after schedule');

beforr schedule
queue
after schedule
asap
async

支持Scheduler的操做符能够分为两类

第一类就是普通的建立或者组合Observable对象的操做符，是一个可选参数，没有rxjs回提供一个默认的。

第二类就是存在的惟一功能就是应用Scheduler，因此Scheduler实例必要要有参数的，就两个：observeOn和subscribeOn

支持scheduler的建立操做符有

bindCallback
bindNodeCallback
empty
from
formPromise
interval
of
range
throw
timer

合并操做符

concat
merge

observeOn

const source$ = Rx.Observable.range(1, 3);
const asapSource$ = source$.observeOn(Rx.Scheduler.asap);

console.log('before subscribe');

asapSource$.subscribe(
  value => console.log('data', value),
  error => console.log('error', error),
  () => console.log('complete')
);

console.log('after subscribe');

before subscribe
after subscribe
data 1
data 2
data 3
complete

subscribeOn 用来调节订阅用法相似以上

11、工具

操做符函数的实现，每一个操做符都是一个函数，无论实现什么功能，都必须考虑下面那些功能要点：

返回一个全新的observable对象
对上游和下游的订阅及退订处理
处理异常状况
及时释放资源

// 返回一个全新的Obervable对象
function map(project) {
  return new Observable(observe => {
    this.subscribe({
      next: value => observer.next(project(value)),
      error: err => observer.error(error),
      complete: () => observer.complete()
    })
  })
}

// 订阅和退订处理
function map(project) {
    return new Observable(observe => {
        const sub = this.subscribe({
            next: value => observer.next(project(value)),
            error: err => observer.error(error),
            complete: () => observer.complete(),
    })
    return {
      unsubscribe: () => {
        sub.unsubscribe();
      }
    }
    })
}

// 处理异常状况
function map(project) {
    return new Observable(observe => {
        const sub = this.subscribe({
            next: value => {
        try {
          observer.next(project(value))
        } catch (error) {
          observer.error(error)
        }
      },
            error: err => observer.error(error),
            complete: () => observer.complete(),
        })
        return {
            unsubscribe: () => {
                sub.unsubscribe()
            },
        }
    })
}

// 写完如何关联

// 给Observable打补丁
Observable.prototype.map = map;

// 使用bind绑定特定Observable对象

const result$ = map.bind(source$)(x => x * 2);

// 使用lift

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

被迫更名的函数

do => tap
catch => catchError
switch => switchAll
finally => finalize

也可使用新的

const result$ = source$
    |> filter(x => x % 2 === 0)
    |> map(x => x * 2)

let 在5.5以后使用pipe

12、调试和测试

do
调试日志

Observable.prototype.dubug = function(fn){
    if (global.debug) {
        return this.do(fn);
    } else {
        return this;
    }
}

数据流程流
弹珠图
rxjs-spy