RXJava2响应式编程框架设计<二>---Rxjava2线程切换原理再梳理

在上一次https://www.cnblogs.com/webor2006/p/12329139.html中对于RxJava2进行了一些简单入门,这次则从源码的角度对它的原理进行一个分析,其中重点是分析线程切换的原因,这个在之前的https://www.cnblogs.com/webor2006/p/10545699.html已经也分析得比较透彻了,这里相当于再查漏补缺吧。

咱们以这个样式的代码进行原理剖析:

just():

点击进去看一下它的源码:

 

其中Observable类中有各种各样的操作符,它的代码量极大:

好,回到主流上来:

这句话的核心其实就是ObservableJust,在之前的分析对于RxJavaPlugins.onAssembly()一带而过了,其实它是一个hook函数【关于hook函数的介绍可以参考:https://www.cnblogs.com/webor2006/p/12268799.html】,何以见得?

那这次来看一下这个勾子函数是如何发挥作用的?

其中onObservableAssembly成员变量默认肯定是为null的,所以最终直接将咱们的source给返回的:

也就是:

好,那如果想要勾子函数发挥作用呢?当然得要来设置勾子函数才行喽,所以咱们来看一下勾子函数的赋值是在哪?

好,回到主流程来,经过这个just()操作符之后,目前创建了一个ObservableJust对像,看一下它的结构:

而Observable又实现了一个接口:

好的,继续往下。

subscribeOn():

这里有个关系需要注意哈!!就是此时的Observable对象是通过just()调用返回的ObservableJust哈,由于所有的操作符都是定义在了父类Observable上了,所以我们在链接每一个操作符的源码时都会链到Observable类而非真实的Observable的子类,这个一定得要记住,知道这层关系才有助于我们更好的理解其原理,所以:

接着再来看一下这个包装类ObservableSubscribeOn,猜它又是一个包装了Observable的子类: 

好,回到主流程来,接下来还有一个参数还没有分析:

各处都能看到勾子函数,此时再来看一下IOTask:

此时就将这上Scheduler作为第二个参数传到Observable中:

好,这个操作符先分析到这,继续再来分析。

observeOn():

其中也传来一个Scheduler,来瞅一下这个:

此时这个Scheduler是在rxandroid库中:

好,再回到主流程中来看操作符内部的细节:

此时再来看一下ObservableObserveOn类:

其中这个source很显然是调用了subscribeOn()生成的那个包装被观察者ObservableSubscribeOn。

好,目前来看一下经过了observeOn()操作符调用之后的被观察者的类层次结构:

subscribe(new Observer()) :

ObservableObserveOn.subscribeActual():

接下来则开始订阅,将被观察者与观察者进行关联:

发现它是一个接口:

咋感觉这好像是一个接口回调呢?

好,究竟这个观察者的回调方法是如何调用的,接着来分析最核心的订阅方法了:

这里先再提醒一下,此时的Observable是指observeOn()操作符所返回的ObservableObserveOn被观察者包装类,要时刻清楚这个观察者是谁,好继续往下分析:

此时则回到ObservableObserveOn中进行方法的分析:

此时的scheduler成员变量应该是它返回的:

也就是:

所以此时则会走else条件:

关于这块的代码在之前RxJava原理剖析中已经看过了,这里再来梳理一遍:

此时转到HandlerScheduler:

而HandlerWorker定义如下:

    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.

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

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

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

好,回到主流程来往下:

此时先来看一下参数,将创建的worker对象包装到一个新的观察者对象中:

此时就得看一眼ObserveOnObserver观察者对象了:

除了实现了两个接口之外,还继承了一个BasicIntQueueDisposable,看到Disposable关键字了,很眼熟嘛,因为在观察者的回调中就有它:

先看一下它的类继承体系:

好,回到主流程:

它的子类为抽象的被观察者:

看一下它的subscribe()的实现:

接下来则又要调用子类的subscribeActual()方法,而当前的子类又是谁呢?得看一下这个source是谁?

很显然它是调用上一个操作符所生成的被观察者对象:

调用了这个操作符所返回的被观察者,也就是:

ObservableSubscribeOn.subscribeActual():

所以,此时转到这个类中来查看一下订阅的详情:

1、包装观察者:

此时看一下该包装观察者:

其中它里面的观察者actual构造参数是来自于:

 

2、调用观察者的onSubscribe()方法:

也就是会调用到这:

此时来看一下这个方法的核心代码:

而其中actual则是:

也就是:

呃,貌似跟我以前的理解有出入,我一直以为:

 

所以为了验证咱们所分析的,我们用实验来确认一下:

运行结果:

呃,颠覆了我的认知,确实是在主线程,好,那如果修改一下程序再来看一下:

 

也就是确实如源码有分析的那样,这个onSubscribe是处于当前线程,那当前线程是以哪个为基准呢?是它:

3、设置被观察者的运行线程:

这里的参数套得比较深,从最里层开始分析:

注意将第一步骤封装的parent观察者给传到递了SubscribeTask当中了:

其中它的run方法中是调用的source的subscribe()方法了,那这个source是啥呢?很明显是它上一个操作符所创建的被观察者:

它的上一个被观察者为:

也就是这个操作符所创建的被观察者:

所以,此时的source为ObservableJust:

至于这个run()是何时调用的目前还不得而知,好,回到主流程再来往外层分析:

其中scheduler是从构造方法中传过来的:

而它是从这传过来的:

它为:

也就是:

分析了这么多,这里用图来对其以上流程进行一个总结:

Scheduler.scheduleDirect():

接下来的流程得看好了,切线程的逻辑就在其中,这里看一下scheduleDirect方法的细节:

上面的这个代码就是切线程的核心啦,好好细品,注意这个方法的Runnable对象则为:

好接下来分析一下scheduleDirect()方法:

1、创建一个Worker线程:

2、钩子函数,若无扩展性特殊处理则返回参数本身:

这个代码就没啥可说的,默认直接就是我们参数的Runnable,再来强调一下这个run参数是它:

3、声明一个处理任务,将Runnable和Worker封装成DisposeTask:

其中在它的run()方法中就调用了咱们的那个DisposeTask的run()方法了,如下:

那这个新封装的DisposeTask又何时调用呢?继续往下分析: 

4、调取Worker对象的schedule()方法:

此时就会调用Worker的schedule()方法了,将新创建的DisposeTask传进去,跟进去瞅一下它的细节:

此时又得来看具体类的实现了:

此时需要注意!

它是咱们新创建的Task,如下:

由于层次非常之深,也很容易看晕,关键的东东一定得要记清楚,这样分析才不会晕,好,此时再往下跟:

此时看一下ScheduledRunnable:

其中看一下executor是否是线程池,确认一下:

还是回到主流程继续,我们还得回溯本源,还没完哈:

此时的actual为在这块新创建的Task,如下:

好,此时它的run()就得到执行了:

这里再来啰嗦一下:

也就是:

那此时看一下SubscribeTask的run()方法:

ObservableJust.subscribeActual():

此时则转到ObservableJust来:

接下来则来分析这个方法的流程,先来明白下这个参数s是啥?很明显是SubscribeOnObserver,也就是:

然后跟进去瞅一眼ScalarDisposable:

好回到主流程:

 

此时:

 

此时则会回调:

此时的actual为:

所以流程得会流转到这:

此时的worker还记得是啥不?说实话分析到这也晕晕的了,这里再来回忆一下:

由于我们是指定的UI线程:

所以createWorker()的细节在:

然后再看一下它里面的schedule()方法的逻辑:

注意在生成ScheduledRunnable对象时的第二个run参数,其实是传的它:

很明显已经开始切到UI线程了,最终会执行ScheduledRunnable.run()方法:

 

此时又会调回到这个地方,注意此时已经切到了UI线程了哈:

好熟悉的三个回调呀,那此时看一下actual又是谁?其实是它:

 

也就是我们的匿名观察者类:

好,这里再回到这块分析一下落掉的这句代码:

此时的s是啥来着?

所以,此时转到它的onSubscribe()方法中:

 

也就是说:

其实subscribeOn()是来决定被观察者的subscribe()事件发送的线程的,下面来看一下:

输出:

此时咱们利用subscribeOn()切换一下线程:

再运行:

最后稍加总结一下关于SubscribeOnObserver的流程:

 

至此,关于整个线程切换的完整流程就已经分析完了,太tm复杂了,真佩服写这框架的人是啥神脑,太逆天了,真是晕晕的~~不过这次完整走一遍之后,确实对于整个RxJava的原理的了解又更进一步了,时间花得还是值得的。

原文地址:https://www.cnblogs.com/webor2006/p/12348890.html