rxjava2源码解析(二)线程切换分析

795 阅读8分钟

引言

先贴上系列的链接:
rxjava2源码解析(一)基本流程分析
rxjava2源码解析(二)线程切换分析
rxjava2源码解析(三)线程池原理分析

上一篇介绍了rxjava2源码解析(一)基本流程分析,这篇准备说说线程切换。

使用方法

还是先从最基本的使用开始看:

        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("1");
                emitter.onNext("2");
                emitter.onNext("3");
                emitter.onNext("4");
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(Schedulers.newThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        Log.d(TAG, "s = " + s);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });

相较于前面的实例,多了两段代码,subscribeOnobserveOn。我们知道,subscribeOn是用来调整被观察者(发射源)的线程,而observeOn是调整观察者(处理器)的线程。

observeOn

我们先从observeOn的源码来看它是如何控制处理器的线程:

    @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) {
        //判空代码,和hock相关机制,我们可以忽略,直接看ObservableObserveOn
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
        return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
    }
    

首先我们知道,observeOnsubscribeOn都是observable这个装饰器的方法,他们的返回值也都是observable(前面讲过这是装饰器模式)。这里还是老样子,直接看ObservableObserveOn这个对象就行了。

    public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
        //将上游source存在本地
        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));
        }
    }

重点还是subscribeActual这个方法。我们看到,初始化ObservableObserveOn的时候传入了我们设置的scheduler。所以在subscribeActual里,先判断scheduler是否是TrampolineSchedulerTrampolineScheduler是什么东西呢?我们看官方注释:

/**
 * Schedules work on the current thread but does not execute immediately. Work is put in a queue and executed
 * after the current unit of work is completed.
 *计划在当前线程上工作,但不会立即执行。 将工作放入队列并在当前工作单元完成后执行。
 */

OK,一目了然,如果是当前线程,不做任何处理,直接用绑定起来。否则,新建一个ObserveOnObserver对象,将上游装饰器(这里的上游是代码流程上的上游,即调用observeOn的装饰器)先于这个对象绑定。这个看起来是不是很眼熟?我们回看上一篇里面说的ObservableCreate里的subscribeActual方法,对比一下有什么不同。

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        observer.onSubscribe(parent);

        try {
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }

可以看到,两者都有在内部新建一个对象,并将上游装饰器(或者是发射源)与之绑定。区别在于,ObservableCreate内部是新建一个发射器CreateEmitter对象,而ObservableObserveOn内部是新建一个处理器ObserveOnObserver对象。ObservableCreate是将之前存储的上游发射源与发射器绑定,ObservableObserveOn是将上游装饰器与处理器绑定。
根据这一点,我们可以将装饰器分为两种样式:

  • 一种是跟ObservableCreate类似,属于每一条流水线的开端,本身是装饰器,上游是发射源,内部生成一个发射器,处理最开始的发射流程。我们称之为起始装饰器
  • 一种是跟ObservableObserveOn类似,属于流水线中间流程,本身是装饰器,上游是装饰器,内部新建一个处理器来处理上游事件,下游是处理器或者其他装饰器。我们称之为流程装饰器。

下面我们可以看看ObserveOnObserver的源码。

    static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
    implements Observer<T>, Runnable {
        ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
            this.downstream = actual;
            this.worker = worker;
            this.delayError = delayError;
            this.bufferSize = bufferSize;
        }
        
        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            //同步异步相关,暂时不管
            if (sourceMode != QueueDisposable.ASYNC) {
                queue.offer(t);
            }
            //重点是这个
            schedule();
        }
        
        void schedule() {
            if (getAndIncrement() == 0) {
                //ObserveOnObserver继承了runnable接口,意味着可以当做是线程任务来执行。这里代表着在新线程中执行run方法。
                worker.schedule(this);
            }
        }
        //ObserveOnObserver继承了runnable接口
        @Override
        public void run() {
        //同步异步相关,我们直接看drainNormal()
            if (outputFused) {
                drainFused();
            } else {
                drainNormal();
            }
        }
        
        void drainNormal() {
            int missed = 1;
            final SimpleQueue<T> q = queue;
            final Observer<? super T> a = downstream;
            for (;;) {
                if (checkTerminated(done, q.isEmpty(), a)) {
                    return;
                }
                for (;;) {
                    ····//省略一些判断的代码
                    v = q.poll();
                    //这里就可以看到,将下游的onNext方法,切换到新线程执行。
                    a.onNext(v);
                }
                ···
            }
        }
        
    }

这里我们就不深入细究了(会在下一篇中详细来说),只需要知道,这是上游的处理器执行onNext,传到这里,使用之前设置的线程执行下游的onNext方法。所以这里就完成了线程切换功能。并且这个切换,延续到下游所有处理的onNext。如果在下游再次调用ObserverOn,就会将后面的处理器切换到另外一个线程。
所以,我们可以得到结论,ObserverOn可以多次调用,每次调用会作用于下游的所有处理器,直到遇到新的ObserverOn

subscribeOn

接下来可以看看,SubscribeOn的源码内容。

    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.CUSTOM)
    public final Observable<T> subscribeOn(Scheduler scheduler) {
        //判空和hock机制代码,忽略。直接看ObservableSubscribeOn
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
    }

一样的配方,一样的味道,我们可以直接看ObservableSubscribeOn类的subscribeActual方法。

    @Override
    public void subscribeActual(final Observer<? super T> observer) {
        //创建SubscribeOnObserver内部类对象
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);
        //这里调用了下游的onSubscribe
        observer.onSubscribe(parent);
        //scheduler.scheduleDirect方法返回一个disposable
        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }

这里可以看到,ObservableSubscribeOn同样是一个流程装饰器,在调用subscribe的时候,内部新建一个处理器,这个处理器与下游处理器相互持有。这里与ObservableObserveOn有所不同,并没有立即执行subscribe将上游装饰器与内部处理器连接起来,而是执行了:

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

我们看看SubscribeTask是啥:

final class SubscribeTask implements Runnable {
        private final SubscribeOnObserver<T> parent;

        SubscribeTask(SubscribeOnObserver<T> parent) {
            this.parent = parent;
        }

        @Override
        public void run() {
            //这里的source是上游装饰器,parent是内部处理器
            source.subscribe(parent);
        }
    }

很简单,SubscribeTask继承了Runnable,其中run方法是执行上游装饰器的subscribe方法。我们再看scheduleDirect方法。

    @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方法,是每个不同的scheduler自行重写的。我们不深究这块的代码,只需要理解为,在subscribeActual中,根据前面设置的scheduler,新建一个线程,在线程中立即执行上游sourcesubscribe方法,与内部处理器SubscribeOnObserver绑定。SubscribeOnObserver处理器与一般处理器没什么区别,就不贴代码了。
那么上面就实现了,在另外一个线程执行subscribe方法。看过前面基本流程的都知道,这里基本上就确定了,上游所有subscribe方法执行的线程。所以我们知道了,subscribeOn()方法会切换上游所有的subscribe方法,至于发射源所在的线程,只跟离它最近的subscribeOn()方法中所切换的线程有关。所以说,subscribeOn()方法只需要执行一次,且只有第一次是生效的。

onSubscribe

我们看到,终端Observer这里有一个onSubscribe方法,我们一般在这里进行一些初始化的操作,而前面的源码中也有很多地方有onSubscribe方法。那这个方法到底是执行在哪个线程呢?我们从源码中寻找答案。
想要知道onSubscribe方法在哪个线程,只需要看,在离终端处理器最近的上游装饰器中,是在哪个线程调用onSubscribe()的。我们来看看subscribeOnobserveOn有没有更改它所运行的线程。 先看ObserveOn,毕竟这是作用于下游处理器的线程切换方法。

    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类中的subscribeActual并没有onSubscribe相关的内容。我们看看ObserveOnObserver的源码:


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

        @Override
        public void onSubscribe(Disposable d) {
            if (DisposableHelper.validate(this.upstream, d)) {
                this.upstream = d;
                ·····//不相干代码
                //这里的downstream是指下游处理器
                downstream.onSubscribe(this);
            }
        }

我们看到,在ObserveOnObserver被调用onSubscribe的时候,会调用下游的onSubscribe,参数是本身。也就是说,ObserveOn并没有切换onSubscribe方法的线程。
再看subscribeOn方法,很明显是直接在subscribeActual中执行下游处理器的onSubscribe方法。

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

这两个线程切换的方法,都没有更改onSubscribe()方法的线程。所以我们能确定,在终端的处理器Observer里面的onSubscribe()方法,是跟外部在同一个线程上。

总结一下

  • observeOn作用于下游的所有处理器,可以多次调用。每一个处理器所运行的线程,决定于它最近的上游observeOn方法中指定的线程。
  • subscribeOn作用于上游的发射源,主要是用来指定subscribe方法所在的线程。针对于发射源,只有离它最近的下游subscribeOn方法中所指定的线程才生效。所以subscribeOn方法多次调用并没有效果。
  • onSubscribe()方法并不会跟随内部线程切换而切换线程。运行在哪个线程,只跟外部创建这一整套观察者模式的线程一致。

OK,上面就是本篇的全部内容了,下一篇我们深入看一看,rxjava2 里面的线程池和线程切换到底是怎么实现的!

每周更新,敬请期待~