RxJS入门

RxJS是什么？

官方：RxJS 是 Reactive Extensions for JavaScript 的缩写，起源于 Reactive Extensions，是一个基于可观测数据流在异步编程应用中的库。RxJS 是 Reactive Extensions 在 JavaScript 上的实现。

通常说到RxJS，都会讲他是基于流的响应式的结合观察者和迭代器模式的一种库。因此下面会从这几个关键词来说。

前置知识点

响应式编程（RP —— Reactive Programming）

响应式编程是一种面向数据流和变化传播的编程范式。在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值经过数据流进行传播。—— wikipedia

• 响应式编程是使用异步数据流进行编程。常见的异步数据流包含Event buses。用包含这些事件在内的任何东西建立数据流（Data stream），监听他并做出响应。
• 只关注业务逻辑互相依赖的事件而不是实现细节
• 适用于大量和数据有关的事件交互，特别是高实时性要求

流

以点击按钮这一个事件流举例：

屡次点击按钮事件流举例：

• 一个流就是一个不间断的按照时间排序的序列。它产生三种不一样类型事件: 值，错误，完成的信号。对这三个定义事件处理函数，就能够异步的捕获这些事件
• 每一个stream有多个方法，调用时会基于原来的流返回一个新的流，原来的流不作修改，保证不可变性
• 数据流支持链式调用，你能够组合不一样的函数来处理流，建立和过滤不一样的流。甚至一个流或多个流能够做为另一个流的输入。你能够合并两个数据流。你能够过滤一个数据流，从中获取一个包含你感兴趣的事件的数据流。你能够未来自一个数据流的值映射到另一个数据流

观察者模式（发布-订阅模式）

例子：购房者和售房部之间的信息订阅。购房者订阅售房部的房价信息，售房部维护一张须要信息的客户表，当有信息时，遍历表给符合条件的购房者推送发布房屋信息。这里，购房者担任观察者的角色，售房部是被观察的角色，当售房部信息发生变化，则自动推送信息给购房者。

结论：流(售房部/rx.js的Observable)是被观察的，某个函数订阅流的某个事件（推送房价），该函数是观察者(购房者/rx.js的Observer)。当流的某个事件产生了，对应的函数就会被执行。

迭代器模式

提供一种方法顺序访问一个聚合对象中的各个元素，而不须要暴露该对象的内部表示。最多见的就是JavaScript 中像 Array、Set 等这些内置的可迭代类型，能够经过 iterator 方法来获取一个迭代对象，调用迭代对象的 next 方法将获取一个元素对象。

var iterable = [1, 2];
var iterator = iterable[Symbol.iterator]();
iterator.next(); // => { value: "1", done: false}
iterator.next(); // => { value: "2", done: false}
iterator.next(); // => { value: undefined, done: true}复制代码

由上可知JavaScript 的 Iterator 只有一个 next 方法，这个 next 方法只会回传这两种结果。iterator经过调用next获取值，是一种pull数据的形式。

与Promise的区别

1. Promise本质上也是一个Observable，能使用fromPromise把Promise转成Observable
2. 可是Promise .then()只能返回一个值，Observable能够返回多个值
3. Promise要么resolve要么reject，而且只响应一次。而Observable能够响应屡次
4. Promise不能取消，Observable能够调用unsubscribe()取消订阅

解决的问题

• 同步和异步的统一
• 可组合的数据变动过程
• 数据和视图的精确绑定
• 条件变动以后的自动从新计算

核心概念

概述关系

Demo： 一个简单的例子

Observable —— 被观察者

Rxjs是观察者 + 迭代器模式的结合，Observable做为被观察者，是一个值或事件的流集合。就像是一个序列，裡面的元素会随着时间推送。

var observable = Rx.Observable
// 经过create方法建立一个Observable
// 回调函数会接受observer参数，也就是观察者角色
	.create(function(observer) {
		observer.next('hi');
		observer.next('world');

        setTimeout(() => {
			observer.next('这一段是异步操做');
		}, 30)
	})

// 订阅这个 observable
// 只有在订阅以后，才会在流Observable变化的时候，调用observer提供的方法，并通知他 
// 订阅以后也能够取消订阅，调用unsubscribe()便可
console.log('start')
var subscription = observable.subscribe(function(value) {
	console.log(value);
})
console.log('end')
setTimeOut(()=> {
  subscription.unsubscribe()
}, 5000)


// 程序会依次输出
'start'
"hi"
'world'
'end'
'这一段是异步操做'复制代码

因此，Observable不一样于观察者模式中的被观察者，他没有一份须要维护订阅者的清单，他只是一个函数。想要订阅他只须要传进回调函数observer就好。而且，Observable 能够同时处理同步和异步操做！

同时，有不少建立Observable的方法，经常使用的以下：

Operator —— 操做符

操做Observable的函数就是操做符。他会接受传入的Observable，可是会返回新的Observable。用map举例说明。

Rx.Observable.of(2)
             .map(v => v * 2) 
             .subscribe(v => console.log('output:' + v));
// output:4复制代码

下面介绍几个经常使用的操做符。具体的API可视化数据流动可参见 宝石图

Observer —— 观察者

和迭代器模式一一对应，提供三个方法，next、error、complete

var Observer = {
    next(value) { /* 处理值*/ },
    error(error) { /* 处理异常 */ },
    complete() { /* 处理已完成态 */ }
};

next(): 接收Observable发出的值  （必传）
error(): 不一样于迭代器里面用try catch，Observer用error方法接收错误 （可选）
complete(): 当没有新的数据发出的时候，触发操做  （可选）复制代码

Subject —— 观察模式的实现，并继承Observable

同一个Observable能够被多个observer订阅。和addListener相似她们由Subject维护列表，Subject能够向多个Observer多路推送数值，是一类特殊的Observable。

• 每个Subject都是一个Observable，能够订阅他。从Observer的视角看，它并不能区分本身的执行环境是普通Observable的单路推送仍是基于Subject的多路推送
• 每个Subject也能够是Observer，由于他一样由next、error、complete方法组成，调用next方法，Subject会给全部在他上面注册的Observer多路推送当前的值

demo地址

// 建立一个Observable，一秒钟输出一个数字，只取三个就结束
var source = Rx.Observable.interval(1000).take(3);

// 定义两个observer对象
var observerA = {
    next: value => console.log('A next: ' + value),
    error: error => console.log('A error: ' + error),
    complete: () => console.log('A complete!')
}

var observerB = {
    next: value => console.log('B next: ' + value),
    error: error => console.log('B error: ' + error),
    complete: () => console.log('B complete!')
}

// 建立一个subject —— 特殊的Observable
var subject = new Rx.Subject()

// observerA订阅Subject
subject.subscribe(observerA)

// Subject又以observer的身份订阅Observable
source.subscribe(subject);

setTimeout(() => {
    subject.subscribe(observerB);
}, 1000);

// 输出：
// "A next: 0"
// "A next: 1"
// "B next: 1"
// "A next: 2"
// "B next: 2"
// "A complete!"
// "B complete!"
 A、B两个observer互不影响，是独立的复制代码

Scheduler —— 控制Observable的时间节点

控制一个Observable的订阅开始执行和值送达的时机。RxJS 5中提供了4种scheduler。通常operator会有预设的scheduler。

• queue —— 递归的时候，queue会把递归阻塞，避免没必要要的性能损耗
• asap —— as soon as possible，表现为setTimeout时间为0。多用于永不退订的Observable，好比轮询
• async —— 规定当前的Observable执行方式为异步，用setInterval实现。
• animationFrame —— Window.requestAnimationFrame这个API实现，适合高频率的UI动画触发

var observable = Rx.Observable.create(function (observer) {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.complete();
});

console.log('before subscribe');
observable.observeOn(Rx.Scheduler.async) // 原本是同步的，变成了异步
.subscribe({
    next: (value) => { console.log(value); },
    error: (err) => { console.log('Error: ' + err); },
    complete: () => { console.log('complete'); }
});
console.log('after subscribe');

// "before subscribe"
// "after subscribe"
// 1
// 2
// 3
// "complete"复制代码

应用场景

例子： Trello 、 Teambition等协做工做软件

核心：视图的状态是一个时间轴上的若干流的叠加效果。

业务场景特色：

1. 有不少数据，很是多关于数据的操做
2. 展现的数据是多个数据组合而成，好比任务、对应owner、标签等
3. 同一个数据的更新，可能来自不一样的发起方
4. 新增的数据须要的数据处理规则应该和原来的相同

解决：

1. 数据经过缓存和异步方式获取
2. 把每一个数据流管道组合起来，流的叠合就是最后的数据
3. 获取和订阅放在一块儿，也就不须要知道数据的来源是哪里了
4. 如今和将来的数据merge以后经过相同的API处理，保证数据的规则相同

其余

仿照优酷视频的小窗效果

拖拽效果

资料

• RxJS 5
• The introduction to Reactive Programming you've been missing
• 流动的数据——使用 RxJS 构造复杂单页应用的数据逻辑