阅读 27540

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()取消订阅

解决的问题

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


核心概念

概述关系

图片

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多路推送当前的值
// 创建一个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处理,保证数据的规则相同

其他


资料



关注下面的标签,发现更多相似文章
评论