前言

2017年，刚毕业的我在项目组里打下手，组里其他人决定在新项目里使用RxJava，那时网络库用的是他们自己再次封装了的noHttp，加上了RxJava后，以前简单的网络请求多写了很多代码，这是我对RxJava的第一感受。
如今的项目基本上都是MVP+Retrofit2+Rxjava2的框架，但往往只是把Rxjava2和Retrofit2搭配进行网络请求而已，我也不例外。也曾看过好几次RxJava2的入门教程、操作符总结等，过段时间就忘了，而且不知在什么样的情景下适用恰当，所以写下这篇文章，打算带着问题出发，重新整理一下。
问题：
1、RxJava是什么，它优势是什么？
2、RxJava基础知识。
3、RxJava2操作符。
4、背压策略。
5、具体的使用场景。

1、RxJava是什么，它优势是什么？

RxJava官方介绍——一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库。

Rxjava本质就是异步，它的优势是简洁。简洁是指相比于AsyncTask 、Handler、runOnUiThread等异步操作，RxJava的逻辑更简洁。一方面它是链式的，逻辑代码直观，二是它提供了很多强大的操作符。

2、RxJava基础知识

RxJava的基础知识，首先要知道它是用了可拓展的观察者模式，然后再了解这些概念，Observable、Observer、subscribe、处理事件、Scheduler。

1）什么是观察者模式？
观察者（Observer）不时刻盯着被观察者（Observable），而是通过注册(Register)或者称为订阅(Subscribe)的方式，告诉被观察者我需要你的某某状态，你要在它变化的时候通知我。
举例1：
A（观察者）和B（被观察者），A不需要每过 1s 就检查一次 B 的状态，而是B状态发生改变时，去通知A。
举例2：
View的点击事件，View（被观察者）->OnClickLisetener（观察者）->setOnClickListener（订阅）->OnClick（事件）。
举例3：
小偷在偷钱时告诉警察它要偷钱。小偷（被观察者）、警察（观察者）、小偷偷钱是某种状态、警察把小偷抓起来(事件)。

2）RxJava的观察者模式
Observable（被观察者）、Observer（观察者）、 subscribe（订阅）、事件。
看代码清楚明了，ObservableEmitter是事件发射器，在onSubscribe方法里可得到Disposable，可用于取消订阅。在观察者的方法体里有4个事件，解释看注释。如果发送了onError() 或 onComplete()事件，就取消了这个订阅。
一般用于网络请求时，都会写一个继承Observer的类来做处理，比如在onSubscribe()里做加载框显示，在onComplete()做某些共性判断，在onError()做统一失败处理等。所以Observer的4个方法肯定是要清楚的。

/**
* 打印结果：  只会打印1和2，3不打印
*/
Observable.create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            //ObservableEmitter类，事件发射器，作用是定义需要发送的事件 & 向观察者发送事件
            emitter.onNext(1);
            emitter.onNext(2);
            emitter.onComplete();
            emitter.onNext(3);
        }
    }).subscribe(new Observer<Integer>() {

        private Disposable mDisposable;  //用于取消订阅

        /**
         * 订阅成功就执行的方法
         * @param d  可切断操作
         */
        @Override
        public void onSubscribe(Disposable d) {
            mDisposable = d;
        }

        /**
         * 接收事件
         * @param integer
         */
        @Override
        public void onNext(Integer integer) {
            KLog.d(TAG, "onNext: " + integer + "\n");
        }

        /**
         * 失败，不再接收事件
         * @param e
         */
        @Override
        public void onError(Throwable e) {

        }

        /**
         * 完成，不再接收事件
         */
        @Override
        public void onComplete() {
        }
    });

3）线程控制Scheduler
Scheduler是线程调度，RxJava通过它来指定每一段代码运行在什么样的线程。例如网络请求用io流线程，更新UI就用mainThread，最常用也是这两个。
Schedulers.newThread()：新线程。
Schedulers.io()：在io()内部有线程池，可以使用空闲的线程，多数情况下比newThread()效率高。
Schedulers.computation()：计算线程（没用过）。
AndroidSchedulers.mainThread()：主线程。

4）取消订阅
RxJava2的取消订阅用Disposable.cancel()，如果订阅事件过多，要写管理类来管理。如果不取消订阅，就会造成内存泄漏，网上有RxLifecycle第三方库，可以很方便做取消订阅，它绑定了Activity or Fragment的生命周期。

3、RxJava2操作符

这里重新整理了一下当初学习的操作符，原本打算整理全部的操作符的，然后发现实在是太多了，就放弃了，下面的操作符绝大多数都来自于Android Rxjava：这是一篇 清晰 & 易懂的Rxjava 入门教程，想详细学习的可看这篇文章。网上也有很多文章有对操作符做简述，这篇文章比较完整Android拾萃 - RxJava2操作符汇总。最好结合代码学习，下载地址在本文最下面。

1）创建操作符
包括完整&快速创建被观察者，定时操作，周期性操作，数据/集合遍历

2）变换操作符

3）组合/合并转换符

4）功能性操作符

5）过滤操作符

6）条件/布尔操作符

4.背压策略。

当在异步订阅中，通过Observable发射、处理、响应数据流时，如果事件产生的速度远远快于事件消费的速度，这些没来得及处理的数据就会越积越多，这些数据不会丢失，也不会被垃圾回收机制回收，而是存放在一个异步缓存池中，缓存池的数据一直得不到处理，最终会导致OOM等异常。这就是响应式编程中的背压问题。总结一下就是： 事件产生的速度大于事件消费的速度，数据堆积，最终造成OOM等异常。
RxJava2把对背压问题的处理逻辑从Observable中抽取出来产生了新的可观察对象Flowable，它是在Observable基础上做了优化，所以Observable能做的，它都能做，但是加了背压支持和其他的逻辑处理，它的效率比Observable慢得多，所以在需要用到背压的时候再用Flowable，其他时候还是正常使用Observable。网上找了两张图，可以很直观地看出RxJava1和2对背压的改动。

RxJava1.png

RxJava2.png

简单举个例子：

Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {

                //异步订阅时，代表的是 异步缓存池中可放入数据的数量，一开始是128，当产生10个事件而没有消费时，此时这个值是128-10=118。
                KLog.d(TAG, "异步缓存池中可放入数据的数量 = " + emitter.requested());

                // 一共发送4个事件
                KLog.d(TAG, "发送事件 1");
                emitter.onNext(1);
                KLog.d(TAG, "发送事件 2");
                emitter.onNext(2);
                KLog.d(TAG, "发送事件 3");
                emitter.onNext(3);
                KLog.d(TAG, "发送事件 4");
                emitter.onNext(4);
                KLog.d(TAG, "发送事件 onComplete()");
                emitter.onComplete();

                KLog.d(TAG, "异步缓存池中可放入数据的数量 = " + emitter.requested());

//                //模拟缓存超过128
//                for (int i = 0;i< 129; i++) {
//                    Log.d(TAG, "发送了事件" + i);
//                    emitter.onNext(i);
//                }
//                emitter.onComplete();
//
            }
        }, BackpressureStrategy.ERROR)  //缓存区超过128，直接抛出异常
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        // 在异步订阅情况下，一定要调用request，否则下流不接收事件
                        // 只接收多少个事件
                        s.request(3);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        KLog.d(TAG, "接收到了事件" + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        KLog.d(TAG, "onError：", t);
                    }

                    @Override
                    public void onComplete() {
                        KLog.d(TAG, "onComplete");
                    }
                });

可以看到Flowable和Subscriber的使用方式和之前的Observable和Observer极其类似，不同点如下：
1、create方法中多了一个BackpressureStrategy类型的参数。
2、Flowable发射数据时，使用FlowableEmitter，而Observable用的是ObservableEmitter。
3、Subscriber中，方法onSubscribe回调的参数不是Disposable而是Subscription。
下面对这三点进行说明。

BackpressureStrategy
缓存策略，也既是当缓存区存满、上流仍然继续发送下事件时，该如何处理的策略。默认缓存区大小是128（与Flowable的buffersize大小有关）。具体有哪些策略如下：
BackpressureStrategy.ERROR： 直接抛异常。
BackpressureStrategy.MISSING： 友好提示：缓存区满了。
BackpressureStrategy.BUFFER： 将缓存区设为无限大。
BackpressureStrategy.DROP： 超过缓存区大小128的事件丢弃。
BackpressureStrategy.LATEST： 只保存最后事件，超过缓存区大小128的事件丢弃。

Subscription
Subscription比Disposable多了一个方法request()。它的作用是告诉上游，下游需要多少数据，如果不设置request默认是0。比如设置为3，那么超过范围之外的数据就不接收了。如果多次调用request()，会累加。

FlowableEmitter
它比ObservableEmitter多了一个方法requested()，这个方法返回的是异步缓存池中可放入数据的数量，比如一开始是128，当产生10个事件而没有消费时，此时这个值是128-10=118。

Tips：不管Subscription.request（xx）设置了什么值，FlowableEmitter都会发送事件的，发送了不接收的就放入缓存里。背压在异步订阅中才有用，如果是同步订阅，是不会有缓存池的。

5.具体的使用场景

实际项目中，我只用来做过验证码倒计时、网络请求，RxBus。虽然了解了一些操作符，但网上的教程讲解这些操作符的时候都是以简单的例子来讲解的，具体的应用场景还是比较玄乎。

1）RxBus
EventBus是一个基于发布/订阅的事件总线，它简化了组件之间的通信操作。而这些RxBus都能做，所以用RxBus替换EventBus，在RxJava1就已经提出来了，如果自己的事件发送的要求不高，可以自己封装一个RxBus使用。17年的时候找的一篇文章 Android 用RxJava模拟一个EventBus ———RxBus，我在这基础上稍微改动了一下，但是没有做粘性事件。
原理简单来说就是找个容器装所有的观察者，当有某个事件产生时，找到所有需要这个事件的观察者，向它们发送事件。这个是否发送事件的依据，其实就是看观察者要什么样的class，比如订阅String类型的，那么Integer类型的事件就不会发给它。

两个新知识：
1）CompositeDisposable
一个disposable的容器，可以容纳多个disposable
2）Subject
Subject可以同时代表 Observer 和 Observable，允许从数据源中多次发送结果给多个观察者。

public class RxBus {

    /**
     * CompositeDisposable定义：
     * 一个disposable的容器，可以容纳多个disposable，添加和去除的复杂度为O(1)
     *
     * 此处使用目的：
     * 是为了一个订阅者能够对应多个Disposable，在需要的时候调用 Disposable 的 dispose()取消订阅。
     *
     * 举个例子：
     * XXActivity，订阅了事件A和事件B，关闭时要取消订阅。
     * 那么只需在mSubscriptionMap里找到key为XXActivity的Value（CompositeDisposable），再取出Disposable
     * 取消订阅即可。
     *
     * 如果不用的话，就要为每一个Activity写一个容器去保存它的订阅的事件了。
     * */
    private HashMap<String, CompositeDisposable> mSubscriptionMap;
    private static volatile RxBus mRxBus;
    private final Subject<Object> mSubject;

    public static RxBus getIntanceBus(){
        if (mRxBus==null){
            synchronized (RxBus.class){
                if(mRxBus==null){
                    mRxBus = new RxBus();
                }
            }
        }
        return mRxBus;
    }

    
    /**
     * 正常订阅可用PublishSubject、黏性事件可用ReplaySubject。
     */
    private RxBus(){
        mSubject = PublishSubject.create().toSerialized();
    }

    /**
     * 一个默认的订阅方法
     * @param <T>
     * @param type   过滤，只返回特定数据类型的数据
     * @param next   next()事件,正常接收的事件
     * @param error  error()事件，错误接收的事件
     * @return
     */
    public <T> Disposable doSubscribe(Class<T> type, Consumer<T> next, Consumer<Throwable> error){
        return getObservable(type)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(next,error);
    }

    /**
     * 返回指定类型的带背压的Flowable实例
     * 这里选定背压模式是 BackpressureStrategy.BUFFER，缓存区无限大
     * @param <T>
     * @param type  过滤，只返回特定数据类型的数据
     * @return
     */
    public <T> Flowable<T> getObservable(Class<T> type){
        return mSubject.toFlowable(BackpressureStrategy.BUFFER).ofType(type);
    }

    /**
     * 发送事件
     * @param o 事件
     */
    public void post(Object o){
        mSubject.onNext(o);
    }


    /**
     * 判断是否已有观察者订阅
     *
     * @return
     */
    public boolean hasObservers() {
        return mSubject.hasObservers();
    }

    /**
     * 保存订阅后的disposable，取消订阅的时候要用
     * @param o       订阅的目标
     * @param disposable
     */
    public void addSubscription(Object o, Disposable disposable) {
        if (mSubscriptionMap == null) {
            mSubscriptionMap = new HashMap<>();
        }
        //这里key值取订阅目标的实体名称(com.xx.xx.xxx)，不是底层类检称
        String key = o.getClass().getName();
        //disposable 放到对应的 CompositeDisposable 里
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).add(disposable);
        } else {
            CompositeDisposable disposables = new CompositeDisposable();
            disposables.add(disposable);
            mSubscriptionMap.put(key, disposables);
        }
    }

    /**
     * 取消订阅
     * @param o
     */
    public void unSubscribe(Object o) {
        if (mSubscriptionMap == null) {
            return;
        }
        String key = o.getClass().getName();
        if (!mSubscriptionMap.containsKey(key)){
            return;
        }
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).dispose();
        }
        mSubscriptionMap.remove(key);
    }

    /********************************* 为RxEventBean 封装  **********************************/
    /**
     * 使用EventBus时，为了方便查找，一般都会封装 EventBean(int code,Object content)
     * 但是所有的订阅者都是订阅的这个类型，所以要自己做判断做类型转换。
     *
     * 这里写死事件是RxEventBean，且错误处理一般不处理,只用处理onNext即可。
     * @param context
     * @param action
     * */
    public void register(Context context, Consumer<RxEventBean> action) {
        Disposable disposable = RxBus.getIntanceBus().doSubscribe(RxEventBean.class, action,
                throwable -> KLog.e("RxEventBean onError()", throwable.toString()));
        RxBus.getIntanceBus().addSubscription(context,disposable);
    }


    /**
     * 发送RxEventBean 事件
     * @param code     code，用于判断
     * @param content  内容，接收后做类型转换
     */
    public void post(int code, Object content){
        RxEventBean<Object> event = new RxEventBean<>();
        event.code = code;
        event.content = content;
        post(event);
    }
}

2）验证码倒计时
其实关键代码就是用intervalRange()操作符。

public class RxCodeHelper {

    private Context mContext;
    private TextView codeBt;
    private Disposable mDisposable;

    /**
     * 构造函数
     * @param mContext 上下文
     * @param codeBt   验证码的button
     */
    public RxCodeHelper(Context mContext, TextView codeBt) {
        this.mContext = mContext;
        this.codeBt = codeBt;
    }

    /**
     * 开启倒计时
     */
    public void start(){
        codeBt.setEnabled(false);
        //从0开始，走60个数，延迟是0s，周期为1次。
        Observable.intervalRange(0,60,0,1, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        mDisposable = d;
                    }

                    @Override
                    public void onNext(Long aLong) {
                        codeBt.setText((60 - aLong) + "s后可重发");
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        //倒计时完毕置为可点击状态
                        codeBt.setEnabled(true);
                        codeBt.setText("获取验证码");
                    }
                });
    }

    /**
     * 请求失败时，重置状态
     * 取消倒计时的订阅事件
     */
    public void reset(){
        if(mDisposable!=null){
            mDisposable.dispose();
        }
        codeBt.setEnabled(true);
        codeBt.setText("获取验证码");
    }

    /**
     * 界面销毁时，取消倒计时订阅事件
     */
    public void stop(){
        if(mDisposable!=null){
            mDisposable.dispose();
        }
    }

}

3）RxLifecycle
目的：解决RxJava的内存泄漏问题。
用法：
bindUntilEvent(@NonNull ActivityEvent event)——绑定指定的生命周期，在指定生命周期时取消订阅。
bindToLifecycle() ——绑定生命周期，取消订阅策略可看源码。以Activity为例，可看RxLifecycleAndroid 下的 ACTIVITY_LIFECYCLE， 实际上在onCreate()订阅的，在onDestroy()取消订阅；在onResume()订阅的，在onPause()取消订阅。

知识点：
compose()：
将一种类型的Observable转换成另一种类型的Observable，保证调用的链式结构。
LifecycleTransformer：
LifecycleTransformer实现了各种Transformer接口，能够将一个Observable/Flowable/Single/Completable/Maybe对象转换成另一个 Observable/Flowable/Single/Completable/Maybe对象。正好配合上文的compose操作符，使用在链式调用中。

举例在网络请求时绑定生命周期，mvp模式。
Activity继承了RxFragmentActivity后，BaseView接口里新增两个方法，其他View层接口继承它。

public interface BaseView {
//为了让 IView 可以调用 RxLifeCycle的生命周期绑定
<T> LifecycleTransformer<T> bindToLifecycle();
<T> LifecycleTransformer<T> bindUntilEvent(@NonNull ActivityEvent event);
}

在P层使用，此处模拟网络请求，3s后回调

Observable.timer(3, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io()) //订阅在io线程
.observeOn(AndroidSchedulers.mainThread()) //回调在主线程
.compose(mView.bindUntilEvent(ActivityEvent.DESTROY)) //指定在onDestroy销毁
// .compose(mView.bindToLifecycle()) //取消订阅交由RxLifeCycle来判断
.subscribe(new Observer<Long>() {
@Override
public void onSubscribe(Disposable d) {
mView.showLoading();
}

@Override
public void onNext(Long aLong) {
KLog.d("onTest1Success()回调成功");
mView.onTest1Success("onTest1 请求成功");
}

@Override
public void onError(Throwable e) {
mView.dismissLoading();
}

@Override
public void onComplete() {
mView.dismissLoading();
}
});

一般使用，可以用compose封装一下，如下：

/**
     * 统一线程处理，且绑定生命周期
     * 用法： xxx .compose(RxUtil.rxSchedulerHelper(mView))
     * @param view
     * @param <T>
     * @return
     */
    public static <T> ObservableTransformer<T,T> rxSchedulerHelper(BaseView view){
        return new ObservableTransformer<T,T>() {
            @Override
            public ObservableSource<T> apply(@NonNull Observable<T> upstream) {
                return upstream
                        .subscribeOn(Schedulers.io())    //订阅在io线程
                        .unsubscribeOn(Schedulers.io())  //取消订阅在io线程，为啥要这个，不太清楚
                        .observeOn(AndroidSchedulers.mainThread()) //回调在主线程
                        .compose(view.bindToLifecycle());  //绑定生命周期
            }
        };
    }

结尾

在网上找了一些操作符的简单例子和应用场景，写了一个Demo，看操作符概述的时候，结合代码运行结果来看更容易理解，有兴趣的可以看看。后续会在这个项目里继续更新操作符和应用场景。
Github地址：
https://github.com/Dengszzzzz/DRxJavaSummary

参考

给 Android 开发者的 RxJava 详解
这可能是最好的RxJava 2.x 入门教程
Android拾萃 - RxJava2操作符汇总
Rxjava2入门教程五：Flowable背压支持——对Flowable最全面而详细的讲解
Android Rxjava：这是一篇 清晰 & 易懂的Rxjava 入门教程
实际应用场景
Android 用RxJava模拟一个EventBus ———RxBus
RxLifecycle详细解析
...