响应式编程和RxJS介绍

响应式编程是一种面向数据流和变化传播的编程范式。
面向变化传播的编程就是看最初的数据是否会随着后续对应变量的变化而变化。比如当变量B的数值改变后，变量C的数值也随之变动。
面向数据流的编程是当监听一系列事件流并对这一系列事件流进行映射、过滤和合并等处理后，再响应整个事件流的回调的过程。例如在ReactiveX编程范式中，数据流被封装在一个叫Observable的对象实例中，通过观察者模式，对数据流进行统一的订阅（Subscribe），并在中间插入像filter()这样的操作函数，从而对Observable所封装的数据流进行过滤处理。

myObservable.filter(fn).subscribe(callback);

响应式编程清楚地表达了动态的异步数据流，而相关的计算模型也自动地将变化的值通过数据流的方式进行了传播。

ReactiveX

ReactiveX一般简写为Rx，它是微软开发并维护的一套工具库集合，用于提供一系列接口规范来处理异步数据流。

Observable

应用中产生的异步数据都需要先包装成Observable对象，Observable对象的作用是把这些异步的数据变换为数据流形式。所以生成的这些Observable对象相当于数据流的源头，后续操作都是围绕着这些被转换的流动数据展开。

Operator

Rx在结合了观察者模式的同时，还结合了函数式编程和迭代器模式的思想。其中，Rx的Operator便是对这两种模式的具体体现。
Operator是Rx中Observable的操作符。在Rx中，每一个Observable对象，或者说数据流，都可以通过某个operator对该Observable对象进行变换、过滤、合并和监听等操作。同时，大多数operator在对该Observable对象处理后返回一个新的Observable对象供下一个operator进行处理。这样方便在各个operator之间通过链式调用的方式编写代码。

//生成一个新的Observable对象
let newObservable=observable.debounceTime(500).take(2);

debunceTime()和take()都是一个Operator，这些Operators通过链式方法的组合，对原有的Observable对象进行操作，最终返回了一个新的Observable对象。
在Rx中，Observable作为观察者模式中的被观察者，需要一个方法来订阅它，而subscribe()便是这样一种方法，订阅Observable对象发出的所有事件。

observable.subscribe(observer);

subscribe()方法会接收一个observer作为参数，来对observable发出的事件进行订阅。每当observable完成并发送（Emit）一个事件时，该事件就会被observer所捕获，进入到observer对应的回调函数中。被subscribe()订阅过的Observable对象并不会返回一个新的Observable对象，因为subscribe()不是一个可以改变原始数据流的函数。相反，subscribe()会返回一个Subscription实例，这个Subscription实例又提供很多操作API，例如具有取消订阅事件功能的unsubscribe()。

其他核心概念

• Observer：对Observable对象发出的每个事件进行响应
• Subscription：Observable对象被订阅后返回的Subscription实例
• Subject：EventEmitter的等价数据结构，可以当作Observable被监听，也可以作为Observer发送新的事件

RxJS

RxJS是Rx在JS层面上的实现，除此之外还有RxJava、Rx.Net等。

创建Observable对象

首先把数据流封装为统一的Observable对象，并对它进行相应的处理。

let button=document.querySelector('button');
Rx.Observable.fromEvent(button,'click') //返回一个Observable对象
    .subscribe(()=>console.log('Clicked'));

通过Observable中的fromEvent静态方法把<button>标签的所有点击事件封装到一个Observable对象中，并转化为数据流的形式，最后通过subscribe()方法对整个点击事件流进行监听。也就是说当按钮被点击时，对应的Observable对象便发出一条相应的消息，这条消息会被subscribe()中的observer回调函数所捕获，从而执行console.log()语句，这样便实现了一个最简单的数据流监听。

使用RxJS处理复杂场景

在实际开发中，会遇到这样的场景：当用户在一个文本输入框进行输入时，需要对用户的输入进行实时的监听，每当用户输入一些新的字符时，会发一个请求到服务器端，来获取一些输入推荐信息展示给用户。但在实现这个功能的时候，需求可能还会有如下的限制条件：

• 不必在每次用户输入的时候都发请求。用户在文本框输入文字时，可能会输入得很快，这时是不需要给用户推荐任何信息的，不让频繁发送网络请求可能会影响性能，因此可当用户输入停顿500ms没有再输入时，才返回推荐信息给用户。
• 要保证请求返回的顺序。在异步请求的情况下，由于服务器返回推荐数据的响应时间会受网络环境等因素影响，有时前端拿到的推荐数据不是最后一次请求的，所以需要保证这些推荐信息的渲染顺序与请求顺序一致。

在RxJS中，这样的数据流操作可以很优雅的实现。

let inputSelector=document.querySelector('input');
Rx.Observable.fromEvent(inputSelector,'keyup')
    .debounceTime(500)
    .switchMap(event=>getRecommend(event.target.value))
    .subscribe(callback);

RxJS和Promise的对比

RxJS的Observable可以通过toPromise()方法把原有弟弟Observable对象转为Promise对象。能用Promise的场景RxJS都实用，RxJS是作为Promise的超集存在。

//Promise实例的创建
let promise=new Promise((resolve,reject)=>{
    //...
    if(/*异步操作成功*/){
        resolve(value);
    }else{
        reject(error);
    }
});
//Observable实例的创建
let Observable=new Observable(observer=>{
    observer.next(value1);
    observe.next(value2);
    observer.error(err);
});

Promise只能针对单一的异步事件进行resolve()操作，而在Observable中，不仅仅能处理一个单一的异步事件（即调用boserver的next()方法），而且能以流的形式响应多个异步事件。

“冷”模式下的Observable

Observable通常可以实现成“热”模式或“冷”模式。在“热”模式下，Observable对象一旦创建，便会开始发送数据。而在“冷”模式下，Observable对象会一直等到自己被订阅，才会开始数据流的发送。在RxJS中，Observable实现的是“冷”模式。

console.log('Observable 的数据发送顺序为：');
let obs=new Observable(observer=>{
    console.log('Observable start');
    observer.next();
});
console.log('start');
obs.subscribe();
结果：
Observable的发送顺序为：
start
Observable start

在RxJS中，obsrver.next()在Observable对象被订阅后才执行。也就是说在RxJS中，Observable对象直到被subscribe()之后，才会进入数据发送的流程中。
除了“冷”模式和“热”模式外，RxJS中还存在另外一种被称作Connectable的模式。这种模式下的Observable对象不管有没有订阅，都不会发送数据，除非ConnectableObservable实例的connect()方法被调用。

console.log('Connectable Observable的数据发送顺序为：')；
let obs=new Observable(observer=>{
    console.log('Observable start');
    observer.complete();
}).publish();
console.log('start');
obs.subscribe();
console.log('after Observable has been subscribed');
obs.connect();
运行结果：
Connectable Observable 的数据发送顺序为：
start
after Observable has been subscribed
Observable start

原来的Observable对象被publish()方法转换为Connectable模式，在Connectable模式下，Observable对象并没有在被subscribe()订阅之后发送数据，而是在被connect()方法调用后才发送，这就是Connectable Observable。

RxJS中的Operator

Operator操作符分类：创建操作符、过滤操作符、组合操作符、工具操作符等。

创建操作符

Observable.fromEvent()、new Observable()和Observable.create()都是创建操作符。

let observable=Rx.Observable.create(observer=>{
    getData(data=>{
        observer.next(data);
        observer.complete();
    })
});
observable.subscribe(data=>{
    //doSomething(data);
});

Observable.create()接受一个工厂函数并返回一个新的Observable对象。这个对象最终被subscribe()方法监听，每当observer.next()方法被调用时，subscribe()中的callback函数便捕获到observer传来的数据并进行相应地处理。这样便实现了对数据流的订阅和监听功能。

变换操作符

有些时候，通过Observable对象获取到的数据需要做一些批量的小调整。比如，数据获取的接口经常会有自己的一套规范去包裹数据。

{
    "err_code":0,
    "data":{"name":"Operators"}
}

只有data字段是实际想要处理的，所以需要对每一个请求做一次变换操作，把原本的数据流变换成需要的数据流，这就需要用到变换操作符。RxJS中最常用的变换操作符是Observable.prototype.map()。

observable.map(response=>{
    return response.data;
}).subscribe(data=>{
    //doSomething(data);
});

当observable拿到响应数据response并传给observer之前，可以通过map操作，来预先对response进行处理，从而让observer得到被加工后的数据格式。

过滤操作符

过滤操作符可用于过滤掉数据流中一些不需要处理的数据。有时候在前端获取数据时，接口会因为各种原因无法返回最终需要的数据，可能由于异常导致返回的数据为空，或只返回一个错误代码以及错误描述告知前端，但并不需要处理这些错误信息，所以就需要过滤这些数据。这时候就需要用到Observable.prototype.filter()来对数据进行过滤。

observable.filter(response=>{
    return !!response.data&&response.status===200;
}).map(response=>{
    return response.data;
}).subscribe(data=>{
    //doSomething(data);
});

结果为false的数据将不会再流向下一个operator。

组合操作符

有时候需要依赖两个甚至更多的接口数据，并在这些接口数据都成功获取后，再进行关联合并。这时候就需要用到组合操作符：Observable.forkJoin()。

let getFirstDataObs=Rx.Observable.create(observer=>{
    observer.next(getFirstData());
    observer.compete();
});
let getSecondDataObs=Rx.Observable.create(observer=>{
    getSecondData(data=>{
        observer.next(data);
        observer.complete();
    });
});
let observable=Rx.Observable.forkJoin(
    getFirstDataObs,getSecondDataObs
);
observable.subscribe(datas=>{
    //data[0]是getFirstDataObs的数据
    //data[1]是getSecondDataObs的数据
    //doSomething(data);
});

Observable.forkJoin()把原本两个互相独立的Observable对象合并为一个新的Observable对象，它会在两个Observable对象的数据都抵达后才开始合并处理。所以odSomething()只会执行一次，此时拿到的datas是包含两个数据流数据的数组。
如果某次数据请求需要依赖前一次请求的结果，也就是说两次请求必须有先后顺序的，这是可以用Observable.prototype.contactMap()。

let getFirstDataObs=Rx.Observable.create(ovserver=>{
    observer.next(getFirstData());
    observer.complete();
});
let createSecondDataObs=function(firstData){
    return Rx.Observable.create(observer=>{
        getSecondData(firstData,secondData=>{
            observer.next(secondData);
            observer.complete();
        });
    });
}
let observable=getFirstDataObs.contactMap(firstData=>{
    return createSecondDataObs(firstData);
}).subscribe(secondData=>{
    doSomethingWithSecondData(secondData);
});

通过Observable.prototype.contactMap()方法，getSecondDataObs()的数据流被紧接在getFirstDataObs()的数据流后，并且最终数据流被subscribe()所捕获。