重學Android——Rxjava2.x源碼解析

Rxjava的誕生背景

首先要從異步編程提及，最開始的原生的系統中，若是UI系統處理耗時任務，會引起ANR，因此都是放在子線程作耗時任務，好比網絡請求或者IO操做，再來更新UI界面，這須要在主線程來完成，這樣就涉及到了異步編程。

最開始的異步編程主要有：

• 使用Java自身提供的Future模型
  • 但這種異步結果獲取比較困難，必須調用Future.get()，回去查看異步是否完成，若是完成，就返回結果，不然繼續等待。固然在JDK8後，提供了completabelFuture，簡化了異步編程
• Android系統提供的異步模型——AsyncTask。相比於Java提供的方法，此模型無主線程阻塞風險，可是最大的問題是有可能陷入層層嵌套的回調。

Rxjava源碼中鏈式調用

多說也無益，先看源碼。

分析問題時，咱們能夠從特殊到普通來分析，有時候會有意想不到的效果，因此此次源碼由Single開始分析，咱們最簡單的用法：

先在app的gradle中

implementation "io.reactivex.rxjava2:rxjava:2.2.9"
    implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
複製代碼

最簡單的實現

Single.just(1)
                .subscribe(new SingleObserver<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onSuccess(Integer integer) {

            }

            @Override
            public void onError(Throwable e) {

            }
        });
    }
複製代碼

這是最簡單的用法，上游發送一個1的事件，下游接到，不牽涉線程切換。

建立被觀察者

咱們先直接進Just的源碼

@CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    @NonNull
    public static <T> Single<T> just(final T item) {
        //判空
        ObjectHelper.requireNonNull(item, "item is null");
        //HOOK方法
        return RxJavaPlugins.onAssembly(new SingleJust<T>(item));
    }
複製代碼

第一行，其實看方法名咱們也能看出來，是判空的，源碼以下

public static <T> T requireNonNull(T object, String message) {
        if (object == null) {
            throw new NullPointerException(message);
        }
        return object;
    }
複製代碼

果真不出所料，忽略

第二行，先看外層的RxJavaPlugins.onAssembly，進它的源碼

/** * Calls the associated hook function. * @param <T> the value type * @param source the hook's input value * @return the value returned by the hook */
    @SuppressWarnings({ "rawtypes", "unchecked" })
    @NonNull
    public static <T> Single<T> onAssembly(@NonNull Single<T> source) {
        Function<? super Single, ? extends Single> f = onSingleAssembly;
        if (f != null) {
            return apply(f, source);
        }
        return source;
    }
複製代碼

注意看註釋，說明了這是一個hook方法，能夠看到直接return的說是傳入進來的source，因此，咱們能夠得出，Single.just(item)就至關於new SingleJust<T>(item)。

訂閱過程

再來看.subscribe(new SingleObserver<Integer>)的源碼

@SchedulerSupport(SchedulerSupport.NONE)
    @Override
    public final void subscribe(SingleObserver<? super T> observer) {
        //判空
        ObjectHelper.requireNonNull(observer, "observer is null");

        //HOOK
        observer = RxJavaPlugins.onSubscribe(this, observer);

        //繼續判空
        ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null SingleObserver. Please check the handler provided to RxJavaPlugins.setOnSingleSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");

        try {
            //執行當前類的subscribeActual
            subscribeActual(observer);
        } catch (NullPointerException ex) {
            throw ex;
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            NullPointerException npe = new NullPointerException("subscribeActual failed");
            npe.initCause(ex);
            throw npe;
        }
    }
複製代碼

代碼裏有作註釋，其實真正調用的方法是subscribeActual(observer);方法

protected abstract void subscribeActual(@NonNull SingleObserver<? super T> observer);
複製代碼

能夠發現，這是一個抽象方法，那麼咱們要找到它的實現。

咱們回到來看上面的方法其實能夠發現，Single.just()調用的subscribe，而Single.just咱們在上面講到，就至關於new SingleJust()，因此咱們只要看SingleJust裏的subscribeActual方法就能夠了。

public final class SingleJust<T> extends Single<T> {

    final T value;

    public SingleJust(T value) {
        this.value = value;
    }

    @Override
    protected void subscribeActual(SingleObserver<? super T> observer) {
        observer.onSubscribe(Disposables.disposed());
        observer.onSuccess(value);
    }

}
複製代碼

這個類超級簡單，就是把上游的事件發送到下游SingleObserver，好比咱們在實例中，Single.just(1)就至關於new SingleJust(1)，因此在這兒，value=1，而後調用subscribeActual方法，SingleObserver是一個接口，有三個方法，也是咱們回調裏的三個方法

public interface SingleObserver<T> {
    void onSubscribe(@NonNull Disposable d);

    void onSuccess(@NonNull T t);

    void onError(@NonNull Throwable e);
}
複製代碼

在subscribeActual方法中，先observer.onSubscribe(Disposables.disposed());，須要注意的是，這也是just方法獨有的，它直接在onSubscribe方法裏就Disposables.disposed了，這個方法在後面講，這是取消了事件訂閱，由於它只會發一次，到了這就意味着已經不用訂閱了。而後再調用observer.onSuccess方法，直接把value傳遞了過去。

Map操做符的源碼

再來看增長一個操做符的源碼，就用最經常使用的map，其實操做符一通百通

Single.just(1)
                .map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        return integer+2;
                    }
                })
                .subscribe(...);
複製代碼

直接看map的源碼

public final <R> Single<R> map(Function<? super T, ? extends R> mapper) {
        //判空
        ObjectHelper.requireNonNull(mapper, "mapper is null");
        //hook，就至關於new SingleMap
        return RxJavaPlugins.onAssembly(new SingleMap<T, R>(this, mapper));
    }
複製代碼

能夠看到，這就至關於new SingleMap(this,mapper)；返回值依然是Single

咱們看SingleMap的源碼

public final class SingleMap<T, R> extends Single<R> {
    final SingleSource<? extends T> source;

    final Function<? super T, ? extends R> mapper;

    public SingleMap(SingleSource<? extends T> source, Function<? super T, ? extends R> mapper) {
        //這就是剛剛傳進來的this，也就是上游的被觀察者
        this.source = source;
        //這是咱們本身在map中寫的new function方法
        this.mapper = mapper;
    }

    //由上文subscribe方法分析可知，當調用subscribe時，這個回調是會被調用的
    @Override
    protected void subscribeActual(final SingleObserver<? super R> t) {
        //能夠看到，就是至關因而把上游的被觀察者source，直接調用了它的subscribe方法
        //咱們主要的精力只要集中看new MapSingleObserver方法就行
        source.subscribe(new MapSingleObserver<T, R>(t, mapper));
    }

    //此observer觀察者中，把處理後的數據都傳遞給了下游，可是，只提供了事件的流向，由於事件是在上游產生的
    static final class MapSingleObserver<T, R> implements SingleObserver<T> {

        final SingleObserver<? super R> t;

        final Function<? super T, ? extends R> mapper;

        MapSingleObserver(SingleObserver<? super R> t, Function<? super T, ? extends R> mapper) {
            this.t = t;
            this.mapper = mapper;
        }

        @Override
        public void onSubscribe(Disposable d) {
            t.onSubscribe(d);
        }

        @Override
        public void onSuccess(T value) {
            R v;
            try {
                //外面是判空，至關於就是mapper.apply(value)，這個方法其實就是咱們本身的map方法
                v = ObjectHelper.requireNonNull(mapper.apply(value), "The mapper function returned a null value.");
            } catch (Throwable e) {
                Exceptions.throwIfFatal(e);
                onError(e);
                return;
            }
			//將map方法處理後的事件，傳遞給下游
            t.onSuccess(v);
        }

        @Override
        public void onError(Throwable e) {
            t.onError(e);
        }
    }
}
複製代碼

看到這兒咱們能夠發現，事件流向是上游的被觀察者流向觀察者，在操做符中，由於操做符自身是繼承了被觀察者（在此處爲Single），而在其自身中，有一個內部類是觀察者（在此處爲實現了SingleObserver的MapSingleObserver），事件由上游的被觀察者，流向下游的觀察者，而全部的操做符的結構都是同樣的，每一個操做符都只須要給上游操做符提供Observer，並給下游提供一個Observable，內部結構就是，從上游流向下游內部的observer被觀察者，而後此下游的觀察者observable會調用它本身下游的內部observer，這樣，整條鏈就能運行了。

由此可知，Rxjava中，每一個操做符內部都實現了一整套PUSH模型的接口體系。

由特殊到普通

如今回到最普通的Rxjava寫法

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onComplete();
            }
        }).map(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Exception {
                return integer+1;
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(Integer integer) {
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });
複製代碼

先看create方法的源碼

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        ObjectHelper.requireNonNull(source, "source is null");
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }
複製代碼

經過上面的分析，咱們一眼能夠看出，就至關於new ObservableCreate(source)

public final class ObservableCreate<T> extends Observable<T> {
    final ObservableOnSubscribe<T> source;

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

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

        try {
            source.subscribe(parent);//3
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }
	static final class CreateEmitter<T> extends AtomicReference<Disposable> implements ObservableEmitter<T>, Disposable {
        ...
    }
    ...
}
複製代碼

這個類比較長，咱們先只看咱們關心的部分。只以看到咱們喜好的subscribeActual方法，在訂閱時，會調用到此方法。

再來逐句分析，在運行1語句時，new CreateEmitter，看到CreateEmitter的源碼

//實現了ObservableEmitter，ObservableEmitter是Emitter的子類，用於發射上游數據
	static final class CreateEmitter<T> extends AtomicReference<Disposable> implements ObservableEmitter<T>, Disposable {

        private static final long serialVersionUID = -3434801548987643227L;

        final Observer<? super T> observer;

		//下游的observer
        CreateEmitter(Observer<? super T> observer) {
            this.observer = observer;
        }

        @Override
        public void onNext(T t) {
            if (t == null) {
                onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
                return;
            }
            if (!isDisposed()) {
            	//把事件傳遞給下游observer，調用觀察者的onNext方法
                observer.onNext(t);
            }
        }
		...
    }
複製代碼

再回到ObservableCreate的源碼，它是被觀察者Observable的子類，

1. 先在1時new了一個發射器CreateEmitter對象，而後咱們把自定義的下游觀察者observer做爲參數傳了進去，這裏一樣也是包裝起來，這個CreateEmitter實現了ObservableEmitter和Disposable接口
2. 在2語句時，觸發咱們自定義的observer的onSubscribe(Disposable)方法，實際就是調用觀察者的onSubscribe方法，告訴觀察者已經成功訂閱到被觀察者了；
3. 再執行在語句3，source.subscribe(parent);就和咱們分析Map同樣了，就是訂閱，把事件從上游傳到下游。

小結

Observable（被觀察者）和Observer（觀察者）創建鏈接，也就是訂閱以後，會建立出一個發射器CreateEmitter，發射器會把被觀察者中產生的事件發送到觀察者中，觀察者對發射器中發出的事件作出響應事件。能夠看到，訂閱成功以後，Observabel纔會開始發送事件

切斷消息源碼分析

如今咱們再來看dispose的實現。Disposabel是一個接口，能夠理解Disposable是一個鏈接器，調用dispose後，這個鏈接就會中斷。其具體實如今CreateEmitter類。咱們如今主要來分析一下它的這一塊源碼。

在CreateEmitter中的dispose()方法

@Override
        public void dispose() {
            DisposableHelper.dispose(this);
        }
複製代碼

就是調用的DisposableHelper的dispose方法

public enum DisposableHelper implements Disposable {
    /** * The singleton instance representing a terminal, disposed state, don't leak it. */
    DISPOSED
    ;
	...
        
    public static boolean isDisposed(Disposable d) {
        //判斷Disposable類型的變量的引用是否爲DISPOSED
        //就能夠判斷這個鏈接器是否中斷
        return d == DISPOSED;
    }
    
    public static boolean dispose(AtomicReference<Disposable> field) {
        Disposable current = field.get();
        Disposable d = DISPOSED;
        if (current != d) {
            //把field設置爲DISPOSED
            current = field.getAndSet(d);
            if (current != d) {
                if (current != null) {
                    current.dispose();
                }
                return true;
            }
        }
        return false;
    }
    ...
}
複製代碼

能夠看到DisposableHelper是個枚舉類，而且只有一個值DISPOSED。dispose方法就是把一個原子引用的field設爲DISPOSED，這就是中斷狀態。而isDisposed()就是根據這個標誌來判斷是否中斷的。

再回過頭來看CreateEmiiter類的onNext這些方法

@Override
        public void onNext(T t) {
            //省略無關代碼

            if (!isDisposed()) {
                //若是沒有dispose()，纔會調用onNext()
                observer.onNext(t);
            }
        }

        @Override
        public void onError(Throwable t) {
            if (!tryOnError(t)) {
                //若是dispose()了，會調用到這裏，即最終會崩潰
                RxJavaPlugins.onError(t);
            }
        }

        @Override
        public boolean tryOnError(Throwable t) {
            //省略無關代碼
            if (!isDisposed()) {
                try {
                    //若是沒有dispose()，纔會調用onError()
                    observer.onError(t);
                } finally {
                    //onError()以後會dispose()
                    dispose();
                }
                //若是沒有dispose()，返回true
                return true;
            }
            //若是dispose()了，返回false
            return false;
        }

        @Override
        public void onComplete() {
            if (!isDisposed()) {
                try {
                    //若是沒有dispose()，纔會調用onComplete()
                    observer.onComplete();
                } finally {
                    //onComplete()以後會dispose()
                    dispose();
                }
            }
        }
複製代碼

很容易得出，

• 若是沒有dispose，observer的onNext纔會被調用
• onError與onComplete方法互斥，只能其中一個調用到，由於調用其中一個，就會把鏈接切斷，dispose
• 先onError後onComplete中是onComplete不會被調用，反過來的話，就會崩潰，由於onError中拋出了異常，實際上，dispose了後調用onError都會崩

再看一下操做符Map

public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
        ObjectHelper.requireNonNull(mapper, "mapper is null");
        return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
    }

public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
    final Function<? super T, ? extends U> function;

    public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
        super(source);
        this.function = function;
    }

    @Override
    public void subscribeActual(Observer<? super U> t) {
        source.subscribe(new MapObserver<T, U>(t, function));
    }

    static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
        final Function<? super T, ? extends U> mapper;

        MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
            super(actual);
            this.mapper = mapper;
        }

        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }

            if (sourceMode != NONE) {
                downstream.onNext(null);
                return;
            }

            U v;

            try {
                v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }
            downstream.onNext(v);
        }

        ...
    }
}
複製代碼

能夠看到，操做符其實和上面分析的特殊狀況下的同樣的，這裏就省略分析了。

Rxjava線程切換

咱們通常是這麼使用的

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onComplete();
            }
        }).map(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) throws Exception {
                return integer+1;
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Integer>() {
            ...
        });
複製代碼

經過subscribeOn來切換上游線程，observeOn來切換下游線程。

那麼在源碼中，是怎麼的呢？

subscribeOn源碼分析

Schedulers.io()

subscribeOn類型有好幾種，這裏就隨便選擇了Schedulers.io()來分析，別的其實都差很少的，分析了一個就好了。

@NonNull
    public static Scheduler io() {
        //又是hook，就至關於IO
        return RxJavaPlugins.onIoScheduler(IO);
    }

public final class Schedulers {
    ...
    @NonNull
    static final Scheduler IO;
	
    ...

    static final class IoHolder {
        static final Scheduler DEFAULT = new IoScheduler();
    }


    static {
        //又是hook，就至關於new IOTask
        IO = RxJavaPlugins.initIoScheduler(new IOTask());
		...
    }
    
   	...
    static final class IOTask implements Callable<Scheduler> {
        @Override
        public Scheduler call() throws Exception {
            return IoHolder.DEFAULT;
        }
    }
}
複製代碼

能夠看到，最後這裏就至關於new IoScheduler，先不看它的具體實現。

subscribeOn

咱們繼續看subscribeOn的源碼

public final Observable<T> subscribeOn(Scheduler scheduler) {
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
    }
複製代碼

能夠看到和前面同樣，就至關於返回new ObservableSubscribeOn

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
    final Scheduler scheduler;

    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是爲了建立鏈接器，以便之後切斷等控制的，能夠只看裏面
        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }
    ...
}
複製代碼

它的構造就是把source和scheduler兩個都保存一下，在後面要用到的。

接下來咱們來看訂閱過程，雖然這裏是線程切換，可是其實它也只是個操做符，和咱們前面分析的是同樣的，訂閱過程和上面也是同樣的，因此咱們能夠知道，當訂閱發生後，ObservableSubscribeOn的subscribeActual方法就會被調用。

一樣的，subscribeActual方法中，它把咱們自定義的下游觀察者observer包裝成了SubscribeOnObserver對象，而後調用observer的onSubscribe方法，能夠看到，目前爲止，尚未發生任何的線程相關的東西，因此observer的onSubscribe()方法是運行在當前線程中的，那咱們重點來看一下parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));方法。

咱們先來看一下SubscribeTask類

//是ObservableSubscribeOn的內部類，實現runnable接口，看到這，咱們嗅到了線程的味道
	final class SubscribeTask implements Runnable {
        private final SubscribeOnObserver<T> parent;

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

        @Override
        public void run() {
            //這是的source就是咱們自定義的Observable對象，就是ObservableCreate
            source.subscribe(parent);
        }
    }
複製代碼

能夠看到，這個類很是簡單，實現了Runnable接口，在run方法中調用source.subscribe(parent);，這是個鏈式調用，會一層一層調用上去。

再來看scheduler.scheduleDirect

這是線程切換的核心部分了，必定要仔細看

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

	//run就是SubscribeTask
    public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
        //createWorker在Schedule類中是個抽象方法，因此實現是在子類中
        //因此這個方法就是在IOSchedule中實現的
        //worker中能夠執行runnabale
        final Worker w = createWorker();

        //實際上decoratedRun仍是個run對象，也就是SubscribeTask
        final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

        //runnable和worker包裝成一個DisposeTask
        DisposeTask task = new DisposeTask(decoratedRun, w);

        //Worker執行這個Task
        w.schedule(task, delay, unit);

        return task;
    }
複製代碼

上面的代碼註釋已經寫得很是詳細了，scheduleDirect方法就是，new一個worker，而後使用這個worker來執行task線程。

再看一下IoIoScheduler中，createWorker以及shedule的過程

public Worker createWorker() {
        //new一個EventLoopWorker並傳一個worker的緩存池進去
        return new EventLoopWorker(pool.get());
    }

    static final class EventLoopWorker extends Scheduler.Worker {
        private final CompositeDisposable tasks;
        private final CachedWorkerPool pool;
        private final ThreadWorker threadWorker;

        final AtomicBoolean once = new AtomicBoolean();

        EventLoopWorker(CachedWorkerPool pool) {
            this.pool = pool;
            this.tasks = new CompositeDisposable();
            //從緩存worker池中取一個worker出來
            this.threadWorker = pool.get();
        }

        ...
            
        @NonNull
        @Override
        public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
            if (tasks.isDisposed()) {
                // don't schedule, we are unsubscribed
                return EmptyDisposable.INSTANCE;
            }
			
            //Runnable交給threadWorker去執行
            return threadWorker.scheduleActual(action, delayTime, unit, tasks);
        }
    }
複製代碼

注意的是，不一樣的Scheduler類會有不一樣的Worker實現，由於Scheduler類最終都是交由worker來執行調度的，不過度析起來差異不大。

接下來咱們看worker的緩存池操做

static final class CachedWorkerPool implements Runnable {
        ...

        ThreadWorker get() {
            if (allWorkers.isDisposed()) {
                return SHUTDOWN_THREAD_WORKER;
            }
            while (!expiringWorkerQueue.isEmpty()) {
                //緩衝池不爲空，就從緩存池中取一個threadWorker
                ThreadWorker threadWorker = expiringWorkerQueue.poll();
                if (threadWorker != null) {
                    return threadWorker;
                }
            }

            // No cached worker found, so create a new one.
            //爲空就一個並返回去
            ThreadWorker w = new ThreadWorker(threadFactory);
            allWorkers.add(w);
            return w;
        }
		...
    }
複製代碼

再看worker的執行代碼threadWorker.scheduleActual

代碼跟進，會發現實如今它的父類NewThreadWorker中

public class NewThreadWorker extends Scheduler.Worker implements Disposable {
	private final ScheduledExecutorService executor;

    volatile boolean disposed;

    public NewThreadWorker(ThreadFactory threadFactory) {
        //在構造中建立一個ScheduledExecutorService對象
        //能夠經過它來使用線程池
        executor = SchedulerPoolFactory.create(threadFactory);
    }
	...
	public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
        //這是decoratedRun就至關於run
        Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

        //將decoratedRun包裝成一個新對象ScheduledRunnable
        ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);

        if (parent != null) {
            if (!parent.add(sr)) {
                return sr;
            }
        }

        Future<?> f;
        try {
            if (delayTime <= 0) {
                //線程池中當即執行ScheduledRunnable
                f = executor.submit((Callable<Object>)sr);
            } else {
                //線程池中延遲執行ScheduledRunnable
                f = executor.schedule((Callable<Object>)sr, delayTime, unit);
            }
            sr.setFuture(f);
        } catch (RejectedExecutionException ex) {
            ...
        }

        return sr;
    }
 	...   
}
複製代碼

這裏的executor就是使用線程池來執行任務，最終subscribeTask的run方法會在線程池中被執行，即上游的Observable的subscribe方法會在IO線程中調用了。

小結

• Observer的onSubscribe方法運行在當前線程中，由於源碼中並無線程切換
• 若是設置了subscribeOn(指定線程)，那麼Observable中的subscribe方法將會運行在指定線程中。
• 當多個subscribeOn調用時，由於從源碼可知，線程的切換是從下往上的，最後也就是鏈式調用的第一個切換過程，纔是有效的切換

observeOn源碼分析

.observeOn(AndroidSchedulers.mainThread())

AndroidSchedulers.mainThread()

一樣的，咱們先看AndroidSchedulers.mainThread()的源碼

public static Scheduler mainThread() {
        return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
    }
    private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
            new Callable<Scheduler>() {
                @Override public Scheduler call() throws Exception {
                    return MainHolder.DEFAULT;
                }
            });
    private static final class MainHolder {
        static final Scheduler DEFAULT
            = new HandlerScheduler(new Handler(Looper.getMainLooper()), false);
    }
複製代碼

這一段代碼相信若是是看了上面的源碼分析的話，一眼就能看出來，其實就至關於new HandlerScheduler(new Handler(Looper.getMainLooper()), false);，把一個主線程的Handler包裝進了HandlerScheduler中。

observeOn

而後咱們繼續看observeOn的源碼

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");
        return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
    }
複製代碼

經過源碼也能夠知道，這裏至關於直接new 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);
        this.scheduler = scheduler;
        this.delayError = delayError;
        this.bufferSize = bufferSize;
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        //判斷是不是當前線程
        if (scheduler instanceof TrampolineScheduler) {
            //是當前線程的話，直接調用下游的subscribe方法
            //也就是調用下一個Observable的subscibe方法
            source.subscribe(observer);
        } else {
            //建立worker
            //本例中的schedule爲HandlerScheduler
            Scheduler.Worker w = scheduler.createWorker();

            //這裏和上面分析有點相似，會將worker包裝到ObserveOnObserver中
            //注意：source.subscribe沒有涉及到worker，因此仍是在之間設置的線程中執行
            source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
        }
    }
 	...   
}
複製代碼

首先，判斷是否已經在要切換的線程上了，若是是的話，那麼直接調用。若是不是，那麼使用HandlerScheduler包裝一下worker，而後經過worker來把下游的事件進行切換，直接上游訂閱，不作線程操做。

咱們來看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) {
            	//在這裏調用
                worker.schedule(this);
            }
        }

        void drainNormal() {
            int missed = 1;

			//存儲消息的隊列
            final SimpleQueue<T> q = queue;
            //這裏的downstram實際就是下游的observer
            final Observer<? super T> a = downstream;

            for (;;) {
                if (checkTerminated(done, q.isEmpty(), a)) {
                    return;
                }

                for (;;) {
                    boolean d = done;
                    T v;

                    try {
                    	//從隊列中取出消息
                        v = q.poll();
                    } catch (Throwable ex) {
                        Exceptions.throwIfFatal(ex);
                        disposed = true;
                        upstream.dispose();
                        q.clear();
                        a.onError(ex);
                        worker.dispose();
                        return;
                    }
                    boolean empty = v == null;

                    if (checkTerminated(d, empty, a)) {
                        return;
                    }

                    if (empty) {
                        break;
                    }

					//調用下游的Observer的onNext
                    a.onNext(v);
                }

                missed = addAndGet(-missed);
                if (missed == 0) {
                    break;
                }
            }
        }

        @Override
        public void run() {
        	//outputFused默認是false
            if (outputFused) {
                drainFused();
            } else {
            //因此默認調用drainNormal
                drainNormal();
            }
        }
		...
    }
複製代碼

由上面鏈式調用的分析能夠知道，source.subscribe(observer)被調用時，會調用它

CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        observer.onSubscribe(parent);

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

能夠知道，會經過發射器把上游的事件都發送到下游，而後訂閱，因此ObserveOnObserver這個類中，onNext這個方法就會調用，而後執行schedule，最後執行worker.schedule(this);，由於傳入的runnable是this，也就是最後線程後調用到本類的run方法，執行run，最終執行drainNormal()方法。

那咱們再來看一下worker中是怎麼create以及調用的。

在上面咱們已經知道，這個schedule是HandlerSchedule

final class HandlerScheduler extends Scheduler {
    private final Handler handler;
    private final boolean async;

    HandlerScheduler(Handler handler, boolean async) {
        this.handler = handler;
        this.async = async;
    }

    ...

    @Override
    public Worker createWorker() {
       	//主線程的handler
        return new HandlerWorker(handler, async);
    }
}
複製代碼

接着看HandlerWorker的schedule方法

private static final class HandlerWorker extends Worker {
        private final Handler handler;
        private final boolean async;

        private volatile boolean disposed;

        HandlerWorker(Handler handler, boolean async) {
            this.handler = handler;
            this.async = async;
        }

        @Override
        @SuppressLint("NewApi") // Async will only be true when the API is available to call.
        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);

            //把Observer的runnable包裝成ScheduledRunnable
            ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);

            //經過主線程Handler，把message Post到主線程去
            Message message = Message.obtain(handler, scheduled);
            message.obj = this; // Used as token for batch disposal of this worker's runnables.

            if (async) {
                message.setAsynchronous(true);
            }

            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;
        }
        ...
    }
複製代碼

能夠看到，observeOn切換到主線程就是使用了主線程Handler來實現

小結

• 若是設置了observerOn，那麼Observer中的onNext等方法都會被切換到指定線程中
• subscribeOn設置的線程不會影響到observerOn的
• 若是設置了多個observerOn，由於它的切換線程是對下游起做用的，因此最後一個切換的纔有效

擴展

Rxjava是基於PUSH模型構建的數據流，經過上游被觀察者流向下游觀察者，observer只負責消費，observable只負責複雜的數據邏輯，上下游沒有溝通，不上游產生的數據遠大於下游的處理能力的時候，會產生一系列問題，因此rxjava2.x除了Observable/observer核心類以外，還提供了一套flowable/subscribe核心類，它實現了背壓功能，支持上下游速度協調，這一塊等有時間再分析。

關於Rxjava的操做符的各類用法，能夠參考RxJava2 只看這一篇文章就夠了juejin.im/post/5b1756…

---

參考

RxJava的消息訂閱和線程切換原理

---

個人CSDN

下面是個人公衆號，歡迎你們關注我