RxJava2源码分析-线程调度

先上被分析的代码：

      val observer = object : Observer<String> {
            override fun onSubscribe(d: Disposable) {
                Log.d(TAG, "onSubscribe")
                Log.d(TAG, "Observer work thread is" + Thread.currentThread().name)
                disposable = d
            }

            override fun onNext(s: String) {
                Log.d(TAG, "Observer onNext data is :$s")
                Log.d(TAG, "Observer work thread is" + Thread.currentThread().name)
                if (s == "hello") {
                    //执行了hello之后终止
                    disposable.dispose()
                }
                val compositeDisposable = CompositeDisposable()
                compositeDisposable.dispose()

            }

            override fun onError(e: Throwable) {
                Log.d(TAG, "Observer onError data is :" + e.toString())
            }

            override fun onComplete() {
                Log.d(TAG, "Observer onComplete")
            }
        }

        val observable = Observable.create(object : ObservableOnSubscribe<String> {
            @Throws(Exception::class)
            override fun subscribe(e: ObservableEmitter<String>) {
                Log.i(TAG, "Observable work thread is" + Thread.currentThread().name)
                e.onNext("hello")
                Log.i(TAG, "Observable 发送 hello")
                e.onNext("world")
                Log.i(TAG, "Observable 发送 world")
                e.onComplete()
                Log.i(TAG, "Observable 调用 onComplete")
            }
        })
        observable.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(observer)

这次我们倒着分析，在调用.subscribe(observer)方法的时候，最终会调用调用方的subscribeActual方法，调用方是observeOn方法返回的Observable，查看observeOn方法内部，发现返回的是ObservableObserveOn类，然后我们看ObservableObserveOn类的subscribeActual方法：

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

if (scheduler instanceof TrampolineScheduler)先跳过，我们的scheduler也不是TrampolineScheduler，所以执行的是else下面的代码，Scheduler.Worker w = scheduler.createWorker();获取到的就是observeOn方法中传入的Scheduler，我们这里传入的是AndroidSchedulers.mainThread()，而source是上游的Observable（也就是调用subscribeOn返回的Observable），source.subscribe传入的参数是包装后的Observer，其中包含了当前的scheduler，然后source最终也会执行它自己的subscribeActual方法，而当前source是ObservableSubscribeOn，所以我们看ObservableSubscribeOn类的该方法：

    @Override
    public void subscribeActual(final Observer<? super T> s) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

        s.onSubscribe(parent);

        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }

然后就和上一篇文章一样了

注意：如果有多个observeOn方法，通过ObservableObserveOn类的subscribeActual方法可知，会从下到上依次订阅observer

未完待续