我所了解的RxJS

简介

RxJS 是使用 Observables的响应式编程的库，它使编写异步或基于回调的代码更容易，是ReactiveX编程理念的JavaScript版本。RxJS的强大之处正是它使用纯函数来产生值的能力。这意味着你的代码更不容易出错。

安装

官方安装

npm install rxjs
/// 导入整个核心功能集：
import Rx from 'rxjs/Rx';
Rx.Observable.of(1,2,3)
复制代码

推荐安装

根据官方安装发现rxjs不能彻底加载，须要依赖rxjs-compat包,推荐使用如下安装

npm i -s Rxjs@6 rxjs-compat@6
import * as Rx from 'rxjs/Rx'
复制代码

RxJS核心概念

Observable简介

Observable举例说明

Rx.Observable.of('1', '2', '3').map(x=>x*10).filter(x=>x>5).subscribe(x=>console.log(x))
复制代码

• 建立过程:Rx.Observable.of('1', '2', '3') 建立一个依次发送一、二、3的observable
• 逻辑过程:*.map().filter()*每一个值乘以10，而后去过滤出大于5的值。若是先写filter操做符，而后再map，则得不到数据
• 订阅过程:*subscribe()*相似回调函数。这个过程会获得一个对象subscription。
• 执行过程:x=>console.log(x) 默认状况下为执行next回调
• 清理过程:示例以下

const subscription = Rx.Observable.of('1','2','3').map(x=>x*10).filter(x=>x>5).delay(1000).subscribe(x=>console.log(x));
subscription.unsubscribe()
复制代码

Subject简介

什么是 Subject？ - RxJS Subject 是一种特殊类型的 Observable，它容许将值多播给多个观察者，因此 Subject 是多播的，而普通的 Observables 是单播的(每一个已订阅的观察者都拥有 Observable 的独立执行)。 每一个 Subject 都是 Observable 。 - 对于 Subject，你能够提供一个观察者并使用 subscribe 方法，就能够开始正常接收值。从观察者的角度而言，它没法判断 Observable 执行是来自普通的 Observable 仍是 Subject 。 在 Subject 的内部，subscribe 不会调用发送值的新执行。它只是将给定的观察者注册到观察者列表中，相似于其余库或语言中的 addListener 的工做方式。 每一个 Subject 都是观察者。 - Subject 是一个有以下方法的对象： next(v)、error(e) 和 complete() 。要给 Subject 提供新值，只要调用 next(theValue)，它会将值多播给已注册监听该 Subject 的观察者们。 Subject 像是 Observable，可是能够多播给多个观察者。Subject 还像是 EventEmitters，维护着多个监听器的注册表。 根据官网，咱们大概能够如下理解： Observable相似单车道单行线，逆行或者多辆车同时开都是不容许的 Subject相似没有监控的双行线，随你往哪里开，怎么开，多少车开都没有问题 因此能够理解Subject是一类特殊的Observable，它能够向多个Observer多路推送数值。普通的Observable并不具有多路推送的能力（每个Observer都有本身独立的执行环境），而Subject能够共享一个执行环境

Subject举例说明

const test = Observable.interval(1000).take(3);
const observerA = {
  v => console.log(`a:${v}`) 
}
const observerB = {
  v => console.log(`b:${v}`)
}                                                                              ///定义好observable
test .subscribe(observerA)
setTimeout(() => {test .subscribe(observerB) }, 2000)    
///由于observable是单播的，因此会输出 a:0、a:一、b:0、a:二、b:一、b:2
const subject = new Subject()
subject.subscribe(observerA)
test.subscribe(subject)
setTimeout(() => {subject.subscribe(observerB)}, 2000)
///由于Subject是多播的，共享一个执行，因此输出为：a:0、a:一、a:二、b:2
复制代码

Subject多态

因为subject的特殊性，衍生出多种subject的变体，具体就不阐述了，他们的对好比下图

Rxjs	是否存储数据	是否须要初始值	什么时候向订阅者发布数据
Subject	否	否	及时发布，有新数据就发布
BehaviorSubject	是，存储最后一条数据或者初始值	是	及时发布，有新数据就发布
ReplaySubject	是，存储全部数据	否	及时发布，有新数据就发布
AsyncSubject	是，存储最后一条数据	是	延时发布，只有当数据源完成时才会发布

Scheduler简介

什么是Scheduler？ - Scheduler控制着什么时候启动 subscription 和什么时候发送通知。它由三部分组成

调度器是一种数据结构。它知道如何根据优先级或其余标准来存储任务和将任务进行排序。 调度器是执行上下文。 它表示在什么时候何地执行任务(举例来讲，当即的，或另外一种回调函数机制(好比 setTimeout 或 process.nextTick)，或动画帧)。 调度器有一个(虚拟的)时钟。 调度器功能经过它的 getter 方法 now() 提供了“时间”的概念。在具体调度器上安排的任务将严格遵循该时钟所表示的时间。 调度器可让你规定 Observable 在什么样的执行上下文中发送通知给它的观察者。

操做符概括

RxJS提供了各类API来建立数据流：

单值：of, empty, never 多值：from 定时：interval, timer 从事件建立：fromEvent 从Promise建立：fromPromise 自定义建立：create

建立出来的数据流是一种可观察的序列，能够被订阅，也能够被用来作一些转换操做，好比：

改变数据形态：map, mapTo, pluck 过滤一些值：filter, skip, first, last, take 时间轴上的操做：delay, timeout, throttle, debounce, audit, bufferTime 累加：reduce, scan 异常处理：throw, catch, finally, retry, 条件执行：takeUntil, delayWhen, retryWhen, subscribeOn, ObserveOn 转接：switch

也能够对若干个数据流进行组合：

concat，保持原来的序列顺序链接两个数据流 merge，合并序列 race，预设条件为其中一个数据流完成 forkJoin，预设条件为全部数据流都完成 zip，取各来源数据流最后一个值合并为对象 combineLatest，取各来源数据流最后一个值合并为数组

RxJS 难点

RxJS 处理异步逻辑，数据流，事件很是擅长。使用Rxjs前处理数据通常是处于一种'上帝'视角来对数据可视化的调试，Rxjs大大缩短了代码量的同时可以更好的达到数据的处理（纯净性）。正是因为其强大的特性，因此学习Rxjs有如下难点（我的认为） 一、抽象程度比较高，须要开发人员具有比较强的概括总结能力 二、操做符多并且杂，须要花大力气记住而且合理使用各个操做符

测试题

• 一、鼠标点击后console相隔2秒输出5的倍数
• 二、现有3个异步操做a、b、c，请提供让三个异步并行完成后同时输出值的方法
• 三、’人和将来大数据‘ ===》 取最后4个字(多种方法)
• 四、模拟一个程序员，工资不涨，天天赚相同的钱n，钱足够了(100n)就买房，买了房而后把房子租给别人，每月收取房租m(5n),而后收入变成n+m，而后钱足够了继续买房，而后继续租给访客，收入变成n+2m

参考答案

//////题目1
const timer = Rx.Observable.interval(2000);
const event = Rx.Observable.fromEvent(document, 'click')
event.switchMap(() => timer)
 .map(x => x * 5)
 .subscribe(x => console.log('第1题:' + x));
复制代码

/////题目2
const fa = (cb) => {
  setTimeout(() => cb('a'), 1000);
}
const fb = (cb) => {
  setTimeout(() => cb('b'), 2000);
}
const fc = (cb) => {
  setTimeout(() => cb('c'), 4000);
}
const oa = Rx.Observable.bindCallback(fa);
const ob = Rx.Observable.bindCallback(fb);
const oc = Rx.Observable.bindCallback(fc);

Rx.Observable.combineLatest(oa(),ob(),oc())
  .subscribe(x => console.log('第2题:' + x));
  /////同时还能够用forkJoin,zip
复制代码

//////题目3
const str = "人和将来大数据";
const param = str.split('');
Rx.Observable.from(param)
  .takeLast(4)
  .subscribe(x => console.log('第3题:' + x));
///////////////////////////////////////////////////////
Rx.Observable.from(param).subscribe(new ReplaySubject(3))
复制代码

///////题目4
const house$ = new Subject()  ///房子
const houseCount$ = house$.scan((acc, num) => acc + num, 0).startWith(0) ///房子数

// 工资始终不涨
const salary$ = Observable.interval(100).mapTo(1) //程序员工资n
const rent$ = Observable.interval(3000)
  .withLatestFrom(houseCount$)
  .map(arr => arr[1] * 5)

// 一买了房，就没现金了……
const income$ = Observable.merge(salary$, rent$)
const cash$ = income$
  .scan((acc, num) => {
    const newSum = acc + num
    const newHouse = Math.floor(newSum / 100)
    if (newHouse > 0) {
      house$.next(newHouse)
    }
    return newSum % 100
  }, 0)
houseCount$.subscribe(num => console.log(`houseCount: ${num}`))
cash$.subscribe(num => console.log(`cash: ${num}`))
复制代码

原文连接：tech.gtxlab.com/sth-about-r…

---

做者简介： 张栓，人和将来大数据前端工程师，专一于html/css/js的学习与开发。