引言

前面我们梳理了RxJava扩展的观察者模式的实现，今天我们学习RxJava的第二块核心内容:订阅方法的线程切换subscribeOn。

subscribeOn方法

public final Observable<T> subscribeOn(Scheduler scheduler) {
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
    }

根据scheduler对象和原来的Observable对象构造了新的ObservableSubscribeOn对象：

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T>{
    ...
}

继承自AbstractObservableWithUpstream,注意这里有两个泛型，结合名字意思，可以猜想这个类可能与Observable的变换相关。
AbstractObservableWithUpstream:

/**
 * Base class for operators with a source consumable.
 *
 * @param <T> the input source type
 * @param <U> the output type
 */
abstract class AbstractObservableWithUpstream<T, U> extends Observable<U> implements HasUpstreamObservableSource<T> {

    /** The source consumable Observable. */
    protected final ObservableSource<T> source;

    /**
     * Constructs the ObservableSource with the given consumable.
     * @param source the consumable Observable
     */
    AbstractObservableWithUpstream(ObservableSource<T> source) {
        this.source = source;
    }

    @Override
    public final ObservableSource<T> source() {
        return source;
    }

}

我们可以看到它内部封装了原ObservableSource<T>，而继承自Observable<U>，其中T为原来的数据类型，U为转换后的数据类型。
下面我们回头再看ObservableSubscribeOn:

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
      final Scheduler scheduler;
      //构造方法传入原来的ObservableSource和Scheduler对象
      public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
        super(source);
        this.scheduler = scheduler;
    }
   @Override
    public void subscribeActual(final Observer<? super T> s) {
        //构造相应的SubscribeOnObserver对象,包装原观察者
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
        s.onSubscribe(parent);
        //核心代码: scheduler.scheduleDirect将 source.subscribe(parent)代码交给scheduler调度
        parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
            @Override
            public void run() {
                source.subscribe(parent);
            }
        }));
    }
...
}

其中的核心代码，是scheduler.scheduleDirect，将source.subscribe(parent)封装到Runnable方法中，交给scheduler调度,实现的订阅方法的线程切换。
下面我们再看看核心类Scheduler:

线程切换Scheduler和Worker

Scheduler

负责

/**
 * A {@code Scheduler} is an object that specifies an API for scheduling
 * units of work with or without delays or periodically.
 * You can get common instances of this class in {@link io.reactivex.schedulers.Schedulers}.
 * 负责线程切换，支持延迟和周期任务调度
 */
public abstract class Scheduler {
    //子类构造任务执行单元Worker
    public abstract Worker createWorker();
    //开始任务
    public void start() {
    }
    //停止任务
    public void shutdown() {
    }
      
    //立即执行任务
    public Disposable scheduleDirect(Runnable run) {
        return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
    }    
    
    //run代码交给Worker调度
    public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
        //取Worker
        final Worker w = createWorker();
        final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
        //Worker任务调度
        w.schedule(new Runnable() {
            @Override
            public void run() {
                try {
                    decoratedRun.run();
                } finally {
                    w.dispose();
                }
            }
        }, delay, unit);
        return w;
    }
//调度周期性任务
public Disposable schedulePeriodicallyDirect(Runnable run, long initialDelay, long period, TimeUnit unit) {
        final Worker w = createWorker();
        final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
        PeriodicDirectTask periodicTask = new PeriodicDirectTask(decoratedRun, w);
        Disposable d = w.schedulePeriodically(periodicTask, initialDelay, period, unit);
        if (d == EmptyDisposable.INSTANCE) {
            return d;
        }
        return periodicTask;
   }
   ....
}

发现它的调度任务都是通过Worker实现，看看它有哪些东西:

Worker

public abstract static class Worker implements Disposable {
     //执行run代码
    public Disposable schedule(Runnable run) {
            return schedule(run, 0L, TimeUnit.NANOSECONDS);
     }
     //子类覆写具体的调度方法 
    public abstract Disposable schedule(Runnable run, long delay, TimeUnit unit);
....
}

主要的线程调度实现类

IoScheduler

这个类源码比较长，我们捡重点分析

/**
 * Scheduler that creates and caches a set of thread pools and reuses them if possible.
 */
public final class IoScheduler extends Scheduler {
//CachedWorkerPool线程安全的引用，CachedWorkerPool维护
final AtomicReference<CachedWorkerPool> pool;
...
//初始化CachedWorkerPool
public IoScheduler() {
        this.pool = new AtomicReference<CachedWorkerPool>(NONE);
        start();
    }

    @Override
    public void start() {
        CachedWorkerPool update = new CachedWorkerPool(KEEP_ALIVE_TIME, KEEP_ALIVE_UNIT);
        if (!pool.compareAndSet(NONE, update)) {
            update.shutdown();
        }
    }
   ...
   @Override
    //关键方法，根据ThreadWorker回收池构造EventLoopWorker
    public Worker createWorker() {
        return new EventLoopWorker(pool.get());
    }
    ....
}

再来看EventLoopWorker:

 static final class EventLoopWorker extends Scheduler.Worker {
        private final CachedWorkerPool pool;
        private final ThreadWorker threadWorker;
        EventLoopWorker(CachedWorkerPool pool) {
            this.pool = pool;
            this.tasks = new CompositeDisposable();
            //从缓存池中取
            this.threadWorker = pool.get();
        }
      ....
   
        @Override
        public Disposable schedule(Runnable action, long delayTime, TimeUnit unit) {
            if (tasks.isDisposed()) {
                // don't schedule, we are unsubscribed
                return EmptyDisposable.INSTANCE;
            }
            //最后调用threadWorker.scheduleActual执行run方法
            return threadWorker.scheduleActual(action, delayTime, unit, tasks);
        }
    }

先看看pool.get()方法:

static final class CachedWorkerPool implements Runnable{
     //未过期的闲置ThreadWorker队列
      private final ConcurrentLinkedQueue<ThreadWorker> expiringWorkerQueue;
       //执行清理过期ThreadWorker的线程池
       private final ScheduledExecutorService evictorService;
        CachedWorkerPool(long keepAliveTime, TimeUnit unit) {
            this.keepAliveTime = unit != null ? unit.toNanos(keepAliveTime) : 0L;
            this.expiringWorkerQueue = new ConcurrentLinkedQueue<ThreadWorker>();
           ...
            if (unit != null) {
                evictor = Executors.newScheduledThreadPool(1, EVICTOR_THREAD_FACTORY);
                 //延时执行清理ThreadWorker方法
                task = evictor.scheduleWithFixedDelay(this, this.keepAliveTime, this.keepAliveTime, TimeUnit.NANOSECONDS);
            }
            evictorService = evictor;
            evictorTask = task;
        }

        @Override
        public void run() {
            evictExpiredWorkers();
        }
        
        //从回收池中取ThreadWorker
        ThreadWorker get() {
            if (allWorkers.isDisposed()) {
                return SHUTDOWN_THREAD_WORKER;
            }
            while (!expiringWorkerQueue.isEmpty()) {
                ThreadWorker threadWorker = expiringWorkerQueue.poll();
                if (threadWorker != null) {
                    return threadWorker;
                }
            }

            // No cached worker found, so create a new one.
            ThreadWorker w = new ThreadWorker(WORKER_THREAD_FACTORY);
            allWorkers.add(w);
            return w;
        }
        //回收threadWorker
        void release(ThreadWorker threadWorker) {
            // Refresh expire time before putting worker back in pool
            threadWorker.setExpirationTime(now() + keepAliveTime);

            expiringWorkerQueue.offer(threadWorker);
        }
        //清理过期的ThreadWorker
        void evictExpiredWorkers() {
            if (!expiringWorkerQueue.isEmpty()) {
                long currentTimestamp = now();

                for (ThreadWorker threadWorker : expiringWorkerQueue) {
                    if (threadWorker.getExpirationTime() <= currentTimestamp) {
                        if (expiringWorkerQueue.remove(threadWorker)) {
                            allWorkers.remove(threadWorker);
                        }
                    } else {
                        // Queue is ordered with the worker that will expire first in the beginning, so when we
                        // find a non-expired worker we can stop evicting.
                        break;
                    }
                }
            }
        }
        ....
}

CachedWorkerPool维护了一个ThreadWorker回收池，EventLoopWorker从中取ThreadWorker来执行任务。
我们再回头看threadWorker.scheduleActual方法如何调度任务的，依然是挑重点看:

public class NewThreadWorker extends Scheduler.Worker implements Disposable {
     //线程池
    private final ScheduledExecutorService executor;
    ...
    public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, TimeUnit unit, DisposableContainer parent) {
        Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
        //封装原始的run
        ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);

        if (parent != null) {
            if (!parent.add(sr)) {
                return sr;
            }
        }

        Future<?> f;
        try {
            //runnable交给线程池调度
            if (delayTime <= 0) {
                f = executor.submit((Callable<Object>)sr);
            } else {
                f = executor.schedule((Callable<Object>)sr, delayTime, unit);
            }
            sr.setFuture(f);
        } catch (RejectedExecutionException ex) {
            parent.remove(sr);
            RxJavaPlugins.onError(ex);
        }
        return sr;
    }
    ...
}

到目前为止我们走完了IO线程调度的大致流程，下面再看看主线程的调度HandlerScheduler.

HandlerScheduler

外部切换主线程使用MainHolder. DEFAULT对象:

static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));

传入绑定MainLooper的Handler对象构造HandlerScheduler。

final class HandlerScheduler extends Scheduler {
    private final Handler handler;

    HandlerScheduler(Handler handler) {
        this.handler = handler;
    }

    @Override
    public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
        if (run == null) throw new NullPointerException("run == null");
        if (unit == null) throw new NullPointerException("unit == null");

        run = RxJavaPlugins.onSchedule(run);
        //封装原始的Runnable对象
        ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
        handler.postDelayed(scheduled, Math.max(0L, unit.toMillis(delay)));
        return scheduled;
    }

    @Override
    public Worker createWorker() {
        return new HandlerWorker(handler);
    }
    
    //最终的调度类
    private static final class HandlerWorker extends Worker {
        private final Handler handler;

        private volatile boolean disposed;

        HandlerWorker(Handler handler) {
            this.handler = handler;
        }

        @Override
        public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
            if (run == null) throw new NullPointerException("run == null");
            if (unit == null) throw new NullPointerException("unit == null");

            if (disposed) {
                return Disposables.disposed();
            }

            run = RxJavaPlugins.onSchedule(run);

            ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);

            Message message = Message.obtain(handler, scheduled);
            message.obj = this; // Used as token for batch disposal of this worker's runnables.
            //发送给主线程执行run
            handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)));

            // Re-check disposed state for removing in case we were racing a call to dispose().
            if (disposed) {
                handler.removeCallbacks(scheduled);
                return Disposables.disposed();
            }

            return scheduled;
        }

        @Override
        public void dispose() {
            disposed = true;
            handler.removeCallbacksAndMessages(this /* token */);
        }

        @Override
        public boolean isDisposed() {
            return disposed;
        }
    }
    //封装原始的Runnable对象
    private static final class ScheduledRunnable implements Runnable, Disposable {
        private final Handler handler;
        private final Runnable delegate;

        private volatile boolean disposed;

        ScheduledRunnable(Handler handler, Runnable delegate) {
            this.handler = handler;
            this.delegate = delegate;
        }

        @Override
        public void run() {
            try {
                delegate.run();
            } catch (Throwable t) {
                IllegalStateException ie =
                    new IllegalStateException("Fatal Exception thrown on Scheduler.", t);
                RxJavaPlugins.onError(ie);
                Thread thread = Thread.currentThread();
                thread.getUncaughtExceptionHandler().uncaughtException(thread, ie);
            }
        }

        @Override
        public void dispose() {
            disposed = true;
            handler.removeCallbacks(this);
        }

        @Override
        public boolean isDisposed() {
            return disposed;
        }
    }
}

分析完订阅方法的线程切换，我们可以考虑这个问题:订阅方法连续切换为什么总是以第一次调用为准？
我的理解如下:
1.在subscribeActual()里开启了Scheduler的工作，source.subscribe(parent);,从这一句开始切换了线程，所以在这之上的代码都是在切换后的线程里的了。
2.连续切换时,执行订阅操作时,最上面的切换(scheduleDirect)最后执行,此时source.subscribe(parent)所在的线程变成了最上面的subscribeOn(xxxx)指定的线程
3.发送数据的方法，最终还是在最上面的source.subscribe(parent)中执行。
接下来的博客我们继续分析观察者方法的线程调度。