一、RxJava简介

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

2012年启动，2013.2发布距今已7年，在Java和Android异步事件处理、数据过滤与变换提供函数响应式编程能力.

缘由：Netflix为增加服务器性能和吞吐量编写RxJava并开源，主要开发维护人员：David Karnok、Ben Christensen、Shixiong Zhu、Jake Wharton...

函数响应式编程：函数式编程+响应式编程，面向数据流和变化传播的编程范式（领域中主流的行事套路、规范化的关系）；
适用场景以及与AsyncTask、Handler、Executor适用场景异同：复杂网络数据流请求以及与操作符对于的数据变换处理，同时包含对OkHttp、Retrofit、Gson支持；
与RxAndroid关系：RxJava针对Android平台的扩展，为Android开发提供响应式扩展组件；
与Kotlin协程对比(待进一步分析)：RxJava基于Rx特性封装的Java函数库，协程是Kotlin编译器级别概念，理解为基于编译器or虚拟机设计的轻量级线程。两者都有数据变换、异步数据流处理能力；
RxKotlin：适配Kotlin的ReactiveX版本；
内存占用、代码混淆、包体积(待进一步分析)：内存占用RxJava > 协程，回收RxJava创建的对象，需给予更高的 CPU占用，也会转变为更高的电量损耗.

二、RxJava第一行代码与Observable订阅过程

RxJava第一行代码

Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) {
        emitter.onNext(1);
        emitter.onNext(2);
        emitter.onNext(3);
        emitter.onComplete();
    }
}).subscribe(new Observer<Integer>() {
    @Override
    public void onSubscribe(Disposable d) {
        Log.d(TAG, "subscribe");
    }

    @Override
    public void onNext(Integer value) {
        Log.d(TAG, "onNext: " + value);
    }

    @Override
    public void onError(Throwable e) {
        Log.d(TAG, "error");
    }

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

Obserbable订阅过程

1.整体流程：上游创建、下游创建、订阅
2.Observable create后subscribe下游Observer，使得Observable触发subscribeActual()方法
3.ObservableCreate类中subscribeActual()方法通过ObservableOnSubscribe类的subscribe()方法订阅CreateEmitter
4.Observer作为成员变量暴露给ObservableEmitter，ObservableEmitter调用onNext()、onError()、onComplete()将回调下游Observer的onNext()、onError()、onComplete()方法
注：RxJava基于观察者模式，而观察者模式本质是回调。上游订阅下游实际表现为：CreateEmitter传入Observer作为其成员变量，CreateEmitter调用onNext()发射数据，将立即回调Observer的onNext()方法

RxJava第一行代码类图及各方法关系如下：

RxJava_Observable_class.png

三、线程切换过程

    ...
        emitter.onNext(3);
        emitter.onComplete();
    }
}) .subscribeOn(Schedulers.io())                // 切换上游线程
   .observeOn(AndroidSchedulers.mainThread())   // 切换下游线程
   .subscribe(new Observer<Integer>() {
    @Override
    public void onSubscribe(Disposable d) {
        Log.d(TAG, "subscribe");
    }

    @Override
    public void onNext(Integer value) {
        Log.d(TAG, "onNext: " + value);
    }
    ...

原理：RxJava提供Scheduler作为线程调度器，底层通过封装Handler和Executor实现线程切换

上游线程切换subscribeOn(Schedulers.newThread())

subscribeOn()调用后，生成ObservableSubscribeOn对象，并将原有Observable作为成员变量传入
subscribeActual()方法中scheduler.scheduleDirect(new SubscribeTask(parent))  //实现线程切换；
SubscribeTask为Runable对象，run()方法执行source.subscribe(parent);            //上游随即切换到对应线程
scheduleDirect(Runnable run)方法内部创建了Worker对象，通过调用worker.shedule(Runnable, long delay, TimeUnit)实现真正线程切换，worker内部包含线程池等切线程操作

下游线程切换observeOn(AndroidSchedulers.mainThread())

observeOn()调用后，生成ObservableObserveOn对象，并将原有Observable作为成员变量传入
subscribeActual()方法中通过Scheduler.Worker w = scheduler.createWorker()创建Worker对象
source.subscribe(new ObserveOnObserver<>(observer, w, delayError, bufferSize)) 完成上下游订阅（给下游传入了Worker对象）
上游发射数据后，即ObserveOnObserver的onNext()方法被调用，导致worker.schedule(this)执行
ObserveOnObserver即上述方法中的this，它实现了Runnable接口，其run()方法运行在指定线程，run()中也会调用下游onNext()方法，下游开始执行，故下游即被切换到对应线程
注：下游线程切换后，由于上下游线程不同，上游发射数据后会将数据发射到消息队列，ObserveOnObserver拥有消息发送队列，被观察者发送消息即入队，下游从消息队列中取数据。消息队列queue是ObserveOnObserver的成员变量，queue创建于ObserveOnObserver的onSubscribe()方法，queue = new SpscLinkedArrayQueue<>(bufferSize)

注：上下游同线程，上游发射一次onNext(),下游onNext随即被调用一次，因为属于同步调用；上下游不同线程，属于异步调用，上游发射onNext()会进入消息队列，下游会从消息队列取数据回调其onNext()，故上下游调用顺序不能保证。
subscribeOn()切换线程时序图注：红色方框内为重点切线程操作

subscribeOn_sequence.png
observeOn()切换线程时序图

observeOn_sequence.png