上一次分析了RxJava的運(yùn)作流程魏保，其中的線程調(diào)度方面只是簡單提了兩句熬尺，以我看來，線程調(diào)度是RxJava中非常重要的一環(huán)谓罗，所以今天單獨(dú)拿出來分析一下粱哼。

subscribeOn

observeOn

subscribeOn調(diào)用可以將之前的操作加如線程池，從而保證運(yùn)行于子線程中檩咱，observeOn會使后邊的執(zhí)行運(yùn)行于主線程揭措，這里的之前和后邊均是指的代碼結(jié)構(gòu)上的前后

subscribeOn

經(jīng)過上一篇的分析，可以知道刻蚯，當(dāng)subscribeOn調(diào)用的時候绊含，會創(chuàng)建一個ObservableSubscribeOn對象返回，與此同時炊汹，上一級產(chǎn)生的對象會被保存在當(dāng)前對象的source變量中躬充，并且，將創(chuàng)建出一個線程池，先看線程池的創(chuàng)建麻裳，這里直接以io線程為例
Schedulers.io(Schedulers.io())

public static Scheduler io() {
       return RxJavaPlugins.onIoScheduler(IO);
}

其中的IO是在Schedulers類加載的時候就創(chuàng)建出來的口蝠，從這個結(jié)構(gòu)可以看出，IO就是IoScheduler對象津坑，RxJavaPlugins.initIoScheduler方法接收一個Callable線程妙蔗，返回callable.call，也就是call方法中返回的就是這個函數(shù)的返回值（Callable是另一種開啟線程的方式疆瑰，這個線程有返回值眉反，當(dāng)返回值獲取到之前，會阻塞當(dāng)前線程）

IO = RxJavaPlugins.initIoScheduler(new Callable<Scheduler>() {
            @Override
            public Scheduler call() throws Exception {
                return IoHolder.DEFAULT;
            }
        });

static final class IoHolder {
        static final Scheduler DEFAULT = new IoScheduler();
    }

那么IoScheduler是什么穆役？當(dāng)IoScheduler創(chuàng)建的時候

    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();
        }
    }

NONE是IoScheduler中創(chuàng)建的一個線程池,所以IoScheduler其實就是一個封裝好了的線程池對象

    static final CachedWorkerPool NONE;
    static {
        NONE = new CachedWorkerPool(0, null);
    }

    CachedWorkerPool(long keepAliveTime, TimeUnit unit) {
            this.keepAliveTime = unit != null ? unit.toNanos(keepAliveTime) : 0L;
            this.expiringWorkerQueue = new ConcurrentLinkedQueue<ThreadWorker>();
            this.allWorkers = new CompositeDisposable();

            ScheduledExecutorService evictor = null;
            Future<?> task = null;
            if (unit != null) {
                evictor = Executors.newScheduledThreadPool(1, EVICTOR_THREAD_FACTORY);
                task = evictor.scheduleWithFixedDelay(this, this.keepAliveTime, this.keepAliveTime, TimeUnit.NANOSECONDS);
            }
            evictorService = evictor;
            evictorTask = task;
        }

Schedulers.io(Schedulers.io())的調(diào)用寸五，執(zhí)行了兩個動作，第一耿币，保存上一級的對象梳杏，第二創(chuàng)建線程池

observeOn(AndroidSchedulers.mainThread())

接下來來看主線程的切換,調(diào)用observeOn方法，創(chuàng)建ObservableObserveOn對象淹接，同樣保存上一級產(chǎn)生的對象到source中十性，這里指的就是subscribeOn返回的對象ObservableSubscribeOn，并且保存?zhèn)魅氲腟cheduler--AndroidSchedulers.mainThread()

    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.CUSTOM)
    public final Observable<T> observeOn(Scheduler scheduler) {
        return observeOn(scheduler, false, bufferSize());
    }

    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.CUSTOM)
    public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
        return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
    }

進(jìn)入AndroidSchedulers.mainThread(),與上邊同樣的寫法塑悼，最后返回HandlerScheduler

    public static Scheduler mainThread() {
        return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
    }
    private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
            new Callable<Scheduler>() {
                @Override public Scheduler call() throws Exception {
                    return MainHolder.DEFAULT;
                }
            });

    private static final class MainHolder {
        //可以猜測這個HandlerScheduler是一個通過對Handler進(jìn)行封裝
        //運(yùn)行于主線程的線程劲适，可以看到Looper.getMainLooper()傳入了一個主線程的
        //looper對象，事實上也是如此
        static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));
    }

所以厢蒜，很類似霞势，observeOn(AndroidSchedulers.mainThread())同樣是做了兩件事，保存source和Scheduler斑鸦，那么兩種線程是如何進(jìn)行調(diào)度的愕贡，其實看到這里，還沒有進(jìn)入正題巷屿，真正的邏輯其實在subscribe方法上颂鸿。

subscribe

以subscribe(new Observer<String>())為例說明（new Consumer最終源碼也是相同的），調(diào)用subscribe方法后會來到Observable的抽象方法subscribeActual中攒庵，所以我們要到當(dāng)前Observable實現(xiàn)類中找這個方法嘴纺，按照上邊程序調(diào)用的順序，此時浓冒，調(diào)用subscribe方法的對象是observeOn方法產(chǎn)生的ObservableObserveOn栽渴，進(jìn)入這個類，找到subscribeActual方法

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        //這個scheduler是指AndroidSchedulers.mainThread()稳懒，也就是HandlerScheduler
        if (scheduler instanceof TrampolineScheduler) {
            source.subscribe(observer);
        } else {
            //創(chuàng)建一個worker
            Scheduler.Worker w = scheduler.createWorker();
            source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
        }
    }

進(jìn)入HandlerScheduler找到createWorker方法闲擦，這里創(chuàng)建了一個HandlerWorker對象,看到這里大概也可以猜測一下慢味，HandlerWorker中的schedule方法將會是一個關(guān)鍵，傳入的handler是主線程中的handler墅冷，明顯是要通過消息機(jī)制發(fā)送到主線程執(zhí)行纯路，問題的關(guān)鍵，在于是怎么發(fā)送到主線程執(zhí)行的寞忿，schedule方法的具體執(zhí)行我們暫且不看驰唬，按照程序執(zhí)行順序繼續(xù)往下走

@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) {
            ......
            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.
            handler.sendMessageDelayed(message, unit.toMillis(delay));
            ......
            return scheduled;
        }
        ......
    }

在創(chuàng)建了worker之后，調(diào)用方法subscribe,source很明顯是ObservableObserveOn對象創(chuàng)建的時候所保存的上一級的調(diào)用subscribeOn方法產(chǎn)生的ObservableSubscribeOn對象腔彰，通過這個對象調(diào)用subscribe方法叫编，又會進(jìn)入到ObservableSubscribeOn的subscribeActual方法。observer指的是我們代碼中傳入的observer（subscribe時new的那個）霹抛，這里對observer封裝了一層搓逾，以O(shè)bserveOnObserver的形式傳入到ObservableSubscribeOn的subscribeActual方法中，向上層傳遞了一級杯拐，可以參考08.RxJava運(yùn)作流程源碼分析中提供的流程圖

//source指ObservableSubscribeOn對象
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));

來到ObservableSubscribeOn的subscribeActual

@Override
    //參數(shù)s指的時對observer封裝了一層之后的ObserveOnObserver（new ObserveOnObserver(new Observer )）
    public void subscribeActual(final Observer<? super T> s) {
        //對ObserveOnObserver對象進(jìn)行一次封裝
        //此時Observer已經(jīng)被封裝了兩層
        //（new SubscribeOnObserver(new ObserveOnObserver(new Observer))）
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
        //調(diào)用ObserveOnObserver對象的onSubscribe
        s.onSubscribe(parent);
        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }

看看ObserveOnObserver的onSubscribe方法

        @Override
        public void onSubscribe(Disposable s) {
            if (DisposableHelper.validate(this.s, s)) {
                this.s = s;
                //注意這里這個判斷這次是不會滿足的霞篡，也就是這里的代碼不會走
                if (s instanceof QueueDisposable) {
                    @SuppressWarnings("unchecked")
                    QueueDisposable<T> qd = (QueueDisposable<T>) s;

                    int m = qd.requestFusion(QueueDisposable.ANY | QueueDisposable.BOUNDARY);
                    //同步，如果是要同步執(zhí)行端逼，就是指如果設(shè)置了在主線程執(zhí)行寇损，那么
                    //就執(zhí)行schedule()，往下看可以發(fā)現(xiàn)是使用我我們創(chuàng)建的worker
                    //發(fā)送到主線程執(zhí)行
                    if (m == QueueDisposable.SYNC) {
                        sourceMode = m;
                        queue = qd;
                        done = true;
                        //actual指的就是我們傳入的最原始的那個observer
                        actual.onSubscribe(this);
                        schedule();
                        return;
                    }
                    //異步裳食，如果是異步執(zhí)行，直接在當(dāng)前線程執(zhí)行芙沥，當(dāng)前線程也就是子線程
                    if (m == QueueDisposable.ASYNC) {
                        sourceMode = m;
                        queue = qd;
                        actual.onSubscribe(this);
                        return;
                    }
                }

                queue = new SpscLinkedArrayQueue<T>(bufferSize);
                //actual是我們new的那個Observer诲祸，所以這里直接回調(diào)了onSubscribe方法
                actual.onSubscribe(this);
            }
        }

scheduler就是Schedulers.io()得到的就是IoSchedule對象，在上邊分析subscribeOn方法時我們已經(jīng)知道這個對象是一個線程池而昨，調(diào)用scheduleDirect方法就是將SubscribeTask這個Runnable放進(jìn)了線程池執(zhí)行救氯，并且是在子線程中

@NonNull
    public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
        final Worker w = createWorker();
        final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
        DisposeTask task = new DisposeTask(decoratedRun, w);
        w.schedule(task, delay, unit);
        return task;
    }

createWorker()是個抽象類，在IoSchedule中找到重寫的方法

@Override
    public Worker createWorker() {
        return new EventLoopWorker(pool.get());
    }

所以這樣一來也就是說new SubscribeTask(parent))這個Runnable被放入了線程池執(zhí)行歌憨，這時候會調(diào)用它的run方法,這樣就又回到了調(diào)用上一級產(chǎn)生對象的subscribe方法中去了着憨，不同的是此時subscribe已經(jīng)是在線程池中執(zhí)行了（子線程）

        @Override
        public void run() {
            source.subscribe(parent);
        }

就這樣一級一級的往上調(diào)用，下一個會走到ObservableMap的subscribeActual方法务嫡，最后走到ObservableJust的subscribeActual,s.onSubscribe(sd)方法并沒有執(zhí)行什么東西甲抖，onSubscribe在之前已經(jīng)被調(diào)用了，重點(diǎn)在 sd.run()

    @Override
    protected void subscribeActual(Observer<? super T> s) {
        ScalarDisposable<T> sd = new ScalarDisposable<T>(s, value);
        s.onSubscribe(sd);
        sd.run();
    }

終于在這里要看到onNext onComplete方法的執(zhí)行了

        @Override
        public void run() {
            if (get() == START && compareAndSet(START, ON_NEXT)) {
                //observer 是 new MapObserver(new SubscribeOnObserver(new ObserveOnObserver(new Observer(){......})))
                observer.onNext(value);
                if (get() == ON_NEXT) {
                    lazySet(ON_COMPLETE);
                    observer.onComplete();
                }
            }
        }

此時一層一層的調(diào)用到這里心铃，observer對象已經(jīng)是經(jīng)過層層封裝包裹的observer了(new MapObserver(new SubscribeOnObserver(new ObserveOnObserver(new Observer(){......})))),所以調(diào)用observer.onNext會首先執(zhí)行MapObserver中的onNext,不管用戶調(diào)用了幾次map操作符准谚，都會一個一個的通過回調(diào)onNext方法執(zhí)行完成（如果有多個map方法被調(diào)用，當(dāng)執(zhí)行完一個apply方法后去扣，后邊的actual.onNext就會進(jìn)入下一個MapObserver中的onNext方法）柱衔，當(dāng)執(zhí)行到最后一個onNext方法的時候，此時這個actual表示的就是SubscribeOnObserver對象了，也就會去執(zhí)行它里邊的onNext

        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            if (sourceMode != NONE) {
                actual.onNext(null);
                return;
            }
            U v;
            try {
                //執(zhí)行apply方法唆铐，也就是map操作符中的回調(diào)方法
                v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }
            actual.onNext(v);
        }

SubscribeOnObserver中的onNext哲戚，這里的actual指的是ObserveOnObserver，所以又要去執(zhí)行它的onNext

        @Override
        public void onNext(T t) {
            actual.onNext(t);
        }

ObserveOnObserver中的onNext

        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            if (sourceMode != QueueDisposable.ASYNC) {
                queue.offer(t);
            }
            schedule();
        }

        void schedule() {
            if (getAndIncrement() == 0) {
                //這個worker是AndroidScheduler.mainThread得到的一個運(yùn)行于主線程的封裝類 HandlerWorker         
                worker.schedule(this);
            }
        }

在分析observeOn方法的時候我們已經(jīng)知道這個worker是AndroidScheduler.mainThread得到的一個運(yùn)行于主線程的封裝類 HandlerWorker 艾岂，worker.schedule(this)傳入的是一個Runnable顺少，也就是會在主線程中執(zhí)行這個Runnable，我們找到重寫的run方法澳盐。終于找到onNext和onComplete的最終執(zhí)行的地方了祈纯，并且我們知道，這兩個方法是在主線程執(zhí)行的

        @Override
        public void run() {
            if (outputFused) {
                drainFused();
            } else {
                //會執(zhí)行這個叼耙，上邊那個先不管
                drainNormal();
            }
        }

        void drainNormal() {
            int missed = 1;
            final SimpleQueue<T> q = queue;
            final Observer<? super T> a = actual;
            for (;;) {
                if (checkTerminated(done, q.isEmpty(), a)) {
                    return;
                }
                for (;;) {
                    boolean d = done;
                    T v;
                    try {
                        v = q.poll();
                    } catch (Throwable ex) {
                        Exceptions.throwIfFatal(ex);
                        s.dispose();
                        q.clear();
                        a.onError(ex);
                        worker.dispose();
                        return;
                    }
                    boolean empty = v == null;
                    if (checkTerminated(d, empty, a)) {
                        return;
                    }
                    if (empty) {
                        break;
                    }
                    a.onNext(v);
                }
                missed = addAndGet(-missed);
                if (missed == 0) {
                    break;
                }
            }
        }

到這里腕窥，RxJava線程調(diào)度的實現(xiàn)方式基本上我們已經(jīng)了解了。
這里可以插一個題外話筛婉，通常我們使用handler發(fā)送的消息都是在handleMessage方法中執(zhí)行簇爆，但是這里我們無論如何找不到這個方法的實現(xiàn)，那么handler是如何處理消息的爽撒？

        @Override
        public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
            ......
            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.
            handler.sendMessageDelayed(message, unit.toMillis(delay));
            ......
            return scheduled;
        }

可以看到這里Message message = Message.obtain(handler, scheduled)入蛆，看一下obtain方法的源碼會發(fā)現(xiàn)傳入的第二個參數(shù)是一個callback，保存到了message的成員變量m.callback中硕勿，當(dāng)handler調(diào)用sendMessageDelayed會將消息加入主線程的消息隊列（因為handler就是主線程的handler）哨毁，我們知道應(yīng)用啟動就會初始化一個主線程的handler一個looper和messageQueue（對消息機(jī)制不理解的可以看另一篇15.源碼閱讀（安卓消息機(jī)制）），調(diào)用looper.loop開啟一個無限循環(huán)不斷的從主線程消息隊列中取消息源武，我們看看它是如何取的

public static void loop() {
    for (;;) {
            Message msg = queue.next(); // might block
            ......
            msg.target.dispatchMessage(msg);
      }
}

無限循環(huán)中取到message后會執(zhí)行發(fā)送這個Message的handler中的dispatchMessage方法,這時候會判斷callback也就是我們上邊那個傳入的扼褪，如果它不能與null,就執(zhí)行handleCallback，執(zhí)行callback的run方法粱栖，找到這里終于找到為什么沒有handlerMessage仍然可以處理消息了

public void dispatchMessage(Message msg) {
        if (msg.callback != null) {
            handleCallback(msg);
        } else {
            if (mCallback != null) {
                if (mCallback.handleMessage(msg)) {
                    return;
                }
            }
            handleMessage(msg);
        }
    }

private static void handleCallback(Message message) {
        message.callback.run();
    }

傳入的callback是哪個话浇，就是 Message message = Message.obtain(handler, scheduled)中的schedule，schedule是哪個ScheduledRunnable 闹究，也就是說執(zhí)行的是ScheduledRunnable 的run方法,delegate就是ScheduledRunnable 中傳入的那個runnable幔崖，追溯上去，這個runnable就是worker.schedule(this)中的this渣淤，所以可以找到重寫的run方法

        @Override
        public void run() {
            try {
                delegate.run();
            } catch (Throwable t) {
                RxJavaPlugins.onError(t);
            }
        }