RxJS: 简单入门

Introduction to RxJS

1. 前言

1.1 什么是RxJS？

RxJS是ReactiveX编程理念的JavaScript版本。ReactiveX来自微软，它是一种针对异步数据流的编程。简单来讲，它将一切数据，包括HTTP请求，DOM事件或者普通数据等包装成流的形式，而后用强大丰富的操做符对流进行处理，使你能以同步编程的方式处理异步数据，并组合不一样的操做符来轻松优雅的实现你所须要的功能。

1.2 RxJS可用于生产吗？

ReactiveX由微软于2012年开源，目前各语言库由ReactiveX组织维护。RxJS在GitHub上已有8782个star，目前最新版本为5.5.2，并持续开发维护中，其中官方测试用例共计2699个。

1.3 RxJS对项目代码的影响？

RxJS中的流以Observable对象呈现，获取数据须要订阅Observable，形式以下：

const ob = http$.getSomeList(); //getSomeList()返回某个由`Observable`包装后的http请求
ob.subscribe((data) => console.log(data));
//在变量末尾加$表示Observable类型的对象。

以上与Promise相似：

const promise = http.getSomeList(); // 返回由`Promise`包装的http请求
promise.then((data) => console.log(data));

实际上Observable能够认为是增强版的Promise，它们之间是能够经过RxJS的API互相转换的：

const ob = Observable.fromPromise(somePromise); // Promise转为Observable
const promise = someObservable.toPromise(); // Observable转为Promise

所以能够在Promise方案的项目中安全使用RxJS，并可以随时升级到完整的RxJS方案。

1.4 RxJS会增长多少体积？

RxJS(v5)整个库压缩后约为140KB，因为其模块化可扩展的设计，所以仅需导入所用到的类与操做符便可。导入RxJS经常使用类与操做符后，打包后的体积约增长30-60KB，具体取决于导入的数量。

不要用 import { Observable } from 'rxjs'这种方式导入，这会导入整个 rxjs库，按需导入的方式以下：
import { Observable } from 'rxjs/Observable' //导入类
import 'rxjs/add/operator/map' // 导入实例操做符
import 'rxjs/add/observable/forkJoin' // 导入类操做符

2. RxJS快速入门

2.1 初级核心概念

• Observable
• Observer
• Operator

Observable被称为可观察序列，简单来讲数据就在Observable中流动，你可使用各类operator对流进行处理，例如：

const ob = Observable.interval(1000);
ob.take(3).map(n => n * 2).filter(n => n > 2);

第一步代码咱们经过类方法interval建立了一个Observable序列，ob做为源会每隔1000ms发射一个递增的数据，即0 -> 1 -> 2。第二步咱们使用操做符对流进行处理，take(3)表示只取源发射的前3个数据，取完第三个后关闭源的发射；map表示将流中的数据进行映射处理，这里咱们将数据翻倍；filter表示过滤掉出符合条件的数据，根据上一步map的结果，只有第二和第三个数据会留下来。

上面咱们已经使用同步编程建立好了一个流的处理过程，但此时ob做为源并不会马上发射数据，若是咱们在map中打印n是不会获得任何输出的，由于ob做为Observable序列必须被“订阅”才可以触发上述过程，也就是subscribe（发布/订阅模式）。

const ob = Observable.interval(1000);
ob.take(3).map(n => n * 2).filter(n => n > 0).subscribe(n => console.log(n));

结果：

2 //第2秒
4 //第3秒

上面代码中咱们给subscribe传入了一个函数，这实际上是一种简写，subscribe完整的函数签名以下：

ob.subscribe({
    next: d => console.log(d),
    error: err => console.error(err),
    complete: () => console.log('end of the stream')
})

直接给subscribe传入一个函数会被当作是next函数。这个完整的包含3个函数的对象被称为observer（观察者），表示的是对序列结果的处理方式。next表示数据正常流动，没有出现异常；error表示流中出错，多是运行出错，http报错等等；complete表示流结束，再也不发射新的数据。在一个流的生命周期中，error和complete只会触发其中一个，能够有多个next（表示屡次发射数据），直到complete或者error。

observer.next能够认为是Promise中then的第一个参数，observer.error对应第二个参数或者Promise的catch。

RxJS一样提供了catch操做符，err流入catch后，catch必须返回一个新的Observable。被catch后的错误流将不会进入observer的error函数，除非其返回的新observable出错。

Observable.of(1).map(n => n.undefinedMethod()).catch(err => {
    // 此到处理catch以前发生的错误
    return Observable.of(0); // 返回一个新的序列，该序列成为新的流。
});

2.2 建立可观察序列

建立一个序列有不少种方式，咱们仅列举经常使用的几种：

Observable.of(...args)

Observable.of()能够将普通JavaScript数据转为可观察序列，点我测试。

Observable.fromPromise(promise)

将Promise转化为Observable，点我测试。

Observable.fromEvent(elment, eventName)

从DOM事件建立序列，例如Observable.fromEvent($input, 'click')，点我测试。

Observable.ajax(url | AjaxRequest)

发送http请求，AjaxRequest参考这里

Observable.create(subscribe)

这个属于万能的建立方法，通常用于只提供了回调函数的某些功能或者库，在你用这个方法以前先想一想能不能用RxJS上的类方法来建立你所须要的序列，点我测试。

2.3 合并序列

合并序列也属于建立序列的一种，例若有这样的需求：进入某个页面后拿到了一个列表，而后须要对列表每一项发出一个http请求来获取对应的详细信息，这里咱们把每一个http请求做为一个序列，而后咱们但愿合并它们。
合并有不少种方式，例如N个请求按顺序串行发出（前一个结束再发下一个）；N个请求同时发出而且要求所有到达后合并为数组，触发一次回调；N个请求同时发出，对于每个到达就触发一次回调。
若是不用RxJS，咱们会比较难处理这么多情形，不只实现麻烦，维护更麻烦，下面是使用RxJS对上述需求的解决方案：

const ob1 = Observable.ajax('api/detail/1');
const ob2 = Observable.ajax('api/detail/2');
...
const obs = [ob1, ob2...];
// 分别建立对应的HTTP请求。

1. N个请求按顺序串行发出（前一个结束再发下一个）

Observable.concat(...obs).subscribe(detail => console.log('每一个请求都触发回调'));

1. N个请求同时并行发出，对于每个到达就触发一次回调

Observable.merge(...obs).subscribe(detail => console.log('每一个请求都触发回调'));

1. N个请求同时发出而且要求所有到达后合并为数组，触发一次回调

Observable.forkJoin(...obs).subscribe(detailArray => console.log('触发一次回调'));

3. 使用RxJS实现搜索功能

搜索是前端开发中很常见的功能，通常是监听<input />的keyup事件，而后将内容发送到后台，并展现后台返回的数据。

<input id="text"></input>
<script>
    var text = document.querySelector('#text');
    text.addEventListener('keyup', (e) =>{
        var searchText = e.target.value;
        // 发送输入内容到后台
        $.ajax({
            url: `/search/${searchText}`,
            success: data => {
              // 拿到后台返回数据，并展现搜索结果
              render(data);
            }
        });
    });
</script>

上面代码实现咱们要的功能，但存在两个较大的问题：

• 多余的请求

当想搜索“爱迪生”时，输入框可能会存在三种状况，“爱”、“爱迪”、“爱迪生”。而这三种状况将会发起 3 次请求，存在 2 次多余的请求。

• 已无用的请求仍然执行

一开始搜了“爱迪生”，而后立刻改搜索“达尔文”。结果后台返回了“爱迪生”的搜索结果，执行渲染逻辑后结果框展现了“爱迪生”的结果，而不是当前正在搜索的“达尔文”，这是不正确的。

减小多余请求数，能够用 setTimeout 函数节流的方式来处理，核心代码以下：

<input id="text"></input>
<script>
    var text = document.querySelector('#text'),
        timer = null;
    text.addEventListener('keyup', (e) =>{
        // 在 250 毫秒内进行其余输入，则清除上一个定时器
        clearTimeout(timer);
        // 定时器，在 250 毫秒后触发
        timer = setTimeout(() => {
            console.log('发起请求..');
        },250)
    })
</script>

已无用的请求仍然执行 的解决方式，能够在发起请求前声明一个当前搜索的状态变量，后台将搜索的内容及结果一块儿返回，前端判断返回数据与当前搜索是否一致，一致才走到渲染逻辑。最终代码为：

<input id="text"></input>
<script>
    var text = document.querySelector('#text'),
        timer = null,
        currentSearch = '';

    text.addEventListener('keyup', (e) =>{
        clearTimeout(timer)
        timer = setTimeout(() => {
            // 声明一个当前所搜的状态变量
            currentSearch ＝ '书'; 

            var searchText = e.target.value;
            $.ajax({
                url: `/search/${searchText}`,
                success: data => {
                    // 判断后台返回的标志与咱们存的当前搜索变量是否一致
                    if (data.search === currentSearch) {
                        // 渲染展现
                        render(data);
                    } else {
                        // ..
                    }
                }           
            });
        },250)
    })
</script>

上面代码基本知足需求，但代码开始显得乱糟糟。咱们来使用RxJS实现上面代码功能，以下:

var text = document.querySelector('#text');
var inputStream = Rx.Observable.fromEvent(text, 'keyup') //为dom元素绑定'keyup'事件
                    .debounceTime(250) // 防抖动
                    .pluck('target', 'value') // 取值
                    .switchMap(url => Http.get(url)) // 将当前输入流替换为http请求
                    .subscribe(data => render(data)); // 接收数据

RxJS能简化你的代码，它将与流有关的内部状态封装在流中，而不须要在流外定义各类变量来以一种上帝视角控制流程。Rx的编程方式使你的业务逻辑流程清晰，易维护，并显著减小出bug的几率。

我的总结的经常使用操做符：

类操做符（一般为合并序列或从已有数据建立序列）
合并 forkJoin, merge, concat
建立 of, from, fromPromise, fromEvent, ajax, throw
实例操做符（对流中的数据进行处理或者控制流程）
map, filter,switchMap, toPromise, catch, take, takeUntil, timeout, debounceTime, distinctUntilChanged, pluck。
对于这些操做符的使用再也不详细描述，请参阅网上资料。

中文官网 http://cn.rx.js.org/
附上我的翻译的一些文章

• RxJS：冷热模式的比较
• RxJS: map, flatMap和flatMapLatest的区别
• RxJS: 详解forkJoin, zip, combineLatest之间的区别

最后打个广告，蚂蚁金服大安全风控+团队招人，前端后台都要，简历发送至huitong.zht@antfin.com。

参考文章：构建流式应用：RxJS 详解