Rx.js实现原理浅析

前言

上次给大家分享了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就不再赘述了，理解了其实现原理，使用起来就很简单了！