浏览新版,请访问 RxJS Observable
在介绍 Observable 以前,咱们要先了解两个设计模式:javascript
这两个模式是 Observable 的基础,下面咱们先来介绍一下 Observer Pattern。html
观察者模式是 软件设计模式的一种。在此种模式中,一个目标对象管理全部相依于它的观察者对象,而且在它自己的状态改变时主动发出通知。这一般透过呼叫各观察者所提供的方法来实现。此种模式一般被用来实时事件处理系统。 — 维基百科
观察者模式又叫发布订阅模式(Publish/Subscribe),它定义了一种一对多的关系,让多个观察者对象同时监听某一个主题对象,这个主题对象的状态发生变化时就会通知全部的观察者对象,使得它们可以自动更新本身。前端
咱们可使用平常生活中,期刊订阅的例子来形象地解释一下上面的概念。期刊订阅包含两个主要的角色:期刊出版方和订阅者,他们之间的关系以下:java
在观察者模式中也有两个主要角色:Subject (主题) 和 Observer (观察者) 。它们分别对应例子中的期刊出版方和订阅者。接下来咱们来看张图,从而加深对上面概念的理解。react
观察者模式的优势:git
观察者模式的缺点:es6
在前端领域,观察者模式被普遍地使用。最多见的例子就是为 DOM 对象添加事件监听,具体示例以下:github
<button id="btn">确认</button> function clickHandler(event) { console.log('用户已点击确认按钮!'); } document.getElementById("btn").addEventListener('click', clickHandler);
上面代码中,咱们经过 addEventListener API 监听 button 对象上的点击事件,当用户点击按钮时,会自动执行咱们的 clickHandler
函数。算法
Subject 类定义:typescript
class Subject { constructor() { this.observerCollection = []; } registerObserver(observer) { this.observerCollection.push(observer); } unregisterObserver(observer) { let index = this.observerCollection.indexOf(observer); if(index >= 0) this.observerCollection.splice(index, 1); } notifyObservers() { this.observerCollection.forEach((observer)=>observer.notify()); } }
Observer 类定义:
class Observer { constructor(name) { this.name = name; } notify() { console.log(`${this.name} has been notified.`); } }
使用示例:
let subject = new Subject(); // 建立主题对象 let observer1 = new Observer('semlinker'); // 建立观察者A - 'semlinker' let observer2 = new Observer('lolo'); // 建立观察者B - 'lolo' subject.registerObserver(observer1); // 注册观察者A subject.registerObserver(observer2); // 注册观察者B subject.notifyObservers(); // 通知观察者 subject.unregisterObserver(observer1); // 移除观察者A subject.notifyObservers(); // 验证是否成功移除
以上代码成功运行后控制台的输出结果:
semlinker has been notified. # 输出一次 2(unknown) lolo has been notified. # 输出两次
须要注意的是,在观察者模式中,一般状况下调用注册观察者后,会返回一个函数,用于移除监听,有兴趣的读者,能够本身尝试一下。(备注:在 Angular 1.x 中调用 $scope.$on() 方法后,就会返回一个函数,用于移除监听)
迭代器(Iterator)模式,又叫作游标(Cursor)模式。它提供一种方法顺序访问一个聚合对象中的各个元素,而又不须要暴露该对象的内部表示。迭代器模式能够把迭代的过程从业务逻辑中分离出来,在使用迭代器模式以后,即便不关心对象的内部构造,也能够按顺序访问其中的每一个元素。
迭代器模式的优势:
迭代器模式的缺点:
在 ECMAScript 中 Iterator 最先实际上是要采用相似 Python 的 Iterator 规范,就是 Iterator 在没有元素以后,执行next
会直接抛出错误;但后来通过一段时间讨论后,决定采更 functional 的作法,改为在取得最后一个元素以后执行next
永远都回传{ done: true, value: undefined }
一个迭代器对象 ,知道如何每次访问集合中的一项, 并记录它的当前在序列中所在的位置。在 JavaScript 中迭代器是一个对象,它提供了一个 next() 方法,返回序列中的下一项。这个方法返回包含 done
和 value
两个属性的对象。对象的取值以下:
{ done: false, value: elementValue }
{ done: true, value: undefined }
详细信息能够参考 - 可迭代协议和迭代器协议
接下来咱们来建立一个 makeIterator 函数,该函数的参数类型是数组,当调用该函数后,返回一个包含 next() 方法的 Iterator 对象, 其中 next() 方法是用来获取容器对象中下一个元素。具体示例以下:
function makeIterator(array){ var nextIndex = 0; return { next: function(){ return nextIndex < array.length ? {value: array[nextIndex++], done: false} : {done: true}; } } }
一旦初始化, next() 方法能够用来依次访问可迭代对象中的元素:
var it = makeIterator(['yo', 'ya']); console.log(it.next().value); // 'yo' console.log(it.next().value); // 'ya' console.log(it.next().done); // true
在 ES 6 中咱们能够经过 Symbol.iterator
来建立可迭代对象的内部迭代器,具体示例以下:
let arr = ['a', 'b', 'c']; let iter = arr[Symbol.iterator]();
调用 next()
方法来获取数组中的元素:
> iter.next() { value: 'a', done: false } > iter.next() { value: 'b', done: false } > iter.next() { value: 'c', done: false } > iter.next() { value: undefined, done: true }
ES 6 中可迭代的对象:
RxJS 是基于观察者模式和迭代器模式以函数式编程思惟来实现的。RxJS 中含有两个基本概念:Observables 与 Observer。Observables 做为被观察者,是一个值或事件的流集合;而 Observer 则做为观察者,根据 Observables 进行处理。
Observables 与 Observer 之间的订阅发布关系(观察者模式) 以下:
Proposal Observable Implementations
若是你想真正了解 Observable,最好的方式就是本身写一个。其实 Observable 就是一个函数,它接受一个 Observer
做为参数而后返回另外一个函数。
它的基本特征:
Observer
对象 (包含 next、error、complete 方法的对象) 做为参数unsubscribe
函数,用于取消订阅它的做用:
做为生产者与观察者之间的桥梁,并返回一种方法来解除生产者与观察者之间的联系,其中观察者用于处理时间序列上数据流。接下来咱们来看一下 Observable 的基础实现:
DataSource - 数据源
class DataSource { constructor() { let i = 0; this._id = setInterval(() => this.emit(i++), 200); // 建立定时器 } emit(n) { const limit = 10; // 设置数据上限值 if (this.ondata) { this.ondata(n); } if (n === limit) { if (this.oncomplete) { this.oncomplete(); } this.destroy(); } } destroy() { // 清除定时器 clearInterval(this._id); } }
myObservable
function myObservable(observer) { let datasource = new DataSource(); // 建立数据源 datasource.ondata = (e) => observer.next(e); // 处理数据流 datasource.onerror = (err) => observer.error(err); // 处理异常 datasource.oncomplete = () => observer.complete(); // 处理数据流终止 return () => { // 返回一个函数用于,销毁数据源 datasource.destroy(); }; }
使用示例:
const unsub = myObservable({ next(x) { console.log(x); }, error(err) { console.error(err); }, complete() { console.log('done')} }); /** * 移除注释,能够测试取消订阅 */ // setTimeout(unsub, 500);
具体运行结果,能够查看线上示例。
上面的示例中,咱们使用一个包含了 next、error、complete 方法的普通 JavaScript 对象来定义观察者。一个普通的 JavaScript 对象只是一个开始,在 RxJS 5 里面,为开发者提供了一些保障机制,来保证一个更安全的观察者。如下是一些比较重要的原则:
Observer
对象能够不实现全部规定的方法 (next、error、complete 方法)complete
或者 error
触发以后再调用 next
方法是没用的unsubscribe
方法后,任何方法都不能再被调用了complete
和 error
触发后,unsubscribe
也会自动调用next
、complete
和error
出现异常时,unsubscribe
也会自动调用以保证资源不会浪费next
、complete
和error
是可选的。按需处理便可,没必要所有处理为了完成上述目标,咱们得把传入的匿名 Observer
对象封装在一个 SafeObserver
里以提供上述保障。SafeObserver 的具体实现以下:
class SafeObserver { constructor(destination) { this.destination = destination; } next(value) { // 还没有取消订阅,且包含next方法 if (!this.isUnsubscribed && this.destination.next) { try { this.destination.next(value); } catch (err) { // 出现异常时,取消订阅释放资源,再抛出异常 this.unsubscribe(); throw err; } } } error(err) { // 还没有取消订阅,且包含error方法 if (!this.isUnsubscribed && this.destination.error) { try { this.destination.error(err); } catch (e2) { // 出现异常时,取消订阅释放资源,再抛出异常 this.unsubscribe(); throw e2; } this.unsubscribe(); } } complete() { // 还没有取消订阅,且包含complete方法 if (!this.isUnsubscribed && this.destination.complete) { try { this.destination.complete(); } catch (err) { // 出现异常时,取消订阅释放资源,再抛出异常 this.unsubscribe(); throw err; } this.unsubscribe(); } } unsubscribe() { // 用于取消订阅 this.isUnsubscribed = true; if (this.unsub) { this.unsub(); } } }
myObservable - 使用 SafeObserver
function myObservable(observer) { const safeObserver = new SafeObserver(observer); // 建立SafeObserver对象 const datasource = new DataSource(); // 建立数据源 datasource.ondata = (e) => safeObserver.next(e); datasource.onerror = (err) => safeObserver.error(err); datasource.oncomplete = () => safeObserver.complete(); safeObserver.unsub = () => { // 为SafeObserver对象添加unsub方法 datasource.destroy(); }; // 绑定this上下文,并返回unsubscribe方法 return safeObserver.unsubscribe.bind(safeObserver); }
使用示例:
const unsub = myObservable({ next(x) { console.log(x); }, error(err) { console.error(err); }, complete() { console.log('done')} });
具体运行结果,能够查看线上示例。
Operator 是一个函数,它接收一个 Observable 对象,而后返回一个新的 Observable 对象。当咱们订阅新返回的 Observable 对象时,它内部会自动订阅前一个 Observable 对象。接下来咱们来实现经常使用的 map 操做符:
Observable 实现:
class Observable { constructor(_subscribe) { this._subscribe = _subscribe; } subscribe(observer) { const safeObserver = new SafeObserver(observer); safeObserver.unsub = this._subscribe(safeObserver); return safeObserver.unsubscribe.bind(safeObserver); } }
map 操做符实现:
function map(source, project) { return new Observable((observer) => { const mapObserver = { next: (x) => observer.next(project(x)), error: (err) => observer.error(err), complete: () => observer.complete() }; return source.subscribe(mapObserver); }); }
具体运行结果,能够查看线上示例。
若是把 Operator
都写成如上那种独立的函数,咱们链式代码会逐渐变丑:
map(map(myObservable, (x) => x + 1), (x) => x + 2);
对于上面的代码,想象一下有 五、6 个嵌套着的 Operator
,再加上更多、更复杂的参数,基本上就无法儿看了。
你也能够试下 Texas Toland 提议的简单版管道实现,合并压缩一个数组的Operator
并生成一个最终的Observable
,不过这意味着要写更复杂的 Operator
,上代码:JSBin。其实写完后你会发现,代码也不怎么漂亮:
pipe(myObservable, map(x => x + 1), map(x => x + 2));
理想状况下,咱们想将代码用更天然的方式链起来:
myObservable.map(x => x + 1).map(x => x + 2);
幸运的是,咱们已经有了这样一个 Observable
类,咱们能够基于 prototype 在不增长复杂度的状况下支持多 Operators
的链式结构,下面咱们采用prototype方式再次实现一下 Observable
:
Observable.prototype.map = function (project) { return new Observable((observer) => { const mapObserver = { next: (x) => observer.next(project(x)), error: (err) => observer.error(err), complete: () => observer.complete() }; return this.subscribe(mapObserver); }); };
如今咱们终于有了一个还不错的实现。这样实现还有其余好处,例如:能够写子类继承 Observable
类,而后在子类中重写某些内容以优化程序。
接下来咱们来总结一下该部分的内容:Observable 就是函数,它接受 Observer 做为参数,又返回一个函数。若是你也写了一个函数,接收一个 Observer 做为参数,又返回一个函数,那么,它是异步的、仍是同步的 ?其实都不是,它就只是一个函数。任何函数的行为都依赖于它的具体实现,因此当你处理一个 Observable 时,就把它当成一个普通函数,里面没有什么黑魔法。当你要构建 Operator 链时,你须要作的其实就是生成一个函数将一堆 Observers 连接在一块儿,而后让真正的数据依次穿过它们。
var observable = Rx.Observable .create(function(observer) { observer.next('Semlinker'); // RxJS 4.x 之前的版本用 onNext observer.next('Lolo'); }); // 订阅这个 Observable observable.subscribe(function(value) { console.log(value); });
以上代码运行后,控制台会依次输出 'Semlinker' 和 'Lolo' 两个字符串。
须要注意的是,不少人认为 RxJS 中的全部操做都是异步的,但其实这个观念是错的。RxJS 的核心特性是它的异步处理能力,但它也是能够用来处理同步的行为。具体示例以下:
var observable = Rx.Observable .create(function(observer) { observer.next('Semlinker'); // RxJS 4.x 之前的版本用 onNext observer.next('Lolo'); }); console.log('start'); observable.subscribe(function(value) { console.log(value); }); console.log('end');
以上代码运行后,控制台的输出结果:
start Semlinker Lolo end
固然咱们也能够用它处理异步行为:
var observable = Rx.Observable .create(function(observer) { observer.next('Semlinker'); // RxJS 4.x 之前的版本用 onNext observer.next('Lolo'); setTimeout(() => { observer.next('RxJS Observable'); }, 300); }) console.log('start'); observable.subscribe(function(value) { console.log(value); }); console.log('end');
以上代码运行后,控制台的输出结果:
start Semlinker Lolo end RxJS Observable
从以上例子中,咱们能够得出一个结论 - Observable 能够应用于同步和异步的场合。
RxJS 中提供了不少操做符,用于建立 Observable 对象,经常使用的操做符以下:
上面的例子中,咱们已经使用过了 create 操做符,接下来咱们来看一下其它的操做符:
var source = Rx.Observable.of('Semlinker', 'Lolo'); source.subscribe({ next: function(value) { console.log(value); }, complete: function() { console.log('complete!'); }, error: function(error) { console.log(error); } });
以上代码运行后,控制台的输出结果:
Semlinker Lolo complete!
var arr = [1, 2, 3]; var source = Rx.Observable.from(arr); // 也支持字符串,如 "Angular 2 修仙之路" source.subscribe({ next: function(value) { console.log(value); }, complete: function() { console.log('complete!'); }, error: function(error) { console.log(error); } });
以上代码运行后,控制台的输出结果:
1 2 3 complete!
Rx.Observable.fromEvent(document.querySelector('button'), 'click');
var source = Rx.Observable .fromPromise(new Promise((resolve, reject) => { setTimeout(() => { resolve('Hello RxJS!'); },3000) })); source.subscribe({ next: function(value) { console.log(value); }, complete: function() { console.log('complete!'); }, error: function(error) { console.log(error); } });
以上代码运行后,控制台的输出结果:
Hello RxJS! complete!
var source = Rx.Observable.empty(); source.subscribe({ next: function(value) { console.log(value); }, complete: function() { console.log('complete!'); }, error: function(error) { console.log(error); } });
以上代码运行后,控制台的输出结果:
complete!
empty 操做符返回一个空的 Observable 对象,若是咱们订阅该对象,它会当即返回 complete 信息。
var source = Rx.Observable.never(); source.subscribe({ next: function(value) { console.log(value); }, complete: function() { console.log('complete!'); }, error: function(error) { console.log(error); } });
never 操做符会返回一个无穷的 Observable,当咱们订阅它后,什么事情都不会发生,它是一个一直存在却什么都不作的 Observable 对象。
var source = Rx.Observable.throw('Oop!'); source.subscribe({ next: function(value) { console.log(value); }, complete: function() { console.log('complete!'); }, error: function(error) { console.log('Throw Error: ' + error); } });
以上代码运行后,控制台的输出结果:
Throw Error: Oop!
throw 操做如,只作一件事就是抛出异常。
var source = Rx.Observable.interval(1000); source.subscribe({ next: function(value) { console.log(value); }, complete: function() { console.log('complete!'); }, error: function(error) { console.log('Throw Error: ' + error); } });
以上代码运行后,控制台的输出结果:
0 1 2 ...
interval 操做符支持一个数值类型的参数,用于表示定时的间隔。上面代码表示每隔 1s,会输出一个递增的值,初始值从 0 开始。
var source = Rx.Observable.timer(1000, 5000); source.subscribe({ next: function(value) { console.log(value); }, complete: function() { console.log('complete!'); }, error: function(error) { console.log('Throw Error: ' + error); } });
以上代码运行后,控制台的输出结果:
0 # 1s后 1 # 5s后 2 # 5s后 ...
timer 操做符支持两个参数,第一个参数用于设定发送第一个值需等待的时间,第二个参数表示第一次发送后,发送其它值的间隔时间。此外,timer 操做符也能够只传递一个参数,具体以下:
var source = Rx.Observable.timer(1000); source.subscribe({ next: function(value) { console.log(value); }, complete: function() { console.log('complete!'); }, error: function(error) { console.log('Throw Error: ' + error); } });
以上代码运行后,控制台的输出结果:
0 complete!
有些时候对于一些 Observable 对象 (如经过 interval、timer 操做符建立的对象),当咱们不须要的时候,要释放相关的资源,以免资源浪费。针对这种状况,咱们能够调用 Subscription
对象的 unsubscribe
方法来释放资源。具体示例以下:
var source = Rx.Observable.timer(1000, 1000); // 取得subscription对象 var subscription = source.subscribe({ next: function(value) { console.log(value); }, complete: function() { console.log('complete!'); }, error: function(error) { console.log('Throw Error: ' + error); } }); setTimeout(() => { subscription.unsubscribe(); }, 5000);
Observer (观察者) 是一个包含三个方法的对象,每当 Observable 触发事件时,便会自动调用观察者的对应方法。
interface Observer<T> { closed?: boolean; // 标识是否已经取消对Observable对象的订阅 next: (value: T) => void; error: (err: any) => void; complete: () => void; }
Observer 中的三个方法的做用:
接下来咱们来看个具体示例:
var observable = Rx.Observable .create(function(observer) { observer.next('Semlinker'); observer.next('Lolo'); observer.complete(); observer.next('not work'); }); // 建立一个观察者 var observer = { next: function(value) { console.log(value); }, error: function(error) { console.log(error); }, complete: function() { console.log('complete'); } } // 订阅已建立的observable对象 observable.subscribe(observer);
以上代码运行后,控制台的输出结果:
Semlinker Lolo complete
上面的例子中,咱们能够看出,complete 方法执行后,next 就会失效,因此不会输出 not work
。
另外观察者能够不用同时包含 next、complete、error 三种方法,它能够只包含一个 next 方法,具体以下:
var observer = { next: function(value) { console.log(value); } };
有时候 Observable 多是一个无限的序列,例如 click 事件,对于这种场景,complete 方法就永远不会被调用。
咱们也能够在调用 Observable 对象的 subscribe
方法时,依次传入 next、error、complete 三个函数,来建立观察者:
observable.subscribe( value => { console.log(value); }, error => { console.log('Error: ', error); }, () => { console.log('complete'); } );
Pull 和 Push 是数据生产者和数据的消费者两种不一样的交流方式。
在 "拉" 体系中,数据的消费者决定什么时候从数据生产者那里获取数据,而生产者自身并不会意识到何时数据将会被发送给消费者。
每个 JavaScript 函数都是一个 "拉" 体系,函数是数据的生产者,调用函数的代码经过 ''拉出" 一个单一的返回值来消费该数据。
const add = (a, b) => a + b; let sum = add(3, 4);
ES6介绍了 iterator迭代器 和 Generator生成器 — 另外一种 "拉" 体系,调用 iterator.next()
的代码是消费者,可从中拉取多个值。
在 "推" 体系中,数据的生产者决定什么时候发送数据给消费者,消费者不会在接收数据以前意识到它将要接收这个数据。
Promise(承诺) 是当今 JS 中最多见的 "推" 体系,一个Promise (数据的生产者)发送一个 resolved value (成功状态的值)来执行一个回调(数据消费者),可是不一样于函数的地方的是:Promise 决定着什么时候数据才被推送至这个回调函数。
RxJS 引入了 Observables (可观察对象),一个全新的 "推" 体系。一个可观察对象是一个产生多值的生产者,当产生新数据的时候,会主动 "推送给" Observer (观察者)。
生产者 | 消费者 | |
---|---|---|
pull拉 | 被请求的时候产生数据 | 决定什么时候请求数据 |
push推 | 按本身的节奏生产数据 | 对接收的数据进行处理 |
接下来咱们来看张图,从而加深对上面概念的理解:
Observable(可观察对象)是基于推送(Push)运行时执行(lazy)的多值集合。
MagicQ | 单值 | 多值 |
---|---|---|
拉取(Pull) | 函数 | 遍历器 |
推送(Push) | Promise | Observable |
Promise
Observable
全部的 Observable 对象必定会等到订阅后,才开始执行,若是没有订阅就不会执行。
var source = Rx.Observable.from([1,2,3,4,5]); var example = source.map(x => x + 1);
上面的示例中,由于 example 对象还未被订阅,因此不会进行运算。这跟数组不同,具体以下:
var source = [1,2,3,4,5]; var example = source.map(x => x + 1);
以上代码运行后,example 中就包含已运算后的值。
数组中的操做符如:filter、map 每次都会完整执行并返回一个新的数组,才会继续下一步运算。具体示例以下:
var source = [1,2,3,4,5]; var example = source .filter(x => x % 2 === 0) // [2, 4] .map(x => x + 1) // [3, 5]
关于数组中的 map、filter 的详细信息,能够参考 - RxJS Functional Programming
为了更好地理解数组操做符的运算过程,咱们能够参考下图:
虽然 Observable 运算符每次都会返回一个新的 Observable 对象,但每一个元素都是渐进式获取的,且每一个元素都会通过操做符链的运算后才输出,而不会像数组那样,每一个阶段都得完整运算。具体示例以下:
var source = Rx.Observable.from([1,2,3,4,5]); var example = source .filter(x => x % 2 === 0) .map(x => x + 1) example.subscribe(console.log);
以上代码的执行过程以下:
为了更好地理解 Observable 操做符的运算过程,咱们能够参考下图: