RxJava2源码分析（三）：线程调度分析

迷南。 2022-03-16 14:14 207阅读 0赞

> 前言：经过前面两篇文章对RxJava2源码的分析，我们已经对RxJava2的基本流程及操作符的原理有了一定程度的认识。这篇文章将在前面两篇文章的基础上，对RxJava2的线程调度进行分析，建议先阅读前面两篇的文章，再阅读本文。
> 
> 注：文章内容过多，建议在空闲时阅读。

### 相关文章 ###

*  [RxJava2源码分析（一）：基本流程分析][RxJava2]
 *  [RxJava2源码分析（二）：操作符原理分析][RxJava2 1]

### 示例代码 ###

为了更好的理解RxJava2的线程调度原理，不被其他的代码所干扰，这里就只贴出与线程调度有关的代码，如下

private void threadScheduleCode() {
            Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                    Log.e("wizardev", "上游所在的线程: "+Thread.currentThread().getName());
                    Thread.sleep(2*1000);
                    emitter.onNext("wizardev");
                }
            }).subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Observer<String>() {
                        @Override
                        public void onSubscribe(Disposable d) {
                            Log.e("wizardev", "onSubscribe: "+Thread.currentThread().getName() );
                        }
                        @Override
                        public void onNext(String s) {
                            Log.e("wizardev", "接收到上游发射的数据为: " + s);
                            Log.e("wizardev", "下游所在的线程: "+ Thread.currentThread().getName());
                        }
                        @Override
                        public void onError(Throwable e) {
                        }
                        @Override
                        public void onComplete() {
    
                        }
                    });
        }
    
    复制代码

可以看下执行这段代码后打印的日志，如下

可以发现上游和下游确实不在同一个线程中，那么RxJava2是怎么进行线程切换的呢？想知道答案，请继续阅读本文。

### 本文要解决的问题 ###

本文要解决的问题其实就一个，就是RxJava2是如何进行线程调度的？但是，围绕着这个问题又会有两个小的问题需要解决：

1.  subscribeOn是怎样将要处理的数据放到到工作线程的？
2.  observeOn是怎样将工作线程切换到主线程的？

为了能够更容易理解线程调度的原理，这里对源码分析的顺序将会按照代码的执行顺序进行分析。

### subscribeOn方法分析 ###

因为前面的文章已经分析过了`create`方法，所以就直接分析`subscribeOn`这个方法，直接上源码，如下

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

有了前面分析源码的经验，可以知道，**`subscribeOn`方法其实就是返回了ObservableSubscribeOn类的实例并将上游的ObservableCreate和`subscribeOn`方法的参数注入到了它的构造方法中。** 继续看下ObservableSubscribeOn类的源码，如下

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
        final Scheduler scheduler;
    
        public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
            super(source);
            this.scheduler = scheduler;
        }
    
        @Override
        public void subscribeActual(final Observer<? super T> observer) {
            final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
    
            observer.onSubscribe(parent);
    
            parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
        }
        //...
        //省略部分源码
    }
    复制代码

**从源码中可以看到，这里分别将ObservableCreate类的实例以及`subscribeOn`方法的参数即`Schedulers.io()`作为了ObservableSubscribeOn类的成员变量。** 好了，上面的这些就是执行`subscribeOn(Schedulers.io())`这句代码所做的事情了，下面来看下`observeOn(AndroidSchedulers.mainThread())`这句代码所做的事情。

### observeOn方法分析 ###

直接看源码，如下

public final Observable<T> observeOn(Scheduler scheduler) {
            return observeOn(scheduler, false, bufferSize());
        }
    
    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));
        }
    复制代码

从上面的代码可以看出，`observeOn`方法最终调用的是含有三个参数的`observeOn`方法，而**这个方法的作用是返回了ObservableObserveOn类的实例并将`observeOn`方法的参数scheduler注入其中。** 现在来看实例化ObservableObserveOn类的时候都做了什么，ObservableObserveOn类的代码如下

public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
        final Scheduler scheduler;
        final boolean delayError;
        final int bufferSize;
        public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
            super(source);
            this.scheduler = scheduler;
            this.delayError = delayError;
            this.bufferSize = bufferSize;
        }
    
        @Override
        protected void subscribeActual(Observer<? super T> observer) {
            if (scheduler instanceof TrampolineScheduler) {
                source.subscribe(observer);
            } else {
                Scheduler.Worker w = scheduler.createWorker();
    
                source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
            }
        }
        //...
        //省略部分代码
    }
    复制代码

可以看到，**实例化ObservableObserveOn类的时候，将ObservableSubscribeOn的实例及`AndroidSchedulers.mainThread()`还有其他的两个默认参数都作为了它的成员变量保存。**

由前面的两篇文章可知，下游的`subscribe`方法最终会调用上游的`subscribeActual`方法，所以会调用这里的`subscribeActual`方法，代码如下

protected void subscribeActual(Observer<? super T> observer) {
            if (scheduler instanceof TrampolineScheduler) {
                source.subscribe(observer);
            } else {
                Scheduler.Worker w = scheduler.createWorker();
    
                source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
            }
        }
    复制代码

现在来一句句的分析上面的代码，首先if条件肯定是不成立的，因为这里`scheduler`其实是`HandlerScheduler`,为什么是HandlerScheduler呢？我们来一点点的分析，由`observeOn(AndroidSchedulers.mainThread())`这句代码可以知道，observeOn方法的参数是`AndroidSchedulers.mainThread()`，那这个`AndroidSchedulers.mainThread()`又是什么呢？看代码

public static Scheduler mainThread() {
            return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
        }
    //上面的方法，返回的就是MAIN_THREAD,而MAIN_THREAD最终返回的是
    //MainHolder.DEFAULT
    private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
                new Callable<Scheduler>() {
                    @Override public Scheduler call() throws Exception {
                        return MainHolder.DEFAULT;
                    }
                });
    //上面的MainHolder.DEFAULT就是实例化了HandlerScheduler
     private static final class MainHolder {
            static final Scheduler DEFAULT
                = new HandlerScheduler(new Handler(Looper.getMainLooper()), false);
        }
    复制代码

上面贴出的代码都是与`AndroidSchedulers.mainThread()`有关的代码，从上面的代码中可以得出结论，`AndroidSchedulers.mainThread()`最终是实例化了`HandlerScheduler`，所以，subscribeActual方法中的scheduler是HandlerScheduler，所以，if语句的条件不成立，这里会执行subscribeActual方法中的else语句，即执行下面的代码

Scheduler.Worker w = scheduler.createWorker();
    
                source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
    复制代码

同样，按照代码的执行顺序来分析，看第一句代码，`Scheduler.Worker w = scheduler.createWorker();`从前文中的分析可以知道，这里的`scheduler`是HandlerScheduler，所以，这里是调用HandlerScheduler类中的`createWorker`方法，HandlerScheduler类中的`createWorker`方法的代码如下

public Worker createWorker() {
            return new HandlerWorker(handler, async);
        }
    复制代码

从上面的代码可以得出，HandlerScheduler类中的`createWorker`方法返回了`HandlerWorker`类的实例，这里传入`HandlerWorker`构造方法中的两个参数是在上面已经分析过的方法中进行初始化的，如下

private static final class MainHolder {
            static final Scheduler DEFAULT
                = new HandlerScheduler(new Handler(Looper.getMainLooper()), false);
        }
    复制代码

从这句代码中可以得知，**HandlerWorker构造方法中的`handler`是实例化在主线程中的Handler，`async`的值是false。** 好了，**到这里我们知道了`Scheduler.Worker w = scheduler.createWorker();`这句代码的作用是实例化了HandlerWorker，而实例化HandlerWorker的同时，在其构造方法中初始化了两个成员变量。**

下面继续看这句代码`source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));`,先看这句代码中的这段`new ObserveOnObserver<T>(observer, w, delayError, bufferSize)`代码做了什么，代码如下

ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
                this.downstream = actual;
                this.worker = worker;
                this.delayError = delayError;
                this.bufferSize = bufferSize;
            }
    复制代码

根据上文的分析可以得出这里的几个参数分别代表什么

*  this.downstream就是这段代码
    
        new Observer<String>() {
                            @Override
                            public void onSubscribe(Disposable d) {
                                Log.e("wizardev", "onSubscribe: "+Thread.currentThread().getName() );
                            }
        
                            @Override
                            public void onNext(String s) {
                                Log.e("wizardev", "接收到上游发射的数据为: " + s);
                                Log.e("wizardev", "下游所在的线程: "+ Thread.currentThread().getName());
                            }
        
                            @Override
                            public void onError(Throwable e) {
        
                            }
        
                            @Override
                            public void onComplete() {
        
                            }
                        }
        复制代码
 *  this.worker就是`new HandlerWorker(handler, async);`
 *  this.delayError的值是false
 *  this.bufferSize就是一个int型的数字

好了，现在继续来看`source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));`这里的`source`就是上游的Observable，这里就是ObservableSubscribeOn类的实例，所以这句代码实际就是调用了ObservableSubscribeOn类中的`subscribe`方法，而ObservableSubscribeOn没有这个方法，所以是调用其父类的`subscribr`方法，由之前的文章可知，最终调用的就是ObservableSubscribeOn类中的`subscribeActual`方法。所以，现在需要把思路切换到ObservableSubscribeOn类中的`subscribeActual`方法了

### ObservableSubscribeOn类中subscribeActual方法分析 ###

还是看代码，subscribeActual方法的如下

public void subscribeActual(final Observer<? super T> observer) {
        //1
            final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
    //2
            observer.onSubscribe(parent);
    //3
            parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
        }
    复制代码

根据前面的分析可知，这个方法中的参数就是`new ObserveOnObserver<T>(observer, w, delayError, bufferSize)`这段代码。照旧，按照代码的执行顺序分析，代码中已经标注了1，2，3的执行步骤，

*  现在来分析“1”处代码，看下SubscribeOnObserver类，代码如下
    
        static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {
        
                private static final long serialVersionUID = 8094547886072529208L;
                final Observer<? super T> downstream;
        
                final AtomicReference<Disposable> upstream;
        //这里的downstream就是new ObserveOnObserver<T>(observer, w, delayError, bufferSize)
                SubscribeOnObserver(Observer<? super T> downstream) {
                    this.downstream = downstream;
                    this.upstream = new AtomicReference<Disposable>();
                }
        
                @Override
                public void onSubscribe(Disposable d) {
                    DisposableHelper.setOnce(this.upstream, d);
                }
        
                @Override
                public void onNext(T t) {
                    downstream.onNext(t);
                }
        
                @Override
                public void onError(Throwable t) {
                    downstream.onError(t);
                }
        
                @Override
                public void onComplete() {
                    downstream.onComplete();
                }
        
                @Override
                public void dispose() {
                    DisposableHelper.dispose(upstream);
                    DisposableHelper.dispose(this);
                }
        
                @Override
                public boolean isDisposed() {
                    return DisposableHelper.isDisposed(get());
                }
        
                void setDisposable(Disposable d) {
                    DisposableHelper.setOnce(this, d);
                }
            }
        复制代码
    
    重要部分已在代码中注释。
 *  接着分析“2”处的代码，这里的observe就是ObserveOnObserver的实例，调用的就是ObserveOnObserver类中的`onSubscribe`方法,`onSubscribe`方法的代码如下
    
        public void onSubscribe(Disposable d) {
            //这里会直接进入if方法中
                    if (DisposableHelper.validate(this.upstream, d)) {
                        //这句代码的作用就是将new SubscribeOnObserver<T>(observer);赋值给了this.upstream
                        this.upstream = d;
        //d的值是SubscribeOnObserver的实例，这里if条件不成立
                        if (d instanceof QueueDisposable) {
                            @SuppressWarnings("unchecked")
                            QueueDisposable<T> qd = (QueueDisposable<T>) d;
                            int m = qd.requestFusion(QueueDisposable.ANY | QueueDisposable.BOUNDARY);
        
                            if (m == QueueDisposable.SYNC) {
                                sourceMode = m;
                                queue = qd;
                                done = true;
                                downstream.onSubscribe(this);
                                schedule();
                                return;
                            }
                            if (m == QueueDisposable.ASYNC) {
                                sourceMode = m;
                                queue = qd;
                                downstream.onSubscribe(this);
                                return;
                            }
                        }
        //实例化大小为bufferSize的队列
                        queue = new SpscLinkedArrayQueue<T>(bufferSize);
        //上文已经分析了downstream的值，最下游的onSubscribe与线程调度
         //无关，在那个线程调用的subscribe就在哪个线程回调
                        downstream.onSubscribe(this);
                    }
                }
        复制代码
    
    主要的代码已在文中注释，下面来分析“3”处的代码
 *  现在一步步的分析“3”处的代码，`new SubscribeTask(parent)`代码如下
    
        final class SubscribeTask implements Runnable {
                private final SubscribeOnObserver<T> parent;
        
                SubscribeTask(SubscribeOnObserver<T> parent) {
                    this.parent = parent;
                }
        
                @Override
                public void run() {
                    source.subscribe(parent);
                }
            }
        复制代码
    
    可以看出SubscribeTask直接实现了的Runnable，并将`new SubscribeOnObserver<T>(observer);`作为成员变量。
    
    继续看`scheduler.scheduleDirect(…)`这里的scheduler是这句代码`Schedulers.io()`,[Schedulers.io][]()代码如下
    
        public static Scheduler io() {
                return RxJavaPlugins.onIoScheduler(IO);
            }
        
        static {
                SINGLE = RxJavaPlugins.initSingleScheduler(new SingleTask());
        
                COMPUTATION = RxJavaPlugins.initComputationScheduler(new ComputationTask());
        //IO是实例化的IOTask
                IO = RxJavaPlugins.initIoScheduler(new IOTask());
        
                TRAMPOLINE = TrampolineScheduler.instance();
        
                NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new NewThreadTask());
            }
        
        static final class IOTask implements Callable<Scheduler> {
                @Override
                public Scheduler call() throws Exception {
                    return IoHolder.DEFAULT;
                }
            }
        //最终会调用这个
        static final class IoHolder {
                static final Scheduler DEFAULT = new IoScheduler();
            }
        
        复制代码
    
    上面的代码可以看出，`Schedulers.io()`最终返回的是IoScheduler，**所以`scheduler.scheduleDirect(…)`这句代码中的scheduler就是IoScheduler**，而`scheduleDirect`方法是IOTask父类中的方法，代码如下
    
        public Disposable scheduleDirect(@NonNull Runnable run) {
                return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
            }
        
        //最终调用的是这个方法
         public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
             //这里的createWorker调用的是IoScheduler中的方法
                final Worker w = createWorker();
        //仍然是Runnable
                final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
        //实例化DisposeTask并将decoratedRun及w注入
                DisposeTask task = new DisposeTask(decoratedRun, w);
        //调用的w的schedule方法，将上面的三个值作为参数
                w.schedule(task, delay, unit);
        
                return task;
            }
        复制代码
    
    这里，详细看下`final Worker w = createWorker();`这句代码，`createWorker()`方法的代码如下
    
        public Worker createWorker() {
                return new EventLoopWorker(pool.get());
            }
        复制代码
    
    可以看到**这句代码的作用是实例化了EventLoopWorker并返回。** 接着看`w.schedule(task, delay, unit);`这句代码，w为EventLoopWorker，所以这里是EventLoopWorker类中的`schedule`方法，代码如下
    
        public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
                    if (tasks.isDisposed()) {
                        // don't schedule, we are unsubscribed
                        return EmptyDisposable.INSTANCE;
                    }
        //最终调用的是NewThreadWorker类中的方法
                    return threadWorker.scheduleActual(action, delayTime, unit, tasks);
                }
        //最终会调用这个方法
        public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
                Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
        //实例化了ScheduledRunnable
                ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);
        
                if (parent != null) {
                    //不会进入这个方法
                    if (!parent.add(sr)) {
                        return sr;
                    }
                }
        
                Future<?> f;
                try {
                    if (delayTime <= 0) {
        //重点，这里是把ScheduledRunnable放进了线程池中，关于java线程
                        //池大家可以自行研究
                        f = executor.submit((Callable<Object>)sr);
                    } else {
                        f = executor.schedule((Callable<Object>)sr, delayTime, unit);
                    }
                    sr.setFuture(f);
                } catch (RejectedExecutionException ex) {
                    if (parent != null) {
                        parent.remove(sr);
                    }
                    RxJavaPlugins.onError(ex);
                }
        
                return sr;
            }
        复制代码
    
    上面中的代码已经有了一些注释，最重要的就是这句代码
    
        if (delayTime <= 0) {
                        f = executor.submit((Callable<Object>)sr);
                    } else {
                        f = executor.schedule((Callable<Object>)sr, delayTime, unit);
                    }
        复制代码
    
    **这段代码的作用就是将任务放进了线程池中等待执行。** 同样，这段代码就是
    
    > subscribeOn是怎样将要处理的数据放到到工作线程的？
    
    这个问题的答案。

通过分析“3”处的代码，可以发现其实就是将SubscribeTask实例进行了一层层的包装，然后丢到线程队列中等待执行，为了便于理解，我画了一下包装层级图，如下

根据上面的层级关系，会发现最后会调到SubscribeTask的`run`方法，这个方法又调用了上游的`subscribe`方法，而这个上游就是`ObservableCreate`类，由于前面两篇已经分析过了这个类，这里就不再分析。根据前面两篇的分析，知道最后会调用发射数据的方法，而这时这个发射数据其实就已经是在工作线程中了。

### 发射数据流程分析 ###

根据前面两篇的分析，可以知道`emitter.onNext("wizardev");`这句代码就是调用下游的`onNext`方法，这里就会调用SubscribeOnObserver类的`onNext`方法，SubscribeOnObserver类的`onNext`方法的源码如下

public void onNext(T t) {
                downstream.onNext(t);
            }
    复制代码

这里直接调用了下游的`onNext`方法，这个下游就是ObserveOnObserver类即这里会调用ObserveOnObserver类中的`onNext`方法,ObserveOnObserver类中的`onNext`方法代码如下

public void onNext(T t) {
         //这里的done为初始值false
                if (done) {
                    return;
                }
    //sourceMode为初始值0
                if (sourceMode != QueueDisposable.ASYNC) {
    //将上游发射的数据放入队列中，这个queue就是在onSubscribe方法中实例化的
                    queue.offer(t);
                }
         //调用方法
                schedule();
            }
    复制代码

继续看`schedule`方法的源码。如下

void schedule() {
                if (getAndIncrement() == 0) {
     //根据上文的分析可以知道这个worker就是HandlerScheduler的内部类
                    //HandlerWorker的实例
                    worker.schedule(this);
                }
            }
    复制代码

这里讲解一下这句`worker.schedule(this);`代码，**这里的`worker`就是HandlerScheduler内部类HandlerWorker的实例，所以这里调用了HandlerScheduler内部类HandlerWorker的`schedule`方法并将ObserveOnObserver实例作为参数传入。** 现在，来看HandlerWorker类中的`schedule`方法，代码如下

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.
    
                if (async) {
                    message.setAsynchronous(true);
                }
    
                handler.sendMessageDelayed(message, 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;
            }
    复制代码

这个方法中重要的就是下面这段代码

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.
    
                if (async) {
                    message.setAsynchronous(true);
                }
    //将message放入messageQueue中等待轮询，这里的handler在主线程中，
    //所以这里执行scheduled的run方法，其实已经切换到主线程中了
                handler.sendMessageDelayed(message, unit.toMillis(delay));
    复制代码

了解Handler原理的同学就会知道上面的这段代码最终会调用scheduled的`run`方法。不了解Handler原理的同学，可以看下我的[这篇文章][Link 1]。这里的scheduled的`run`方法会调用`run = RxJavaPlugins.onSchedule(run);`这句代码的`run`方法,即调用的是ObserveOnObserver类中的`run`方法，看下ObserveOnObserver类中的`run`方法的代码，如下

public void run() {
                if (outputFused) {
                    drainFused();
                } else {
                    drainNormal();
                }
            }
    复制代码

上面代码中的`outputFused`的初始值为false，所以会执行else语句中的代码，看下`drainNormal`的代码，如下

void drainNormal() {
                int missed = 1;
    //这个queue是在onSubscribe初始化的，在onNext中将上游的数据添加进去的
                final SimpleQueue<T> q = queue;
      //将下游的observe赋值给a
                final Observer<? super T> a = downstream;
    //
                for (;;) {
                    if (checkTerminated(done, q.isEmpty(), a)) {
                        return;
                    }
    //开始轮询
                    for (;;) {
                        boolean d = done;
                        T v;
    
                        try {
                 //轮询取出q中的值，这里的值就是在上游发射的
                            v = q.poll();
                        } catch (Throwable ex) {
                            Exceptions.throwIfFatal(ex);
                            disposed = true;
                            upstream.dispose();
                            q.clear();
                            a.onError(ex);
                            worker.dispose();
                            return;
                        }
                        boolean empty = v == null;
    
                        if (checkTerminated(d, empty, a)) {
                            return;
                        }
    
                        if (empty) {
                            break;
                        }
    //取出的值，传递给下游的onNext方法
                        a.onNext(v);
                    }
    
                    missed = addAndGet(-missed);
                    if (missed == 0) {
                        break;
                    }
                }
            }
    复制代码

重要的代码，已经在代码中进行了注释，这里就不再讲解，**上面代码的作用就是，不断取出上游发射的数据，然后调用下游的`onNext`方法并将取出的值传递进去。**

### 总结 ###

分析到这里，算是将RxJava2线程调度的源码理清楚了。可以发现在进行线程调度的时候大量的使用Runnable，一层层的包装，然后在一层层的来调用。首先，将线程切换到工作线程中的方法是将调用上游`subscribe`方法放在了Runnable类中的`run`方法中，然后将这个Runnable层层包装后放进线程队列中等待执行，最后在工作线程中处理发射的数据。

将线程切换到主线程中的方法是利用Handler，将处理好的数据放进一个队列中，放进队列中的这个动作还是在工作线程中完成的，然后，利用Handler将线程切换到主线程，最后不断的取出队列中的数据，不断调用下游的`onNext`方法。通过这种方式来完成线程的调度。

### 结束语 ###

通过上面的分析，可以发现RxJava2线程调度还是挺复杂的，牵涉到的知识点也是比较多的，为了更简单，更有条理的讲解RxJava2线程调度的原理，同时为了让大家不至于在源码中迷失，所以这里分析源码按照代码的执行顺序一步步的进行的。因为，代码执行的时候会在不同的类之间来回切换，所以，大家会发现分析的时候在各个类中跳来跳去。

由于篇幅的原因，文章的一些知识没有详细的来讲解，如Java的线程池，Handler原理等，大家可以自己查阅相关资料或者留言一起讨论，如果发现文中有不对的地方，也欢迎指正。

扫码关注公众号，回复“获取资料”有惊喜

[RxJava2]: https://link.juejin.im?target=http%3A%2F%2Fwizardev.cn%2FRxJava2%25E6%25BA%2590%25E7%25A0%2581%25E5%2588%2586%25E6%259E%2590%25EF%25BC%2588%25E4%25B8%2580%25EF%25BC%2589%25EF%25BC%259A%25E5%259F%25BA%25E6%259C%25AC%25E6%25B5%2581%25E7%25A8%258B%25E5%2588%2586%25E6%259E%2590.html
[RxJava2 1]: https://link.juejin.im?target=http%3A%2F%2Fwizardev.cn%2FRxJava2%25E6%25BA%2590%25E7%25A0%2581%25E5%2588%2586%25E6%259E%2590%25EF%25BC%2588%25E4%25BA%258C%25EF%25BC%2589%25EF%25BC%259A%25E6%2593%258D%25E4%25BD%259C%25E7%25AC%25A6%25E5%258E%259F%25E7%2590%2586%25E5%2588%2586%25E6%259E%2590.html
[Schedulers.io]: https://link.juejin.im?target=http%3A%2F%2FSchedulers.io
[Link 1]: https://juejin.im/post/5aa4ee1f518825556140e39c