RxJava的使用总结

RxJava简介

RxJava的github地址https://github.com/ReactiveX/RxJava
RxJava的英文wikihttps://github.com/ReactiveX/RxJava/wiki

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

简单来说，RxJava是利用可观察序列和操作符来编写异步和基于事件的程序。RxJava源于ReactiveX库，简称Rx库，是一个使用可观察数据流进行异步编程的编程接口，ReactiveX结合了观察者模式、迭代器模式和函数式编程的精华。目前为止，Rx已经覆盖了几乎全部流行的编程语言。ReactiveX文档中文翻译
Android对于异步线程有着严格的管理机制，UI的更新只能在主线程中进行，而耗时操作及网络请求则是在主线程中严禁执行。谷歌官方给出了Handler，AsyncTask等解决方法，但实际运用下来，效果并不是非常好，尤其当异步操作比较复杂时，开发者会十分头痛于繁多的逻辑代码，以及时不时出现的内存泄露。而RxJava提供了一种十分优雅的处理异步线程的方式，以数据流的方式将复杂的runnable转化为清晰简便的数据流。

1. 函数式风格：对可观察数据流使用无副作用的输入输出函数，避免了程序里错综复杂的状态。数据流的形式十分利于功能扩展。
2. 简化代码：Rx的操作符通通常可以将复杂的难题简化为很少的几行代码。配合java的lambda可以极大的减少代码量以及实现代码的简洁性。
3. 异步错误处理：传统的try/catch没办法处理异步计算，Rx提供了合适的错误处理机制。
4. 轻松使用并发：Rx的Observables和Schedulers让开发者可以摆脱底层的线程同步和各种并发问题。

RxJava的入门

RxJava的兴起也有一年多的时间了，RxJava已经成为了众多Android项目必备的第三方依赖库，配合Retrofit的使用已经替代了绝大多数包括volley在内的网络请求库。但RxJava的学习曲线比较陡峭，这里推荐一篇非常的RxJava入门级文章给 Android 开发者的 RxJava 详解。

RxJava的使用场景

RxJava本质是一个方便实现异步操作的库，所以说RxJava最适用的场景就是需要频繁切换线程的操作，所以说它是十分适用于Android开发的。

1.与Retrofit的开发结合

Retrofit的github地址https://github.com/square/retrofit

Retrofit 是 Square 的一个非常著名的网络请求库。他即提供传统的Callback接口回调模式的API，又提供了基于RxJava库Observable回调的�API。前者是volley及自定义网络框架中经常使用的回调方式，而后者则是Retrofit为配合RxJava的特殊定义的回调方式，也是Retrofit库现在被广泛采用的原因之一（还包括十分方便的接口定义，网络配置和数据转换）。

Retrofit本身就是一个非常牛逼的第三方开源库，可以极大�简化网络请求。现在已经发布了Retrofit2.0版本，这是Retrofit两种不同的接口定义方式：

interface GitHubService {
    @GET("/repos/{owner}/{repo}/contributors")
    Call<List<Contributor>> repoContributors(
        @Path("owner") String owner, 
        @Path("repo") String repo);
    );
    
    @GET("/repos/{owner}/{repo}/contributors")
    Observable<List<Contributor>> repoContributors2(
        @Path("owner") String owner, 
        @Path("repo") String repo);
    ); 
}

第一种方法想要异步实现网络请求

Call<List<Contributor>> call = 
    gitHubService.repoContributors("square","retrofit");

call.enqueue(new Callback<List<Contributor>>() { 
    @Override void onResponse(/* ... */) { 
        // ... 
    } 

    @Override void onFailure(Throwable t) { 
        // ... 
    }
});

需要在异步请求时实现一个回调接口Callback，在对Response做复杂逻辑处理时往往会产生许多杂乱无章的代码，对于团队协作及后期的维护改进是十分致命的。
而二种方法返回的是RxJava的Observable对象，我们不用在考虑是否 new Thread()去处理一些耗时操作，比如说数据库的添加和比对。因为RxJava有完善的线程切换机制，可以通过线程调度器Scheduler实现线程的切换。

• Schedulers.immediate() 直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。
• Schedulers.newThread() 总是启用新线程，并在新线程执行操作.
• Schedulers.io() I/O 操作（读写文件、读写数据库、网络信息交互等）所使用的 Scheduler。行为模式和newThread()差不多，区别在于 io() 的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中，可以避免创建不必要的线程。
• Schedulers.computation() 计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算，即不会被 I/O 等操作限制性能的操作，例如图形的计算。这个 Scheduler 使用的固定的线程池，大小为 CPU 核数。不要把 I/O 操作放在 computation() 中，否则 I/O 操作的等待时间会浪费 CPU。
• 还有RxAndroid里面专门提供了AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。

Observable<List<Contributor>> observable = 
    gitHubService.repoContributors2("square","retrofit");

observable.subscribeOn(Schedulers.io()) //io线程网络请求
    .doOnNext(list -> {/.../}) //耗时操作，数据库读写
    .observeOn(AndroidSchedulers.mainThread()) //主线程更新UI
    .subscribe(list -> {/.../});

代码简洁明了，逻辑一目了然。
当然你可以轻松的做业务逻辑的扩展，相对list的非空校验，只需要在subscribe订阅前添加filter(�list -> list != null)。
如果更深入，你也可以加上统一的错误处理，实现全局的Token过期自动刷新。RxJava基于其各种各样的操作符可以帮助你实现绝大多数逻辑，而且更加简洁，易于理解。

2.与RxBinding的开发结合

RxBinding的github地址https://github.com/JakeWharton/RxBinding

RxBinding的作者就是大名鼎鼎的JakeWharton，这位大神是RxAndroid、Retorfit及okhttp的主要贡献者，自己则开发了 RxBinding 及Butterknife。就是把发布-订阅模式用在了Android控件的点击，文本变化上。通过RxBinding把点击监听转换成Observable之后，就有了对它进行扩展的可能

这里提供两个使用案例：

1. 按钮需要防抖功能:

RxView.clicks(findViewById(R.id.btn_throttle))
    .throttleFirst(1, TimeUnit.SECONDS)//一秒钟之内只取一个点击事件，防抖操作
    .subscribe(aVoid -> {
        Toast.makeText(this,"�click",Toast.LENGTH_SHORT).show();
    });

2. 搜索栏监听输入变化自动请求搜索结果：

RxTextView.textChanges(findViewById(R.id.edit_search))
    .debounce(300, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread())//每300毫秒采样1次
    .distinctUntilChanged()//关键词去重
    .doOnNext(s -> showProgress())//展示加载动画
    .observeOn(Schedulers.io())//切换线程，网络请求
    .switchMap(Observable::just)//当Observable发射一个新的数据项时，如果�上一次订阅还未完成，就取消旧订阅数据和停止监视那个数据项产生的Observable,开始监视新的数据项
    .flatMap(s -> {//通过Retrofit请求网络接口})
    .retry((integer, throwable) -> throwable instanceof InterruptedIOException)//中断自动重试
    .subscribeOn(AndroidSchedulers.mainThread())//切换到主线程准备更新UI
    .filter(response ->{//必要的数据校验})
    .subscribe(response ->{//搜索结果的展示});

另外，在这里推荐一个使Subscription与组件的生命周期同步的框架RxLifecycle，当然你也可以在Activity或者Fragment的onDestroy()中取消订阅防止内存溢出，但这种方法明显没有RxLifecycle灵活。

3.RxBus的使用原理

RxBus是使用RxJava实现类似于otto、eventbus等开源库提供的功能，它的代码量极少却能完成绝大多数后者的使用场景。
RxBus的实现主要依赖于RxJava的Subject类，对于Subject的定义:

Subject在ReactiveX是作为observer和observerable的一个bridge或者proxy。因为它是一个观察者，所以它可以订阅一个或多个可观察对象，同时因为他是一个可观测对象，所以它可以传递和释放它观测到的数据对象，并且能释放新的对象。

Subject根据用途的不同设计了四种子类，分别为AsyncSubject、BehaviorSubject、ReplaySubject和PublishSubject。

1. AsyncSubject
   AsyncSubject仅释放Observable释放的最后一个数据，并且仅在Observable完成之后。然而如果当Observable因为异常而终止，AsyncSubject将不会释放任何数据，但是会向Observer传递一个异常通知。
2. BehaviorSubject
   当Observer订阅了一个BehaviorSubject，它一开始就会释放Observable最近释放的一个数据对象，当还没有任何数据释放时，它则是一个默认值。接下来就会释放Observable释放的所有数据。如果Observable因异常终止，BehaviorSubject将不会向后续的Observer释放数据，但是会向Observer传递一个异常通知。
3. ReplaySubject
   不管Observer何时订阅ReplaySubject，ReplaySubject会向所有Observer释放Observable释放过的数据。
4. PublishSubject
   PublishSubject仅会向Observer释放在订阅之后Observable释放的数据。

通常而言，我们的RxBus将会使用的是PublishSubject，即我们只需要接受我们订阅了之后的数据，但可以根据不同的业务场景采用不同的Subject来实现我们所需要的订阅逻辑。下面提供一种基于PublishSubject的RxBus实现。

public class RxBus {
    private static volatile RxBus defaultInstance;
    private final Subject bus;    // PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者
    
    public RxBus() {
        bus = new SerializedSubject<>(PublishSubject.create());
    }
        
    /** 
      * 获取RxBus单例
      */
    public static RxBus getDefault() {        
        RxBus rxBus = defaultInstance;        
        if (defaultInstance == null) {            
            synchronized (RxBus.class) {                
                rxBus = defaultInstance;                
                if (defaultInstance == null) {                    
                    rxBus = new RxBus();                    
                    defaultInstance = rxBus;                
                }            
            }        
        }        
        return rxBus;    
    }    

    // 提供了一个新的事件    
    public void post (Object o) {        
        bus.onNext(o);    
    }    
  
    // 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者    
    public <T> Observable<T> toObserverable (Class<T> eventType) {        
        return bus.ofType(eventType);    
    }}

我们只需要在Activity或Fragment创建时订阅，就可以收到事件了。同时，对订阅的取消是十分重要的

4.Android耗时操作的RxJava运用

Android中又许多耗时操作场景，比如数据库的读写，大容量的图片处理保存以及文件的压缩/解压工作。但由于Android的机制，主线程无法处理长耗时的操作，所以这类操作往往需要在异步线程中进行，完毕后在主线程更新。这些都可以使用RxJava处理。
我们可以在Github上面轻易地搜出许许多多的RxJava处理Android耗时操作的库，包括用RxJava处理和操作高斯模糊效果的RxBlur，用于下载操作的rxloader，用于数据库操作的RxCupboard等等。而这些通常是对现有的AndroidAPI进行针对RxJava的封装，使得这些操作具有更好的拓展性，而不是在回调接口中书写大段冗余复杂的代码，你也完全书写定义自己的RxJava逻辑。