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

未完待续

©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容

  • 我从去年开始使用 RxJava ,到现在一年多了。今年加入了 Flipboard 后,看到 Flipboard 的...
    Jason_andy阅读 5,570评论 7 62
  • 前言我从去年开始使用 RxJava ,到现在一年多了。今年加入了 Flipboard 后,看到 Flipboard...
    占导zqq阅读 9,202评论 6 151
  • 先来个RxAndroid的github地址 https://github.com/ReactiveX/RxAndr...
    大批阅读 594评论 0 0
  • 我从去年开始使用 RxJava ,到现在一年多了。今年加入了 Flipboard 后,看到 Flipboard 的...
    huqj阅读 1,886评论 0 21
  • 有时 你像是 一场凶猛的春梦 闯入脑缝 而我则是一滴 早已蒸发成云的泪水 在人间轮回 后半夜的小雨 不算是真的下雨...
    段童阅读 176评论 0 1