转载请标明出处：

http://blog.csdn.net/xmxkf/article/details/51821940

本文出自:【openXu的博客】

目录：

• 使用示例
• subscribeOn原理
  • 多次subscribeOn的情况
• observeOn原理
• 调度器的种类
• 各种操作符的默认调度器
• 源码下载

??RxJava中 使用observeOn和subscribeOn操作符，你可以让Observable在一个特定的调度器上执行，observeOn指示一个Observable在一个特定的调度器上调用观察者的onNext, onError和onCompleted方法，subscribeOn则指示Observable将全部的处理过程（包括发射数据和通知）放在特定的调度器上执行。

1. 使用示例

先看看下面的例子，体验一下在RxJava中 如何使用线程的切换：

private void logThread(Object obj, Thread thread){
    Log.v(TAG, "onNext:"+obj+" -"+Thread.currentThread().getName());
}
Observable.OnSubscribe onSub = new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        Log.v(TAG, "OnSubscribe -"+Thread.currentThread());
        subscriber.onNext(1);
        subscriber.onCompleted();
    }
};
Log.v(TAG, "--------------①-------------");
Observable.create(onSub)
        .subscribe(integer->logThread(integer, Thread.currentThread()));
Log.v(TAG, "--------------②-------------");
Observable.create(onSub)
        .subscribeOn(Schedulers.io())
        .subscribe(integer->logThread(integer, Thread.currentThread()));
Log.v(TAG, "--------------③-------------");
Observable.create(onSub)
        .subscribeOn(Schedulers.newThread())
        .subscribe(integer->logThread(integer, Thread.currentThread()));
Log.v(TAG, "--------------④-------------");
Observable.create(onSub)
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(integer->logThread(integer, Thread.currentThread()));
Log.v(TAG, "--------------⑤-------------");
Observable.create(onSub)
        .subscribeOn(Schedulers.newThread())
        .observeOn(Schedulers.newThread())
        .subscribe(integer->logThread(integer, Thread.currentThread()));
Log.v(TAG, "--------------⑥-------------");
Observable.interval(100, TimeUnit.MILLISECONDS)
        .take(1)
        .subscribe(integer->logThread(integer, Thread.currentThread()));
/*
 输出：
 --------------①-------------
 OnSubscribe -Thread[main,5,main]
 onNext:1 -Thread[main,5,main]
 --------------②-------------
 OnSubscribe -Thread[RxIoScheduler-2,5,main]
 onNext:1 -Thread[RxIoScheduler-2,5,main]
 --------------③-------------
 OnSubscribe -Thread[RxNewThreadScheduler-1,5,main]
 onNext:1 -Thread[RxNewThreadScheduler-1,5,main]
 --------------④-------------
 OnSubscribe -Thread[RxNewThreadScheduler-2,5,main]
 onNext:1 -Thread[main,5,main]
 --------------⑤-------------
 OnSubscribe -Thread[RxNewThreadScheduler-4,5,main]
 onNext:1 -Thread[RxNewThreadScheduler-3,5,main]
 --------------⑥-------------
 onNext:0 -RxComputationScheduler-3
 */

从上面的输出结果中，我们大概知道了下面几点：

①. RxJava中已经封装了多种调度器，不同的调度器可以指定在不同的线程中执行和观察

②. create创建的Observable默认在当前线程（主线程）中执行任务流，并在当前线程观察

③. interval创建的Observable会在一个叫Computation的线程中执行任务流和观察任务流

④. 除了observeOn和subscribeOn ，使用其他创建或者变换操作符也有可能造成线程的切换

2. subscribeOn()原理

??subscribeOn()用来指定Observable在哪个线程中执行事件流，也就是指定Observable中OnSubscribe(计划表)的call方法在那个线程发射数据。下面通过源码分析subscribeOn是怎样实现线程的切换的。

下面看看subscribeOn方法：

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

我们看到他创建了一个新的Observable，并为新的Observable创建了新的计划表OperatorSubscribeOn对象，新的计划表保存了原始Observable对象和调度器scheduler。接着我们看看OperatorSubscribeOn：

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

    final Scheduler scheduler;   //调度器
    final Observable<T> source;  //原始Observable
    //①.原始观察者订阅了新的Observable后，将执行此call方法
    @Override
    public void call(final Subscriber<? super T> subscriber) {
        final Scheduler.Worker inner = scheduler.createWorker();
        subscriber.add(inner);
        //②. call方法中使用传入的调度器创建的Worker对象的schedule方法切换线程
          inner.schedule(new Action0() {
            @Override
            public void call() {
                final Thread t = Thread.currentThread();
                //③ .创建了一个新的观察者
                    Subscriber<T> s = new Subscriber<T>(subscriber) {
                    @Override
                    public void onNext(T t) {
                        //⑤. 新的观察者收到数据后直接发送给原始观察者
                        subscriber.onNext(t);
                    }
                    ...
                };
                //④. 在切换的线程中，新的观察者订阅原始Observable，用来接收数据
                source.unsafeSubscribe(s);
            }
        });
    }
}

??上面源码中注释已经写的很清楚了，OperatorSubscribeOn其实就是一个普通的任务表，用于为新的Observable发射数据，只是不是真正的发射，它创建了一个新的观察者订阅原始Observable，这样就可以接受原始Observable发射的数据，然后直接发送给原始观察者。

??在call方法中通过scheduler.createWorker().schedule()完成线程的切换，这里就牵扯到两个对象了，Scheduler和Worker，不要着急，一个个的看，先看Scheduler，由于RxJava中有多种调度器，我们就看一个简单的Schedulers.newThread()，其他调度器的思路是一样的，下面一步一步看源码：

public final class Schedulers {
    //各种调度器对象
    private final Scheduler computationScheduler;
    private final Scheduler ioScheduler;
    private final Scheduler newThreadScheduler;
    //单例，Schedulers被加载的时候，上面的各种调度器对象已经初始化
    private static final Schedulers INSTANCE = new Schedulers();
    //构造方法
    private Schedulers() {
        RxJavaSchedulersHook hook = RxJavaPlugins.getInstance().getSchedulersHook();
        ...
        Scheduler nt = hook.getNewThreadScheduler();
        if (nt != null) {
            newThreadScheduler = nt;
        } else {
            //①.创建newThreadScheduler对象
            newThreadScheduler = RxJavaSchedulersHook.createNewThreadScheduler();
        }
    }
    //②. 获取NewThreadScheduler对象
    public static Scheduler newThread() {
        return INSTANCE.newThreadScheduler;
    }
    ...
}

Schedulers中保存了多种调度器对象，在Schedulers被加载的时候，他们就被初始化了，Schedulers就像是一个调度器的管理器，接着跟踪RxJavaSchedulersHook.createNewScheduler()，最终会找到一个叫NewThreadScheduler的类：

public final class NewThreadScheduler extends Scheduler {
    private final ThreadFactory threadFactory;
    public NewThreadScheduler(ThreadFactory threadFactory) {
        this.threadFactory = threadFactory;
    }
    @Override
    public Worker createWorker() {
        return new NewThreadWorker(threadFactory);
    }
}

NewThreadScheduler就是我们调用subscribeOn(Schedulers.newThread() )传入的调度器对象，每个调度器对象都有一个createWorker方法用于创建一个Worker对象，而NewThreadScheduler对应创建的Worker是一个叫NewThreadWorker的对象，在新产生的OperatorSubscribeOn计划表中就是通过NewThreadWorker.schedule(Action0)实现线程的切换，下面我们跟踪schedule(Action0)方法：

public class NewThreadWorker extends Scheduler.Worker implements Subscription {
    private final ScheduledExecutorService executor;   //
    public NewThreadWorker(ThreadFactory threadFactory) {
        //创建一个线程池
        ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, threadFactory);
        executor = exec;
    }
    @Override
    public Subscription schedule(final Action0 action) {
        return schedule(action, 0, null);
    }
    @Override
    public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
        return scheduleActual(action, delayTime, unit);
    }
    //重要：worker.schedule()最终调用的是这个方法
    public ScheduledAction scheduleActual(final Action0 action, long delayTime, TimeUnit unit) {
        //return action;
        Action0 decoratedAction = schedulersHook.onSchedule(action);
        //ScheduledAction就是一个Runnable对象，在run()方法中调用了Action0.call()
        ScheduledAction run = new ScheduledAction(decoratedAction);
        Future<?> f;
        if (delayTime <= 0) {
            f = executor.submit(run);   //将Runnable对象放入线程池中
        } else {
            f = executor.schedule(run, delayTime, unit);  //延迟执行
        }
        run.add(f);

        return run;
    }
    ...
}

我们发现OperatorSubscribeOn计划表中通过NewThreadWorker.schedule(Action0)，将Action0放入到一个线程池中执行，这样就实现了线程的切换。

通过上面的分析，我们知道subscribeOn是怎样将任务表放入线程池中执行的，感觉还是有点绕，看下图：

????

多次subscribeOn()的情况

我们发现，每次使用subscribeOn都会产生一个新的Observable，并产生一个新的计划表OnSubscribe，目标Subscriber最后订阅的将是最后一次subscribeOn产生的新的Observable。在每个新的OnSubscribe的call方法中都会有一个产生一个新的线程，在这个新线程中订阅上一级Observable，并创建一个新的Subscriber接受数据，最终原始Observable将在第一个新线程中发射数据，然后传送给给下一个新的观察者，直到传送到目标观察者，所以多次调用subscribeOn只有第一个起作用（这只是表面现象，其实每个subscribeOn都切换了线程，只是最终目标Observable是在第一个subscribeOn产生的线程中发射数据的）。看下图：

????

??多次subscribeOn()只有第一个会起作用，后面的只是在第一个的基础上在外面套了一层壳，就像下面的伪代码，最后执行是在第一个新线程中执行：

...
//第3个subscribeOn产生的新线程
new Thread(){
    @Override
    public void run() {
        Subscriber s1 = new Subscriber();
        //第2个subscribeOn产生的新线程
        new Thread(){
            @Override
            public void run() {
                Subscriber s2 = new Subscriber();
                //第1个subscribeOn产生的新线程
                new Thread(){
                    @Override
                    public void run() {
                        Subscriber<T> s3 = new Subscriber<T>(subscriber) {
                            @Override
                            public void onNext(T t) {
                                subscriber.onNext(t);
                            }
                            ...
                        };
                        //①. 最后一个新观察者订阅原始Observable
                        原始Observable.subscribe(s3);
                        //②. 原始Observable将在此线程中发射数据

                              //③. 最后一个新的观察者s3接受数据

                              //④. s3收到数据后，直接发送给s2，s2收到数据后传给s1,...最后目标观察者收到数据
                         }
                }.start();
            }
        }.start();
    }
}.start();

3. observeOn原理

??observeOn调用的是lift操作符，lift操作符在上一篇博客中讲过。lift操作符创建了一个代理的Observable，用于接收原始Observable发射的数据，然后在Operator中对数据做一些处理后传递给目标Subscriber。

??observeOn一样创建了一个代理的Observable，并创建一个代理观察者接受上一级Observable的数据，代理观察者收到数据之后会开启一个线程，在新的线程中，调用下一级观察者的onNext、onCompete、onError方法。

我们看看observeOn操作符的源码：

public final class OperatorObserveOn<T> implements Observable.Operator<T, T> {
    private final Scheduler scheduler;
    //创建代理观察者，用于接收上一级Observable发射的数据
    @Override
    public Subscriber<? super T> call(Subscriber<? super T> child) {
        if (scheduler instanceof ImmediateScheduler) {
            return child;
        } else if (scheduler instanceof TrampolineScheduler) {
            return child;
        } else {
            ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler, child, delayError, bufferSize);
            parent.init();
            return parent;
        }
    }

    //代理观察者
    private static final class ObserveOnSubscriber<T> extends Subscriber<T> implements Action0 {
        final Subscriber<? super T> child;
        final Scheduler.Worker recursiveScheduler;
        final NotificationLite<T> on;
        final Queue<Object> queue;
        //接受上一级Observable发射的数据
        @Override
        public void onNext(final T t) {
            if (isUnsubscribed() || finished) {
                return;
            }
            if (!queue.offer(on.next(t))) {
                onError(new MissingBackpressureException());
                return;
            }
            schedule();
        }
        @Override
        public void onCompleted() {
            ...
            schedule();
        }
        @Override
        public void onError(final Throwable e) {
            ...
            schedule();
        }
        //开启新线程处理数据
        protected void schedule() {
            if (counter.getAndIncrement() == 0) {
                recursiveScheduler.schedule(this);
            }
        }
        // only execute this from schedule()
        //在新线程中将数据发送给目标观察者
        @Override
        public void call() {
            long missed = 1L;
            long currentEmission = emitted;
            final Queue<Object> q = this.queue;
            final Subscriber<? super T> localChild = this.child;
            final NotificationLite<T> localOn = this.on;
            for (;;) {
                while (requestAmount != currentEmission) {
                    ...
                    localChild.onNext(localOn.getValue(v));
                }
            }
        }
    }
}

可以发现，observeOn操作符对它后面的操作产生影响，比如下面一段代码：

Observable.just(100)
        .subscribeOn(Schedulers.computation())     //Computation线程中发射数据
        .map(integer -> {return "map1-"+integer;}) //Computation线程中接受数据
        .observeOn(Schedulers.io())          //②. 切换
        .map(integer -> {return "map2-"+integer;}) //io线程中接受数据，由②决定
        .observeOn(Schedulers.newThread())   //③. 切换
        .map(integer -> {return "map3-"+integer;}) //newThread线程中接受数据，由③决定
        .observeOn(AndroidSchedulers.mainThread()) //④. 切换
        .delay(1000, TimeUnit.MILLISECONDS)        //主线程中接受数据，由④决定
        .subscribe(str -> logThread(str, Thread.currentThread()));  //Computation线程中接受数据，由④决定

/*
    说明：最后目标观察者将在Computation线程中接受数据，这取决于delay操作符，
    delay操作符是在Computation线程中执行的，执行完后就会将数据发送给目标观察者。
    而他上面的observeOn将决定于delay产生的代理观察者在主线程中接受数据
 */

/*
 输出：
 onNext:map3-map2-map1-100 -RxComputationScheduler-3
 */

只要涉及到lift操作符，其实就是生成了一套代理的Subscriber(观察者)、Observable(被观察者)和OnSubscribe(计划表)。Observable最典型的特征就是链式调用，我们暂且将每一步操作称为一级。代理的OnSubscribe中的call方法就是让代理Subscriber订阅上一级Observable，直到订阅到原始Observable发射数据，代理Subscriber收到数据后，可能对数据做一些操作也有可能切换线程，然后将数据传送给下一级Subscriber，直到目标观察者接收到数据，目标观察者在那个线程接受数据取决于上一个Subscriber在哪一个线程调用目标观察者的方法。示意图如下：

????

4. 调度器的种类

RxJava中可用的调度器有下面几种：

在RxAndroid中新增了一个：

5. 各种操作符的默认调度器

??在之前学习各种操作符的时候，都会介绍xx操作符默认在xxx调度器上执行，当时可能不太注意这是什么意思，下面总结了一些操作符默认的调度器：

源码下载：

https://github.com/openXu/RxJavaTest