强烈推荐 《给 Android 开发者的 RxJava 详解》 这篇文章，本文的部分图片来源于此，感谢原作者！

注：本文研究的 RxJava 源码版本是 1.1.1，使用的根据代码自动生成类图的工具为 Eclipse 插件 AmaterasUML。

SubscribeOn 与 ObserveOn

RxJava 通过Observable 中的 subscribeOn() 和 observeOn() 方法实现线程切换。

subscribeOn() : 指定 subscribe() 所发生的线程，即 Observable.OnSubscribe被激活时所处的线程，或者叫做事件产生的线程。
observeOn() : 指定 Subscriber 所运行在的线程，或者叫做事件消费的线程。

示例

Observable.create(new Observable.OnSubscribe<String>() {
o@Override
opublic void call(Subscriber<? super String> s) {
os.onNext("Hello, world!");
os.onCompleted();
o}
o})
o.subscribeOn(Schedulers.io())// 指定 subscribe() 发生在 IO 线程
o.observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
o.subscribe(new Subscriber<String>() {
o@Override
opublic void onNext(String s) {
oLog.d(tag, s);
o}

o@Override
opublic void onCompleted() {
o}

o@Override
opublic void onError(Throwable e) {
o}
o});

SubscribeOn

原理

事件产生的线程就是初始 Observable 的 OnSubscribe 的 call 方法被调用时所在线程。

subscribeOn 方法返回一个新的 Observable。订阅（subscribe）发生时，新的 Observable 的 OnSubscribe 的 call 方法会被调用，这个方法将会在新的线程中（由调度器 Scheduler 指定）调用初始的 Observable 中的 OnSubscribe 的 call 方法。

原理图

流程

初始的 Observable 调用 subscribeOn 方法，接着该方法调用 create 方法，参数是 OperatorSubscribeOn。

public final Observable<T> subscribeOn(Scheduler scheduler) {
        return create(new OperatorSubscribeOn<T>(this, scheduler));
}

OperatorSubscribeOn 是一个操作符，它实现的是 OnSubscribe 接口。OperatorSubscribeOn 持有初始Observable 和调度器 Scheduler 的引用，并实现了 OnSubscribe 的 call 方法。

public final class OperatorSubscribeOn<T> implements OnSubscribe<T> {

    final Scheduler scheduler;
    final Observable<T> source;

    @Override
    public void call(final Subscriber<? super T> subscriber) {
o//……
    }
}

create 方法内创建了一个新的 Observable，这个新的 Observable 中的 OnSubscribe 对象就是操作符 OperatorSubscribeOn 的实例。

public static <T> Observable<T> create(OnSubscribe<T> f) {
        return new Observable<T>(hook.onCreate(f));
}

SubscribeOn 方法将返回这个新的 Observable。

之后的 subscribe 方法的是这个新的 Observable 的方法，传入的参数是一个 subscriber。subscribe 方法会去调用这个新的 Observable 中的 onSubscribe 的 call 方法，也就是 OperatorSubscribeOn 的 call 方法，参数是就是原来的 subscriber。

private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
    //……
ohook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
o//……
}

OperatorSubscribeOn 从调度器 scheduler 获取一个 worker，worker 知道将要在哪个线程上调度任务。call 方法中创建了一个新的任务交给 worker 调度。这个任务中根据方法传入的 subscriber 创建了一个新的 subscriber，这个新的 subscriber 相当于原来的 subscriber 的代理，onNext、onError、onComplete 方法都直接交给原来的 subscriber 来做，但是重写了一个 setProducer 方法(不知道做什么的，大概与 backpressure 有关，暂时不用管)。任务中用初始的 Observable 来订阅（unsafeSubscribe）这个新的 subscriber。unsafeSubscribe 和 subscribe 差不多，区别在它没有考虑错误处理等。unsafeSubscribe 会调用初始的 Observable 中的 OnSubscribe 对象的 call 方法，这个方法将在 worker 指定的线程中执行，也就是事件将在 worker 指定的线程中产生。

public final class OperatorSubscribeOn<T> implements OnSubscribe<T> {

    @Override
    public void call(final Subscriber<? super T> subscriber) {
        final Worker inner = scheduler.createWorker();
        subscriber.add(inner);

        inner.schedule(new Action0() {
            @Override
            public void call() {
                Subscriber<T> s = new Subscriber<T>(subscriber) {
                    @Override
                    public void onNext(T t) {
                        subscriber.onNext(t);
                    }

                    @Override
                    public void onError(Throwable e) {
                        try {
                            subscriber.onError(e);
                        } finally {
                            inner.unsubscribe();
                        }
                    }

                    @Override
                    public void onCompleted() {
                        try {
                            subscriber.onCompleted();
                        } finally {
                            inner.unsubscribe();
                        }
                    }

                    @Override
                    public void setProducer(final Producer p) {
                        //……
                    }
                };

                source.unsafeSubscribe(s);
            }
        });
    }
}

ObserveOn

原理

事件消费的线程就是目标 Subscriber 的 OnNext 等方法被调用时所在的线程。

ObserveOn 方法返回一个新的 Observable，并创建一个新的 Subscriber 把目标 Subscriber 包装起来。 订阅（subscribe）发生时，新的 Observable 的 OnSubscribe 的 call 方法会被调用，这个方法接着调用初始 Observable 中的 OnSubscribe 的 call 方法，调用的参数是新的 subscriber。当新的 Subscriber 中的方法被调用时，会转为在新的线程中（由调度器 Scheduler 指定）调用目标 Subscriber 中对应的方法。

原理图

流程

初始的 Observable 调用 observeOn 方法，接着该方法调用 lift 方法（与 subscribeOn 不太一样），参数是 OperatorSubscribeOn。

public final Observable<T> observeOn(Scheduler scheduler) {
        //……
oreturn lift(new OperatorObserveOn<T>(scheduler, false));
}

OperatorObserveOn 也是一个操作符，但是它实现的是 Operator 接口。OperatorSubscribeOn 持有调度器Scheduler 的引用，并实现了 Operator 接口的 call 方法。

public final class OperatorObserveOn<T> implements Operator<T, T> {

    private final Scheduler scheduler;

    @Override
    public Subscriber<? super T> call(Subscriber<? super T> child) {
         //……
    }
}

与 SubscribeOn 类似，lift 方法中同样是创建了一个新的 Observable，在新的 Observable 的 OnSubscribe 的 call 方法中，调用初始的 Observable 中的 OnSubscribe 的 call 方法，方法参数是调用 OperatorObserveOn 的 call 方法返回的 subscriber，这个 subscriber 的类型是 ObserveOnSubscriber。

public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
        return new Observable<R>(new OnSubscribe<R>() {
            @Override
            public void call(Subscriber<? super R> o) {
                     Subscriber<? super T> st = hook.onLift(operator).call(o);
                     st.onStart();
                     onSubscribe.call(st);
        });
}



public final class OperatorObserveOn<T> implements Operator<T, T> {
    @Override
    public Subscriber<? super T> call(Subscriber<? super T> child) {
            ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler, child, delayError);
            parent.init();
            return parent;
}

ObserveOnSubscriber 相当于在原来的 Subscriber 上做了一层包装，它持有一个 worker 引用，它在 worker 中调用原来的 subscriber 的方法。

private static final class ObserveOnSubscriber<T> extends Subscriber<T> implements Action0 {
        final Subscriber<? super T> child;
        final Scheduler.Worker recursiveScheduler;

        @Override
        public void onNext(final T t) {
            //……
            schedule();
        }

        @Override
        public void onCompleted() {
            //……
            schedule();
        }

        @Override
        public void onError(final Throwable e) {
            //……
            schedule();
        }

        protected void schedule() {
            recursiveScheduler.schedule(this);
        }

        @Override
        public void call() {
              //……
              localChild.onNext(localOn.getValue(v));
              //……
        }

}

当 subscribe() 方法被调用时，新的 Observable 的 OnSubscribe 的 call 方法被调用，该方法会调用初始 Observable 的 OnSubscribe 的 call 方法，参数为新的 subscriber。调用新的 subscriber 中的方法时，会转为在指定的线程中调用原来的 subscriber 的方法，也就是事件在指定的线程中消费。

多个SubscribeOn和ObserveOn混合使用

图中共有 5 处含有对事件的操作。
由图中可以看出，①和②两处受第一个 subscribeOn() 影响，运行在红色线程；③和④处受第一个 observeOn() 的影响，运行在绿色线程；⑤处受第二个 onserveOn() 影响，运行在紫色线程；而第二个 subscribeOn() ，由于在通知过程中线程就被第一个 subscribeOn() 截断，因此对整个流程并没有任何影响。

所以当使用了多个 subscribeOn() 的时候，只有第一个 subscribeOn() 起作用。

Scheduler

调度器 Scheduler，相当于线程控制器，RxJava 通过它来指定每一段代码应该运行在什么样的线程。

Schedulers

Schedulers 是一个创建 Scheduler 的工厂，提供了多种调度器（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。

Scheduler 是一个抽象类，主要负责创建 worker，具体的任务由 worker 完成。 抽象类 Scheduler.Worker 代表了一个工作序列，它包含了具体的调度方法。Worker 实现了接口 Subscription 的 unsubscribe() 方法，用来在取消订阅时对资源进行回收。

我们选 Schedulers.newThread() 和 Schedulers.io() 两个例子进行说明。

Schedulers.newThread()

Schedulers.newThread() 返回了一个 NewThreadScheduler 对象。

NewThreadScheduler 继承了 Scheduler，并重写了 createWorker() 方法，在该方法中创建了一个NewThreadWorker。NewThreadWork 继承了 Scheduler.Worker，重写了 schedule() 方法。

NewThreadScheduler 每次通过 createWorker 创建一个 NewThreadWorker，任务由交给新的 worker 完成，并不需要做额外的操作。

NewThreadWorker 每次创建一个数量为 1 的线程池。

ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, threadFactory);

ScheduledExecutorService 是 java.util.concurrent 包中提供的类，它可以定时或者周期性地执行任务。

NewThreadWorker 中的 schedule 方法都通过调用 ScheduledExecutorService 的方法完成。 这样的话 NewThreadWorker 的代码应该不多，但是它又考虑新线程中的任务在执行之前被取消的问题。在 Java 7+，只需要开启 RemoveOnCancel 策略即可，任务被取消时会立即从任务队列移除，避免内存泄漏。对于不支持这个策略的 Java 版本，NewThreadWorker 会再创建一个新的线程，这个线程是 static 的，由所有 NewThreadWorker 共享，它将周期性地调用 exec.purge()；将每个 NewThreadWorker 中还未执行但已经被取消的任务移除。

Schedulers.io()

Schedulers.io() 返回的是一个 CachedThreadScheduler 对象。

类图如上所示，EventLoopWorker、CachedWrokerPool、ThreadWorker 都是 CachedThreadScheduler 的内部类。

CachedThreadScheduler 除了用来创建 EventLoopWorker 外，还持有一个 CachedWorkerPool。

CachedWorkerPool 管理一个 ThreadWorker 队列，这些 Threadworker 可以重用。ThreadWorker 继承自 NewThreadWorker，添加了一个过期时间的属性。CachedWorkerPool 另外还有一个 evictorService 线程，用于周期性地将过期的 ThreadWorker 从队列中移除。

EventLoopWorker 每次从 CachedWorkerPool 中取出一个 ThreadWorker。CachedThreadScheduler 的 createWorker 方法创建的是 EventLoopWorker，但实际的调度由 EventLoopWorker 从 CachedWorkerPool 获取的 ThreadWorker 来完成。

AndroidSchedulers

AndroidSchedulers 在 RxAndroid 的 rx.android.schedulers 包中，这个包只有 AndroidSchedulers.java 和 HandlerScheduler.java 两个文件。

AndroidSchedulers 中只有一个公开的方法 mainThread()，它返回的是一个 HandlerScheduler 对象。在创建这个 Scheduler 对象时，new Handler(Looper.getMainLooper()) 被作为构造参数。

同样的，HandlerScheduler 通过 createWorker 方法创建一个新的 HandlerWorker，HandlerWorker 是HandlerScheduler 的内部类。在 HandlerWorker 的 schedule 方法中，将任务交给 handler 完成。

handler.postDelayed(scheduledAction, unit.toMillis(delayTime));