前言
上次给大家分享了cycle.js的内容，这个框架核心模块的代码其实只有一百多行，要理解这个看似复杂的框架，其实最核心的是理解它依赖的异步数据流处理框架——rx.js。今天，给大家分享一下rx.js的实现原理，大家有兴趣可以继续研究它的源码，会让你对异步和响应式编程有更深的理解，进而将rx.js、cycle.js或者仅仅是函数式、响应式编程的思想融入到自己手里的业务中。
为了更好地理解rx.js，需要先谈谈异步编程的实现方案。
异步实现方案
1. 回调函数
makeHttpCall('/items',
items => {
for (itemId of items) {
makeHttpCall(`/items/${itemId}/info`,
itemInfo => {
makeHttpCall(`/items/${itemInfo.pic}`,
img => {
showImg(img);
});
});
}
});
beginUiRendering();
一旦你需要多块数据时你就陷入了流行的”末日金字塔“或者回调地狱。这段代码有很多的问题。 其中之一就是风格。当你在这些嵌套的回调函数中添加越来越多的逻辑，这段代码就会变得很复杂很难理解。因为循环还产生了一个更加细微的问题。for循环是同步的控制流语句，这并不能很好的配合异步调用，因为会有延迟，这可能会产生很奇怪的bug。
2. Promise
makeHttpCall('/items')
.then(itemId => makeHttpCall(`/items/${itemId}/info`))
.then(itemInfo => makeHttpCall(`/items/${itemInfo}.pic}`))
.then(showImg);
链式调用毫无疑问是一个进步，理解这段代码的难度显著下降。然而，尽管Promises在处理这种单值（或单个错误）时非常高效，它有也一些局限性。Promises在处理用户连续输入的数据流时效率怎么样呢？ 这时Promises处理起来也并不高效，因为它没有事件的删除、分配、重试等等的语法定义。
3. async/await
使用async/await，配合Promise可以以同步的方式编写异步代码，是我现在最喜欢也最常用的异步编程方式。比如：上述实现在请求加载完图片后再显示图片的逻辑，也可以这样实现：
const showLoadedImg = async () => {
let getImgInfo = makeHttpCall('/items')
.then(itemId => makeHttpCall(`/items/${itemId}/info`))
.then(itemInfo => makeHttpCall(`/items/${itemInfo}.pic}`));
let loadedImg = await getImgInfo;
showImg(loadedImg);
}
4. generator
generator我用的很少，之前用过一段时间koa（express原班团队搞的基于generator的现代web开发框架），不太喜欢代码在各个模块间跳来跳去的编写思路。整个 Generator 函数就是一个封装的异步任务，或者说是异步任务的容器。异步操作需要暂停的地方，都用 yield 语句注明。调用 Generator 函数，会返回一个内部指针（即遍历器 ）g，调用指针 g 的 next 方法，会移动内部指针（即执行异步任务的第一段），指向第一个遇到的 yield 语句。下面贴一段koa中间件级联的代码，大家感受一下：
var koa = require('koa');
var app = koa();
// x-response-time
app.use(function *(next){
var start = new Date;
yield next;
var ms = new Date - start;
this.set('X-Response-Time', ms + 'ms');
});
// logger
app.use(function *(next){
var start = new Date;
yield next;
var ms = new Date - start;
console.log('%s %s - %s', this.method, this.url, ms);
});
// response
app.use(function *(){
this.body = 'Hello World';
});
app.listen(3000);
上面的例子在页面中返回 "Hello World"，然而当请求开始时，请求先经过 x-response-time 和 logging 中间件，并记录中间件执行起始时间。 然后将控制权交给 reponse 中间件。当中间件运行到 yield next 时，函数挂起并将控制前交给下一个中间件。当没有中间件执行 yield next 时，程序栈会逆序唤起被挂起的中间件来执行接下来的代码。
5. rxjs
RxJS是一个解决异步问题的JS开发库.它起源于 Reactive Extensions 项目，它带来了观察者模式和函数式编程的相结合的最佳实践。 观察者模式是一个被实践证明的模式，基于生产者（事件的创建者）和消费者（事件的监听者）的逻辑分离关系。
况且函数式编程方式的引入，如说明性编程，不可变数据结构，链式方法调用会使你极大的简化代码量。（和回调代码方式说再见吧）
如果你熟悉函数式编程，请把RxJS理解为异步化的Underscore.js。
RxJS 引入了一个重要的数据类型——流（stream）。
rxjs实现原理
观察者模式
观察者模式在 Web 中最常见的应该是 DOM 事件的监听和触发。
订阅：通过 addEventListener 订阅 document.body 的 click 事件。
发布：当 body 节点被点击时，body 节点便会向订阅者发布这个消息。
document.body.addEventListener('click', function listener(e) {
console.log(e);
},false);
document.body.click(); // 模拟用户点击
迭代器模式
迭代器模式可以用 JavaScript 提供了 Iterable Protocol 可迭代协议来表示。Iterable Protocol 不是具体的变量类型，而是一种可实现协议。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}
元素对象中：value 表示返回值，done 表示是否已经到达最后。
遍历迭代器可以使用下面做法。
var iterable = [1, 2];
var iterator = iterable[Symbol.iterator]();
while(true) {
let result;
try {
result = iterator.next(); // <= 获取下一个值
} catch (err) {
handleError(err); // <= 错误处理
}
if (result.done) {
handleCompleted(); // <= 无更多值（已完成）
break;
}
doSomething(result.value);
}
主要对应三种情况：
获取下一个值
调用 next 可以将元素一个个地返回，这样就支持了返回多次值。
无更多值(已完成)
当无更多值时，next 返回元素中 done 为 true。
错误处理
当 next 方法执行时报错，则会抛出 error 事件，所以可以用 try catch 包裹 next 方法处理可能出现的错误。
RxJS 的观察者 + 迭代器模式
RxJS 中含有两个基本概念：Observables 与 Observer。Observables 作为被观察者，是一个值或事件的流集合；而 Observer 则作为观察者，根据 Observables 进行处理。
Observables 与 Observer 之间的订阅发布关系(观察者模式) 如下：
订阅：Observer 通过 Observable 提供的 subscribe() 方法订阅 Observable。
发布：Observable 通过回调 next 方法向 Observer 发布事件。
下面为 Observable 与 Observer 的伪代码：
// Observer
var Observer = {
next(value) {
alert(`收到${value}`);
}
};
// Observable
function Observable (Observer) {
setTimeout(()=>{
Observer.next('A');
},1000)
}
// subscribe
Observable(Observer);
上面实际也是观察者模式的表现，那么迭代器模式在 RxJS 中如何体现呢？
在 RxJS 中，Observer 除了有 next 方法来接收 Observable 的事件外，还可以提供了另外的两个方法：error() 和 complete()，与迭代器模式一一对应。
var Observer = {
next(value) { /* 处理值*/ },
error(error) { /* 处理异常 */ },
complete() { /* 处理已完成态 */ }
};
结合迭代器 Iterator 进行理解：
next()
Observer 提供一个 next 方法来接收 Observable 流，是一种 push 形式；而 Iterator 是通过调用 iterator.next() 来拿到值，是一种 pull 的形式。
complete()
当不再有新的值发出时，将触发 Observer 的 complete 方法；而在 Iterator 中，则需要在 next 的返回结果中，当返回元素 done 为 true 时，则表示 complete。
error()
当在处理事件中出现异常报错时，Observer 提供 error 方法来接收错误进行统一处理；Iterator 则需要进行 try catch 包裹来处理可能出现的错误。
下面是 Observable 与 Observer 实现观察者 + 迭代器模式的伪代码，数据的逐渐传递传递与影响其实就是流的表现。
// Observer
var Observer = {
next(value) {
alert(`收到${value}`);
},
error(error) {
alert(`收到${value}`);
},
complete() {
alert("complete");
},
};
// Observable
function Observable (Observer) {
[1,2,3].map(item=>{
Observer.next(item);
});
Observer.complete();
// Observer.error("error message");
}
// subscribe
Observable(Observer);
RxJS 基础实现
有了上面的概念及伪代码，那么在 RxJS 中是怎么创建 Observable 与 Observer 的呢?
创建 Observable
RxJS 提供 create 的方法来自定义创建一个 Observable，可以使用 next 来发出流。
var Observable = Rx.Observable.create(observer => {
observer.next(2);
observer.complete();
return () => console.log('disposed');
});
创建 Observer
Observer 可以声明 next、err、complete 方法来处理流的不同状态。
var Observer = Rx.Observer.create(
x => console.log('Next:', x),
err => console.log('Error:', err),
() => console.log('Completed')
);
最后将 Observable 与 Observer 通过 subscribe 订阅结合起来。
var subscription = Observable.subscribe(Observer);
关于rx.js的使用和API就不再赘述了，理解了其实现原理，使用起来就很简单了！