1. 程式人生 > >RxJava2.0(四)執行緒之間切換的內部原理

RxJava2.0(四)執行緒之間切換的內部原理

基本程式碼

來看一下基本程式碼:

 Observable.create((ObservableOnSubscribe<Integer>) e -> {
            e.onNext(1);
            e.onNext(2);
            e.onComplete();
        }).subscribeOn(Schedulers.io())
          .observeOn(AndroidSchedulers.mainThread())
          .subscribe(i -> System.out.println("onNext : i= "
+ i)); 複製程式碼

很簡單,即訂閱時將task交給子執行緒去做,而資料的回撥則在Android主執行緒中執行。

一、subscribeOn()

點選檢視原始碼:

public final Observable<T> subscribeOn(Scheduler scheduler) {
        //非空判斷和hook
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this
, scheduler)); } 複製程式碼

實際上這個方法返回了一個ObservableSubscribeOn物件。我們有理由猜測這個ObservableSubscribeOn應該和上文的ObservableMap及ObservableDoOnEach相似,都是Observable的一個包裝類(裝飾器):

//1.ObservableSubscribeOn也是Observable的一個裝飾器
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> 
{ final Scheduler scheduler; public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) { //2.儲存上游的ObservableSource和排程器 super(source); this.scheduler = scheduler; } @Override public void subscribeActual(final Observer<? super T> s) { //3.new 一個SubscribeOnObserver final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s); //4.回撥方法,這說明下游的onSubscribe回撥方法所線上程和執行緒排程無關 // 是訂閱時所在的執行緒 s.onSubscribe(parent); //5.立即執行執行緒排程 parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent))); } } 複製程式碼

前兩步我們不需要 再多解釋,直接看第三點,我們看看SubscribeOnObserver這個類:

SubscribeOnObserver

static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {

        private static final long serialVersionUID = 8094547886072529208L;
        //下游的Observer
        final Observer<? super T> actual;
        //儲存上游的Disposable,自身dispose時,連同上游一起dispose
        final AtomicReference<Disposable> s;

        SubscribeOnObserver(Observer<? super T> actual) {
            this.actual = actual;
            this.s = new AtomicReference<Disposable>();
        }

        @Override
        public void onSubscribe(Disposable s) {
            DisposableHelper.setOnce(this.s, s);
        }

        @Override
        public void onNext(T t) {
            actual.onNext(t);
        }

        @Override
        public void onError(Throwable t) {
            actual.onError(t);
        }

        @Override
        public void onComplete() {
            actual.onComplete();
        }

        @Override
        public void dispose() {
            DisposableHelper.dispose(s);
            DisposableHelper.dispose(this);
        }


複製程式碼

類似Observable和ObservableMap,SubscribeOnObserver同樣是Disposable和Observer的一個裝飾器,提供了對下游資料的傳遞,以及將task dispose的介面。

第4步我們之前就講過了,直接看第5步:

    //5.立即執行執行緒排程
        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
複製程式碼

我們看看SubscribeTask這個類:

SubscribeTask

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 介面的實現類而已,通過將SubscribeOnObserver作為引數存起來,在run()方法中添加了上游Observable的被訂閱事件,就沒有了別的操作,

接下來我們看一下scheduler.scheduleDirect(SubscribeTask)中的程式碼:

public abstract class Scheduler {
    //...
    public Disposable scheduleDirect(@NonNull Runnable run) {
        return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
    }

    public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
        // Worker 本身就是Disposable 的實現類
        // 請注意, createWorker()所建立的worker,
        // 實際就是Schdulers.io()所提供的IoScheduler所建立的worker
        final Worker w = createWorker();

        //hook相關
        final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

        DisposeTask task = new DisposeTask(decoratedRun, w);

        //即 worker.schedule(task, 0, TimeUnit.NANOSECONDS): 立即執行task
        w.schedule(task, delay, unit);

        return task;
    }
    //...
}

複製程式碼

我們不要追究過深,我們看一下這個createWorker方法的註釋說明:

/**
     * Retrieves or creates a new {@link Scheduler.Worker} that represents serial execution of actions.
     * 檢索或建立一個新的{@link Scheduler.Worker}表示一系列的action
     * 
     * When work is completed it should be unsubscribed using {@link Scheduler.Worker#dispose()}.
     * 當work完成後,應使用{@link Scheduler.Worker#dispose()}取消訂閱。
     * 
     * Work on a {@link Scheduler.Worker} is guaranteed to be sequential.
     *  {@link Scheduler.Worker} 上面的work保證是順序執行的
     */

複製程式碼

現在我們知道了:我們通過呼叫subscribeOn()傳入Scheduler,當下遊ObservableSource被訂閱時(請注意,訂閱順序是由下到上的),距離最近的執行緒排程subscribeOn()方法中,儲存的Scheduler會建立一個worker(對應相應的執行緒,本文中為IoScheduler),在其對應的執行緒中,立即執行task

多次subscribeOn()

現在考慮一個問題,假如在我們的程式碼中,多次使用了subscribeOn()程式碼,到執行緒會怎麼處理呢?

上文已經講到了,不管我們怎麼通過subscribeOn()方法切換執行緒,由於訂閱執行順序是由下到上,因此當最上游的ObservableSource被訂閱時,所線上程當然是距離上游最近的subscribeOn()所提供的執行緒,即最終Observable總是在第一個subscribeOn()所在的執行緒中執行。

二、observeOn()

先看observeOn()內部,果然是hook+Observable的包裝類:

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

        //例項化ObservableObserveOn物件並返回
        return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
    }

複製程式碼

再看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);
        //1.相關依賴注入
        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 {
            //2.建立主執行緒的worker
            Scheduler.Worker w = scheduler.createWorker();
            //3.上游資料來源被訂閱
            source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
        }
    }
}

複製程式碼

和subscribeOn()不同的是,我們並不是立即在對應的執行緒執行task,而是將對應的執行緒(實際上是worker)作為引數,例項化ObserveOnObserver並存儲起來。

當上遊的資料傳遞過來時,ObserveOnObserver執行對應的方法,比如onNext(T),再切換到對應執行緒中,並交由下游的Observer去接收:

ObserveOnObserver

ObserveOnObserver中程式碼極多,我們簡單瞭解原理後,以onNext(T)為例:

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.actual = actual;
            this.worker = worker;
            this.delayError = delayError;
            this.bufferSize = bufferSize;
        }
        //佇列
        SimpleQueue<T> queue;

       @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            //將資料存入佇列
            if (sourceMode != QueueDisposable.ASYNC) {
                queue.offer(t);
            } 
            //對應執行緒取出資料並交由下游的Observer
            schedule();
        }

        void schedule() {
            if (getAndIncrement() == 0) {
                worker.schedule(this);
            }
        }
         //...省略其他程式碼
}

複製程式碼

多次observerOn()

由上文得知,與subscribeOn()相反,observerOn()操作會將切換到對應的執行緒,然後交由下游的Observer處理,因此observerOn()僅對下游的Observer生效,並且,如果多次呼叫,observerOn()的執行緒排程會持續到下一個observerOn()操作之前。

總結

subscribeOn()

  • 訂閱順序當從下到上,上游的ObservableSource被訂閱時,先切換執行緒,然後立即執行task;

  • 當存在多個subscribeOn()方法時,僅第一個subscribeOn()有效。

observerOn()

  • 訂閱順序當從下到上,上游的ObservableSource被訂閱時,會將對應的worker建立並作為構造引數儲存在Observer的裝飾器中,並不會立即切換執行緒;

  • 當資料由上游傳送過來時,先將資料儲存到佇列中,然後切換執行緒,然後在新的執行緒中將資料傳送給下游的Observer;

  • 當存在多個observerOn()方法時,僅對距下游下一個observerOn()之前的observer有效

有興趣可以關注我的小專欄,學習更多知識:小專欄