RxJava 小试

1、基本概念

官方wiki这样介绍： https://github.com/ReactiveX/RxJava/wiki

RxJava is a Java VM implementation of ReactiveX (Reactive Extensions): a library for composing asynchronous and event-based programs by using observable sequences.

翻译：RxJava是ReactiveX的JVM实现版本，一个可以使用可观察对象序列来编写异步和基于事件程序的library。

RxJava是观察者模式的扩展实现，可观察对象（Observable）在数据产生或者发生改变时，主动通知观察者（Observer）或者Subscriber（订阅者），观察者在接收到通知后，就可以根据通知作出不同类型的响应。比如，网络请求成功或者请求失败等。

2、优势

减少回调嵌套

多数据源请求处理

强大的操作符

自由线程切换

3、实践

如果项目中有一个页面包含多个模块，每个模块又分归不同业务线负责（大公司常有的事，业务划分比较细，聚合数据、聚合页面非常多），那么数据就可能需要从多个不同的api请求。在实现下拉刷新功能时，就需要同时监控多个请求的结果。

传统的解决方案——计数

每个模块开始刷新的时候，计数加1，刷新完成时计数减1，从而达到监听所有请求结果的目的。这对每个模块来说，需要明确调用刷新开始或者结束的时机，不小心调用错了或者忘了调用，就可能刷新永远也结束不了。相对来说，维护成本较大。

RxJava解决方案

subscription = Observable.merge(getObservableList(false)).subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
... // 请求成功，刷新结束
subscription.unsubscribe();
}

@Override
public void onError(Throwable e) {
... // 请求失败，刷新结束
subscription.unsubscribe();
showToast("网络连接失败,请检查网络");
}

@Override
public void onNext(String str) {

}
});

RxJava强大的操作符merge就可以解决问题。merge可以接收一组Observable对象序列作为参数，合并所有Observable的输出，对订阅者来说就相当于单个Observable对象，因此很轻松就能够监听所有请求的请求结果。

关于merge需要注意的是：

（1）merge不会保证Observable序列的结果输出顺序，想要保证顺序可以使用concat操作符。

（2）任何一个Observable对象调用OnError事件，会立刻输出结果给合并后的Observable对象，后面未输出的Observable不再监听了。如果想要继续监听后续Observable，合并所有Observable结果再输出，可以使用mergeDelayError。

4、内存泄漏

RxJava虽然方便，但是如果使用不当，也容易造成内存泄漏。Subscriber通常是通过匿名内部类方式创建的，持有外部类引用。如果在Activity中创建，并且没有及时取消订阅，会导致Activity无法及时释放，从而造成内存泄漏。

解决方案：及时取消订阅

@Override
public void onDestroy() {
super.onDestroy();
if(subscription != null && !subscription.isUnsubscribed()){
subscription.unsubscribe();
subscription = null;
}
}