Android框架式编程之RxJava

RxJava是一个很好的开源库，将链式编程和异步很好的结合在一起。在介绍RxJava之前我们先介绍一下观察者模式，这样我们能更好的学习和理解RxJava。

简单介绍一下，A和B两个，A是被观察者，B是观察者，B对A进行观察，B并不是需要时刻盯着A，而是A如果发生了变化，会主动通知B，B会对应做一些变化。

 

如上图，假设A是连载小说，B是读者，读者订阅了连载小说，当小说出现了新的连载的时候，会推送给读者。读者不用时刻盯着小说连载，而小说有了新的连载会主动推送给读者。这就是观察者模式。而RxJava正是基于观察者模式开发的。

一、RxJava的基础使用

目前主流的RxJava使用的版本为RxJava2系列，在使用之前我们需要先引入RxJava2.0相关的类库：

 implementation 'io.reactivex.rxjava2:rxjava:2.2.13'
 implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'

使用方式：

1). 创建被观察者：

//  创建被观察者
Observable novelObservable = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
        observableEmitter.onNext("新章节1001");
        observableEmitter.onNext("新章节1002");
        observableEmitter.onNext("新章节1003");
        observableEmitter.onComplete();
    }
});

Observable中文意思就是被观察者，通过create方法生成对象，里面放的参数ObservableOnSubscribe<T>，可以理解为一个计划表，泛型T是要操作对象的类型，重写subscribe方法，里面写具体的计划，本文的例子就是推送新章节1001、新章节1002和新章节1003，在subscribe中的ObservableEmitter<String>对象的Emitter是发射器的意思。

ObservableEmitter有三种发射的方法，分别是void onNext(T value)、void onError(Throwable error)、void onComplete()，onNext方法可以无限调用，Observer（观察者）所有的都能接收到，onError和onComplete是互斥的，Observer（观察者）只能接收到一个，OnComplete可以重复调用，但是Observer（观察者）只会接收一次，而onError不可以重复调用，第二次调用就会报异常。

2). 创建观察者

// 创建观察者
Observer<String> reader = new Observer<String>() {
    @Override
    public void onSubscribe(Disposable disposable) {
        doLog("subscribe success");
    }

    @Override
    public void onNext(String s) {
        doLog("收到最新章节 ===》" + s);
    }

    @Override
    public void onError(Throwable throwable) {

    }

    @Override
    public void onComplete() {
        doLog("接受结束！");
    }
};
novelObservable.subscribe(reader);

通过new创建接口，并实现其内部的方法，看方法其实就应该差不多知道干嘛的，onNext、onError、onComplete都是跟被观察者发射的方法一一对应的，这里就相当于接收了。onSubscribe（Disposable d）里面的Disposable对象要说一下，Disposable英文意思是可随意使用的，这里就相当于读者和连载小说的订阅关系，如果读者不想再订阅该小说了，可以调用 mDisposable.dispose()取消订阅，此时连载小说更新的时候就不会再推送给读者了。

3). 将订阅者和被订阅者建立关系

novelObservable.subscribe(reader);

运行代码，查看输出结果如下：

 总结：以上就是RxJava2最简单的用法，总结一下就是：

1).  创建被订阅者

2).  创建订阅者

3).  建立订阅关系

二、RxJava异步链式编程

RxJava 是支持异步的，核心就是 Scheduler （调度器），RxJava用 Scheduler 来控制线程。

当我们没有设置 Scheduler 的时候，RxJava遵循在哪个线程启动，就在哪个线程运行的原则。

例：一般情况下我们在使用RxJava的时候都是后台执行，前台调用的，本着这个原则，我们需要调用observeOn(AndroidSchedulers.mainThread())，subscribeOn(Schedulers.io())。

这里observeOn是事件回调的线程，AndroidSchedulers.mainThread()一看就知道是主线程；subscribeOn是事件执行的线程，Schedulers.io()是子线程，这里也可以用Schedulers.newThread()，只不过io线程可以重用空闲的线程，因此多数情况下 io() 比 newThread() 更有效率。

下面我们改造一下第一节的代码，改造后如下：

Log.e("111", "主Thread Hash = " + Thread.currentThread().hashCode());
Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
        Log.e("111", "被订阅Thread Hash = " + Thread.currentThread().hashCode());
        observableEmitter.onNext("新章节1001");
        observableEmitter.onNext("新章节1002");
        observableEmitter.onNext("新章节1003");
        observableEmitter.onComplete();
    }
}).subscribeOn(Schedulers.io())  // 执行在io线程
        .observeOn(AndroidSchedulers.mainThread())  //  回调在主线程
        .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable disposable) {
                doLog("subscribe success");
            }

            @Override
            public void onNext(String s) {
                Log.e("111", "订阅Thread Hash = " + Thread.currentThread().hashCode());
                doLog("收到最新章节 ===》" + s);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {
                doLog("接受结束！");
            }
        });

输出结果：

E/111: 主Thread Hash = 72810649
E/111: 日志 -----> subscribe success
E/111: 被订阅Thread Hash = 68064815
E/111: 订阅Thread Hash = 72810649
E/111: 日志 -----> 收到最新章节 ===》新章节1001
    订阅Thread Hash = 72810649
    日志 -----> 收到最新章节 ===》新章节1002
    订阅Thread Hash = 72810649
    日志 -----> 收到最新章节 ===》新章节1003
    日志 -----> 接受结束！

三、RxJava的应用场景

1). RxJava + Retrofit 封装网络请求框架

2). 基于RxJava的开源类库 Rxpermissions、RxBinding、RxBus

四、参考资料

1). RxJava2.0——从放弃到入门：https://www.jianshu.com/p/cd3557b1a474

2). 打造终极MVP+Retrofit2+okhttp3+Rxjava2网络请求：https://www.jianshu.com/p/df4eee78085c