1 RX
全称是 Reactive Extensions，它是微软开发并维护的基于 Reactive Programming 范式实现的一套工具库集合；RX结合了观察者模式、迭代器模式、函数式编程来管理事件序列
RX官方文档：点击前往
2 RXJS
RXJS就是RX在JavaScript层面上的实现；RxJS 是一个库，它通过使用 observable 序列来编写异步和基于事件的程序。它提供了一个核心类型 Observable，附属类型 (Observer、 Schedulers、 Subjects) 和受 [Array#extras] 启发的操作符 (map、filter、reduce、every, 等等)，这些数组操作符可以把异步事件作为集合来处理。
RXJS官方文档：点击前往
3 RXJS中解决异步事件管理的一些基本概念
3.1 Observable
可观察对象：表示一个可调用的未来值或者事件的集合
官方文档：点击前往
3.2 Observer
观察者对象：一个回调函数集合，它知道怎么去监听被可观察对象Observable发送的值
官方文档：点击前往
3.3 Subscription
订阅：表示一个可观察对象Observable的执行，主要用于取消可观察对象Observable执行
官方文档：点击前往
3.4 Operators
操作符：就是一些义函数式编程来处理可观察对象Observable，使用像 map、filter、concat、flatMap 等这样的操作符来处理集合。
官方文档：点击前往
3.5 Subject
相当于 EventEmitter，并且是将值或事件多路推送给多个 Observer 的唯一方式。
官方文档：点击前往
3.6 Schedulers
用来控制并发并且是中央集权的调度员，允许我们在发生计算时进行协调，例如 setTimeout 或 requestAnimationFrame 或其他。
官方文档：点击前往
4 简单实例
技巧01：利用 jsbin 这个在线的JS编辑器作为测试编辑器，连接地址 -> 点击前往
技巧02：本博文使用JS版本都是ES6
4.1 单击按钮实例
4.1.1 在HTML中通过script标签引入RXJS库
<script src="https://unpkg.com/@reactivex/rxjs@5.0.3/dist/global/Rx.js"></script>
4.1.2 在HTML中添加一个button标签
<button id="btn" >单击查看效果</button>
4.1.2 通过JS实现RXJS编程
const btn = document.getElementById('btn');
const btn$ = Rx
.Observable
.fromEvent(btn, 'click');
btn$.subscribe(value => console.log('你点击了按钮哟'));
代码解释：
第一行：获取ID为“btn”的DOM节点
第二行：将获取到的DOM节点的单机事件添加到一个可观察对象Observable的序列中
第三行：订阅可观察对象Observable，当对应DOM节点的单机事件被触发时就会执行相应的操作（本实例是打印出一些信息）
4.1.3 效果展示
4.1.4 代码汇总
<!DOCTYPE html> <html> <head> <meta charset="utf-8"> <meta name="viewport" content="width=device-width"> <title>JS Bin</title> </head> <body> <button id="btn" >单击查看效果</button> <script src="https://unpkg.com/@reactivex/rxjs@5.0.3/dist/global/Rx.js"></script> </body> </html>
const btn = document.getElementById('btn');
const btn$ = Rx
.Observable
.fromEvent(btn, 'click');
btn$.subscribe(value => console.log('你点击了按钮哟'));
4.2 文本框输入实例
4.2.1 在HTML中通过script标签引入RXJS库
<script src="https://unpkg.com/@reactivex/rxjs@5.0.3/dist/global/Rx.js"></script>
4.2.2 在HTML中添加一个input标签
<input id="test" type="text" placeholder="请输入一些信息" />
4.2.3 通过JS实现RXJS编程
const test = document .getElementById('test'); const test$ = Rx .Observable .fromEvent(test, 'keyup') .debounceTime(500) .pluck('target', 'value'); test$ .subscribe(value => console.log(value));
代码解释：
fromEvent -> 将DOM节点的keyup事件添加到可观察对象Observable中
debounceTime -> DOM节点的keyup事件触发后0.5秒后才推送
pluck -> 获取触发事件DOM节点的value值
4.2.4 效果展示
4.2.5 代码汇总
<!DOCTYPE html> <html> <head> <meta charset="utf-8"> <meta name="viewport" content="width=device-width"> <title>JS Bin</title> </head> <body> <input id="test" type="text" placeholder="请输入一些信息" /> <script src="https://unpkg.com/@reactivex/rxjs@5.0.3/dist/global/Rx.js"></script> </body> </html>
const test = document
.getElementById('test');
const test$ = Rx
.Observable
.fromEvent(test, 'keyup')
.debounceTime(500)
.pluck('target', 'value');
test$
.subscribe(value => console.log(value));
5 弹珠图
要解释操作符是如何工作的，文字描述通常是不足以描述清楚的。许多操作符都是跟时间相关的，它们可能会以不同的方式延迟(delay)、取样(sample)、节流(throttle)或去抖动值(debonce)。图表通常是更适合的工具。弹珠图是操作符运行方式的视觉表示，其中包含输入 Obserable(s) (输入可能是多个 Observable )、操作符及其参数和输出 Observable 。
6 可观察对象Observable
技巧01：可观察对象Observable是数据的产生者，它会将数据以流的方式推送给观察者Observer；可以将可观察对象Observable看作是一个可以同步、异步返回多个值的函数
const test$ = Rx.Observable .create(observer => { observer.next(1); observer.next(2); observer.next(3); setTimeout(() => observer.next('一秒钟后异步推送的数据'), 1000); setTimeout(() => { observer.next('2秒钟后异步推送的数据'); observer.complete(); }, 2000); });
代码解释：上面的代码中利用create操作符创建了一个Observable，该Observable一旦被订阅就会向Observer推送1、2、3，然后1秒钟后推送“一秒钟后异步推送的数据”，再过1秒钟后推送“2秒钟后异步推送的数据”，紧接着是完成数据流的推送
技巧02：订阅Observable是Observable执行和发送值/事件给Observer最简单的方式；订阅Observable就类似于调用一个拥有多个返回值的函数
const test$Sub = test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('数据推送完成') );
代码解释：上面的代码是订阅Observable的一个实例test$，订阅方法subscribe中的参数是有顺序的，其中第一个参数是位置参数用于处理数据，第二三个参数是可选参数，其中第二个参数用来处理错误，第三个参数用来处理数据推送完成后的一些操作；订阅后会返回一个Subscription对象，可以通过该对象来关闭正在执行的Observable
技巧03：利用Subscription对象来关闭正在执行的Observable
const test$ = Rx.Observable .create(observer => { // const t = setInterval(() => observer.next('hello'), 1000); observer.next(1); observer.next(2); observer.next(3); const st01 = setTimeout(() => observer.next('一秒钟后异步推送的数据'), 1000); const st02 = setTimeout(() => { observer.next('2秒钟后异步推送的数据'); observer.complete(); }, 2000); return function unsubscribe() { clearTimeout(st01); clearTimeout(st02); // clearInterval(t); } }); const test$Sub = test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('数据推送完成') ); setInterval(() => test$Sub.unsubscribe(), 1000); // test$Sub.unsubscribe();
代码解释：使用create创建Observable的时候回返回一个函数unsubscribe，该函数用来定义如何清理执行的资源；订阅该Observable后就会返回一个Subscription对象，通过该Subscription对象来调用unsubscribe方法来取消正在执行的Observable
7 观察者Observer
Observer是Observable推送的值的消费者；Observer是一组回调函数的集合，每个回调函数对应一种 Observable 发送的通知类型：next、error 和 complete。下面的示例是一个典型的观察者对象：
var observer = { next: x => console.log('Observer got a next value: ' + x), error: err => console.error('Observer got an error: ' + err), complete: () => console.log('Observer got a complete notification'), };
技巧01：使用Observer就必须将他们提供给Observable的subscirbe方法
坑01：虽然所Observer是一个对象，但是在将Observer传给Observable的subscribe方法时只需要将对象中的内容传进去就可以啦，例如下面第一种方式就是错误的，第二种方式才是正确的
test$.subscribe( { value => console.log(value), error => console.log(error), () => console.log('数据推送完成') } );
test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('数据推送完成') );
8 操作符
操作符是 Observable 类型上的方法，比如 .map(...)、.filter(...)、.merge(...)，等等。当操作符被调用时，它们不会改变已经存在的 Observable 实例。相反，它们返回一个新的 Observable ，它的 subscription 逻辑基于第一个 Observable。（操作符是函数，它基于当前的 Observable 创建一个新的 Observable。这是一个无副作用的操作：前面的 Observable 保持不变）
操作符本质上是一个纯函数 (pure function)，它接收一个 Observable 作为输入，并生成一个新的 Observable 作为输出。订阅输出 Observalbe 同样会订阅输入 Observable 。
8.1 小案例之mapTo
每次源 Observble 发出值时，都在输出 Observable 上发出给定的常量值；接收常量 value 作为参数，并每当源 Observable 发出值时都发出这个值。换句话说， 就是忽略实际的源值，然后简单地使用这个发送时间点以知道何时发出给定的 value 。
8.1.1 效果展示
8.1.2 代码汇总
<!DOCTYPE html> <html> <head> <meta charset="utf-8"> <meta name="viewport" content="width=device-width"> <title>JS Bin</title> </head> <body> <button id="btn" type="button">点击测试</button> <script src="https://unpkg.com/@reactivex/rxjs@5.0.3/dist/global/Rx.js"></script> </body> </html>
// 根据ID获取DOM对象
const btn = document.getElementById("btn");
// 将对应DOM对象的单击事件变成事件流
const btn$ = Rx.Observable
.fromEvent(btn, "click")
.mapTo('ello');
// 订阅事件流
btn$.subscribe(value => console.log(value));
8.2 创建类操作符
8.2.1 interval
public static interval(period: number, scheduler: Scheduler): Observable
创建一个 Observable ，该 Observable 使用指定的 IScheduler ，并以指定时间间隔发出连续的数字。
坑01：interval产生的第一个数并不是立即就产生的，而是当第一个时间间隔过后才会推送第一个数字，而且推送的数字是从0开始的自增数字
const test$ = Rx.Observable .interval(1000); test$.subscribe(value => console.log(value));
技巧01：官方文档 => 点击前往
8.2.2 timer
public static timer(initialDelay: number | Date, period: number, scheduler: Scheduler): Observable
根据给定的时间间隔自增的产生数字
参数解释：
参数01：产生第一个数字的间隔时间
参数02：产生下一个数字的时间间隔（可选）
参数03：调度器，用来调度值的发送, 提供“时间”的概念 (可选，默认值: async)
坑01：如果只给定第一个参数，那么会在时间间隔到达后推送数字0，然后就不在继续推送数据啦
技巧01：其实timer的用法和interval非常想，都是自动产生自动的数字
const test$ = Rx.Observable .timer(1000, 1000) .take(3) .map(value => value * 10); test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.3 合并类操作符
8.3.1 combineLatest
组合多个 Observables 来创建一个 Observable ，该 Observable 的值根据每个输入 Observable 的最新值计算得出的；组合多个 Observables 来创建一个 Observable ，该 Observable 的值根据每个输入 Observable 的最新值计算得出的。
技巧01：每个Observables都必须有值，而且他们其中必须有一个的值更新后才会执行combineLatest
》效果展示
》代码汇总
<!DOCTYPE html> <html> <head> <meta charset="utf-8"> <meta name="viewport" content="width=device-width"> <title>JS Bin</title> </head> <body> <input type="number" id="weight" placeholder="请输入体重" /> <span>Kg</span> <br /> <input type="number" id="height" placeholder="请输入身高" /> <span>cm</span> <script src="https://unpkg.com/@reactivex/rxjs@5.0.3/dist/global/Rx.js"></script> </body> </html>
const weight = document.getElementById('weight'); const height = document.getElementById('height'); const weight$ = Rx.Observable .fromEvent(weight, 'keyup') .debounceTime(500) .pluck('target', 'value'); const height$ = Rx.Observable .fromEvent(height, 'keyup') .debounceTime(500) .pluck('target', 'value'); const bmi$ = Rx.Observable .combineLatest( weight$, height$, (w, h) => { const str = '成人BMI数值为：'; const bmi = w / (h * h / 100 / 100); if (bmi < 18.5) { return str + bmi + ' -> 过轻'; } else { if (bmi < 23.9) { return str + bmi + ' -> 正常'; } else { if (bmi < 27) { return str + bmi + ' -> 过重'; } else { if (bmi < 32) { return str + bmi + ' -> 肥胖'; } else { return str + bmi + ' -> 非常肥胖'; } } } } } ); // 过轻：低于18.5 // 正常：18.5-23.9 // 过重：24-27 // 肥胖：28-32 // 非常肥胖, 高于32 // weight$.subscribe(value => console.log(value)); // height$.subscribe(value => console.log(value)); bmi$.subscribe(value => console.log(value));
8.3.2 merge
public merge(other: ObservableInput, concurrent: number, scheduler: Scheduler): Observable
将两个数据源的数据合并到一起
const test01$ = Rx.Observable .create(observer => { observer.next(1); setTimeout(() => { observer.next(2); }, 1000); observer.next(3); }); const test02$ = Rx.Observable .create(observer => { observer.next(4); observer.next(5); observer.next(6); }); const test$ = Rx.Observable .merge(test01$, test02$); test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.3.3 startWith
public startWith(values: ...T, scheduler: Scheduler): Observable
向原数据源添加一个数据作为第一个推送的数据
const test01$ = Rx.Observable .create(observer => { observer.next(1); setTimeout(() => { observer.next(2); }, 1000); observer.next(3); }) .startWith('HELLO'); test01$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.3.4 zip
根据顺序将两个数据源中的数据进行组合
技巧01：如果一个数据源中有3个数据，另外一个数据源中有4个数据，那么只会将这两个数据源中前面推送的3个数据进行组合
const test01$ = Rx.Observable .create(observer => { observer.next(1); setTimeout(() => { observer.next(2); }, 1000); observer.next(3); }); const test02$ = Rx.Observable .create(observer => { observer.next('a'); observer.next('b'); observer.next('c'); observer.next('d'); }); const test$ = Rx.Observable .zip(test01$, test02$); test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.4 其他操作符
8.4.1 take
public take(count: number): Observable<T>
只会推送原数据流中前面的count个数据
const test$ = Rx.Observable .interval(1000) .take(3); test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.4.2 map
public map(project: function(value: T, index: number): R, thisArg: any): Observable<R>
对原数据源的每个值做同样的处理
const test$ = Rx.Observable .interval(1000) .take(3) .map(value => value * 10); test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.4.3 filter
public filter(predicate: function(value: T, index: number): boolean, thisArg: any): Observable
对原数据源的数据进行过滤
参数解释：
参数01：过滤操作函数，操作函数中第一个参数代表正在过滤的数据，第二个参数代表原数据源序列的索引（注意：索引是从0开始的，用处还问弄清楚）？？？
参数02：可选参数，用来决定 predicate 函数中的 this 的值
const test$ = Rx.Observable .timer(1000, 1000) .take(5) .map(value => value * 10) .filter((value) => value > 10); test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.4.4 first
public first(predicate: function(value: T, index: number, source: Observable<T>): boolean, resultSelector: function(value: T, index: number): R, defaultValue: R): Observable<T | R>
只推送原数据源中的第一个值或者第一个满足条件的值
技巧01：如果first不携带任何参数，那么将会推送原数据源中的第一个值(和take(1)的效果相同)；如果携带一个返回boolean的方法，那么将返回满足该方法的第一个数据
const test$ = Rx.Observable .timer(1000, 1000) .take(5) .map(value => value * 10) // .first(); .first((value) => value > 10); test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.4.5 last
public last(predicate: function): Observable
只推送原数据源中的最后一个数据
技巧01：如果last不指定参数时，就会默认只推送最后一个数据；如果指定了一个返回boolean类型的方法作为参数，那么就会返回最后一个满足条件的数据
const test$ = Rx.Observable .interval(1000) .take(5) .last(); // .last(value => value < 3 ); test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.4.6 skip
public skip(count: Number): Observable
在推送数据时，跳过前面count个数据
const test$ = Rx.Observable .interval(1000) .take(5) .skip(2); test$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.4.7 scan
public scan(accumulator: function(acc: R, value: T, index: number): R, seed: T | R): Observable<R>
对原数据源序列进行依次累加，每累加玩完一次就会推送一个结果
参数解释：
参数01：一个累加方法，该方法的第一个参数代表累加结果，第二个参数代表当前累加值，该累加值随便用一个变量表示即可
参数02：可选参数，累加器的初始值，如果不填就会默认将原数据源的第一个数据作为初始累加值
const b = 100; const test01$ = Rx.Observable .interval(1000) .take(5) .scan((acc, a) => acc + a, b); test01$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.4.8 reduce
public reduce(accumulator: function(acc: R, value: T, index: number): R, seed: R): Observable<R>
对原数据源中的数据进行累加，最后只将累加的结果推送
参数解释：
参数01：第一个参数是一个累加方法，该累加方法接受两个参数，第一个参数代表累加结果，第二个参数代表当前累加值
参数02：可选参数，该参数如果指定了就会被用作累加的初始值
const b = 100; const test01$ = Rx.Observable .interval(1000) .take(5) .reduce((acc, a) => acc + a, b); test01$.subscribe( value => console.log(value), error => console.log(error), () => console.log('完成') );
8.4.9 debounceTime
public debounceTime(dueTime: number, scheduler: Scheduler): Observable
当没有新的数据添加到原数据源时，延迟发送原数据源中的数据
<!DOCTYPE html> <html> <head> <meta charset="utf-8"> <meta name="viewport" content="width=device-width"> <title>JS Bin</title> </head> <body> <input type='text' id="test" placeholder="测试文本框" /> <script src="https://unpkg.com/@reactivex/rxjs@5.0.3/dist/global/Rx.js"> </script> </body> </html>
const test = document.getElementById('test');
const test$ = Rx.Observable
.fromEvent(test, 'keyup')
.debounceTime(500)
.pluck('target', 'value');
test$.subscribe(
value => console.log(value),
error => console.log(error),
() => console.log('完成')
);
8.4.10 distinct
public distinct(keySelector: function, flushes: Observable): Observable
去除掉原数据源中重复的数据
技巧01：如果distinct方法提供了keySelector方法，就会先利用keySelector方法对原数据源你的数据进行处理后在进行去重操作
官方文档 => 点击前往
const test01$ = Rx.Observable .create(observer => { observer.next(1); observer.next(2); observer.next(3); observer.next(1); observer.next(2); observer.next(3); }) .distinct(); test01$.subscribe(value => console.log(value));
8.4.11 distinctUntilChanged
public distinctUntilChanged(compare: function): Observable
去除掉原数据源相邻数据的重复数据
const test01$ = Rx.Observable .create(observer => { observer.next(1); observer.next(2); observer.next(3); observer.next(3); observer.next(3); observer.next(1); observer.next(2); observer.next(3); }) .distinctUntilChanged(); test01$.subscribe(value => console.log(value));