RxJava2原始碼分析

NO IMAGE
1 Star2 Stars3 Stars4 Stars5 Stars 給文章打分!
Loading...

RxJava2原始碼分析

RxJava的鼎鼎大名相信Android開發的同學都非常熟悉了,其實不僅僅有RxJava,還有RxJs,RxKotlin等等一系列。可以說Rx並不是一種侷限於Android的框架,Rx是一種思想,我們深入瞭解了RxJava,同樣會加深我們對其他Rx系列的認知。

使用方法

我們來看一個常見的例子:

Observable.create(ObservableOnSubscribe<Int> { e ->
            e.onNext(1)
            e.onComplete()
        }).map {
            Log.d("map-thread : ", Thread.currentThread().name)
            Log.d("--", "------------------------------")
            "result : $it"
        }.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribeWith(object : Observer<String> {
            override fun onSubscribe(d: Disposable) {
                Log.d("onSubscribe-thread : ", Thread.currentThread().name)
                Log.d("--", "------------------------------")
            }

            override fun onNext(s: String) {
                Log.d("onNext-thread : ", Thread.currentThread().name)
                Log.d("onNext-result", s)
                Log.d("--", "------------------------------")
            }

            override fun onError(e: Throwable) {
                Log.d("onError-thread : ", Thread.currentThread().name)
                Log.d("onError-message : ", e.message)
                Log.d("--", "------------------------------")
                Log.d("--", "------------------------------")
                Log.d("--", "------------------------------")
            }

            override fun onComplete() {
                Log.d("onComplete-thread : ", Thread.currentThread().name)
                Log.d("--", "------------------------------")
                Log.d("--", "------------------------------")
                Log.d("--", "------------------------------")
            }
        })

這是一個使用Kotlin寫的例子,對Kotlin不熟悉的同學無需關注程式碼細節,大致能看懂什麼意思就行。首先傳送一個數字1,然後通過 map 操作符把數字1變成 result : 1 ,將之前的操作切換到IO執行緒,將之後的操作切換到主執行緒。

整個RxJava的使用可以分為三個部分:

  • 建立傳送資料的原始Observable
  • 使用Rxjava的操作符對傳送事件做相應變換
  • 使用subscribeOn和observeOn做執行緒切換

接下來我會對這三個部分做詳細的說明。

建立傳送資料的原始Observable

建立傳送資料的原始Observable採用的是 Observable.create() ,當然使用 Observable.just(1)Observable.from(list) 等等這些方法也可以建立傳送資料的Observable,其實這些操作符的本質都是建立了一個新的Observable,在訂閱的時候傳送了一些資料。我們著重看一下 Observable.create()

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        //確保source非空
        ObjectHelper.requireNonNull(source, "source is null");
        //返回一個ObservableCreate物件
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }

第一句很好理解,確保傳入的 ObservableOnSubscribe 物件非空,第二句本質是直接返回了 new ObservableCreate<T>(source) 執行的結果,我們以後在看到 RxJavaPlugins.onAssembly(obj) 類似的程式碼時,直接可以理解為返回了一個 obj 。所以這個方法實際上做了一件事:建立了一個 ObservableCreate 型別的物件並返回。

這裡我們先了解一下在訂閱時,也就是呼叫 subscribeWith(observer) 時具體是做了什麼:

    public final <E extends Observer<? super T>> E subscribeWith(E observer) {
        subscribe(observer);
        return observer;
    }
    
    public final void subscribe(Observer<? super T> observer) {
        //確保observer非空
        ObjectHelper.requireNonNull(observer, "observer is null");
        try {
            //這裡一般都是直接返回傳入的observer
            observer = RxJavaPlugins.onSubscribe(this, observer);
            ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
            
            //不同型別的Observable的具體訂閱的方法,所有的資料傳送操作都在這個方法中
            subscribeActual(observer);
        } catch (NullPointerException e) { // NOPMD
            throw e;
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            // can't call onError because no way to know if a Disposable has been set or not
            // can't call onSubscribe because the call might have set a Subscription already
            RxJavaPlugins.onError(e);

            NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
            npe.initCause(e);
            throw npe;
        }
    }

先看第一個方法 subscribeWith(E observer) ,在這個方法內部呼叫了 subscribe(observer) 這個方法。 subscribe(observer) 這個方法的核心是 subscribeActual(observer) 這一句。subscribeActual(observer) 這個方法是Observable內的一個抽象方法,Observable經過create操作符後會變成一個 ObservableCreate,經過map操作符會變成一個 ObservableMap 其他的操作符都是一個套路。經過操作符轉化而來的Observable都是Observable的子類,在這些子類的內部都會實現 subscribeActual(observer) 這個抽象方法,並在這個方法內做傳送事件的相關操作。

經過上面的分析,我們知道了 subscribeActual(observer) 這個方法是不同Observable的關鍵,有了這個前置知識,我們接著上面的進度看一下 ObservableCreate 這個Observable子類中的 subscribeActual(observer) 方法:

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

        try {
            //這是關鍵
            source.subscribe(parent);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            parent.onError(ex);
        }
    }

關鍵是 source.subscribe(parent) 這一句。這個source就是 ObservableOnSubscribe 的例項,是在 Observable.create(source) 中傳入的引數。 ObservableOnSubscribe 是一個介面,在這個介面的 subscribe 方法中我們定義了資料的傳送方式。這個 ObservableOnSubscribe 可以理解為一個 資料傳送的劇本 ,這個劇本具體的實現細節寫在了 subscribe方法內。在最上面的例子中,我們通過 e.onNext(1);e.onComplete() 傳送了一個數字1,緊接著通知事件已經傳送完畢。

使用RxJava的操作符對傳送事件做相應變換

由於RxJava的事件操作符太多,我們這裡只講map操作符的原始碼分析:

    public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
        //確保mapper非空
        ObjectHelper.requireNonNull(mapper, "mapper is null");
        //建立並返回一個ObservableMap物件
        return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
    }

是不是很熟悉,map操作符和create操作符是一個套路,事實上所有的操作符都是一個套路: 建立並返回了一個對原有Observable的包裝類 ,那我們來看一下這個 ObservableMap

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

    //對原始observer的包裝類
    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) {
            U v;
            try {
                //在這裡將T型別的原始資料t變換為型別U的新資料v
                v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }
            //呼叫原始observer的onNext
            actual.onNext(v);
        }
        
        //忽略無關程式碼......
    }

ObservableMap 內部有有兩個值得我們注意的點:一個是我們上面提過的 subscribeActual 方法;另一個是它的內部類 MapObserver

subscribeActual 中依然是 source.subscribe(new MapObserver<T, U>(t, function)) 這句熟悉的程式碼,值得注意的是這句程式碼並沒有傳入原始的observer,而是傳入了 MapObserver 物件。簡而言之, MapObserver 是一個包裝類,它的內部包含我們原始的observer: actual 和我們變換的具體操作: mapper 。新的 MapObserver 包裝類在它的 onNext 方法中做了兩步操作:

  • 通過 mapper.apply(t) 實現資料的型別變換,獲取新的變換後的資料
  • 將新型別的資料傳入,通過 actual.onNext(v) 實現原始observer接收變換後新資料的功能

其實在不瞭解RxJava的原始碼之前,我們常常會驚歎於RxJava的神奇:明明我傳送的是數字1,我要接收的卻是字串型別的資料,這中間僅僅是通過了一個map操作符,簡直太神奇了!在我們瞭解了map操作符的原始碼之後,就會知道:我們在訂閱的時候map操作符將原始的observer包裝成了一個 MapObserver 物件,在這個 MapObserver 內部的 onNext 方法中首先會將原始資料 1 通過我們傳入的變換規則變換為 result : 1,在變換之後才會呼叫我們編寫的原始observer來處理新的 String 型別的資料。

使用subscribeOn和observeOn做執行緒切換

在分析執行緒切換之前,我們先明確一些前置知識:從主執行緒切換到子執行緒的操作很簡單,新開一條執行緒,在新的執行緒執行即可。從子執行緒切換到主執行緒,在Android開發中,所有的第三方框架使用的都是Handler機制。所以不要把這些第三方框架想的特別難,他們的本質都是Android中的基礎知識。

我們以 subscribeOn(Schedulers.io()) 這個例子來分析,首先傳入的是 Schedulers.io() ,直接說結論吧,它就是一個 IoScheduler 型別的單例物件。繼續看程式碼:

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

我們都很熟悉了,本質是建立並返回了一個 ObservableSubscribeOn 型別的物件。我們繼續看 ObservableSubscribeOn 的內部:

    public void subscribeActual(final Observer<? super T> s) {
        //建立一個對原始observer的包裝物件
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

        s.onSubscribe(parent);
        //這裡是關鍵
        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    }

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

        private static final long serialVersionUID = 8094547886072529208L;
        final Observer<? super T> actual;

        final AtomicReference<Disposable> s;

        SubscribeOnObserver(Observer<? super T> actual) {
            this.actual = actual;
            this.s = new AtomicReference<Disposable>();
        }
        
        @Override
        public void onNext(T t) {
            actual.onNext(t);
        }

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

        @Override
        public void onComplete() {
            actual.onComplete();
        }
        //忽略無關程式碼......
    }

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

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

        @Override
        public void run() {
            //這一句我們非常熟悉了,在這裡實現了事件的訂閱
            source.subscribe(parent);
        }
    }

這裡跟 ObservableMap 中的 subscribeActual 也是一個套路:

  • 建立一個下一級observer的包裝類 SubscribeOnObserver
  • 通過 source.subscribe(parent) 實現事件的訂閱

通過看 SubscribeOnObserver 這個包裝類中的 onNext 方法也可以明白:它的內部直接呼叫的就是 actual.onNext(t) ,沒有像map一樣做資料的變換,這很好理解,因為 subscribeOn(schedule) 本身就只是為了切換執行緒,並不做其他多餘的操作,所以這個包裝類中的 onNext 才會直接呼叫下一級observer的onNext。

可能有的小夥伴要有問題了:你說通過 source.subscribe(parent) 實現事件的訂閱,但 subscribeActual 中並沒有你說的程式碼啊?其實在這裡 scheduler.scheduleDirect(new SubscribeTask(parent)) 這裡傳入了一個 SubscribeTask 物件,這個物件其實是個 Runnable ,在它的 run() 方法中呼叫了 source.subscribe(parent) 。到目前為止,我們唯一陌生的就是 scheduleDirect 這個方法了,這個方法定義在 Scheduler 這個執行緒排程類中,它實際呼叫的是:

@NonNull
    public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
        //createWorker是一個抽象方法,不同的執行緒排程器有不同的實現
        final Worker w = createWorker();

        final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

        DisposeTask task = new DisposeTask(decoratedRun, w);
        //事件排程處理
        w.schedule(task, delay, unit);

        return task;
    }

scheduleDirect 這個方法做了兩件事:建立一個 Worker 工作類,呼叫工作類的 schedule 來進行事件排程。不同的執行緒排程器會有不同的處理,對於 IoSchedule 這個排程器來說,它最終是執行這個方法:

    public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
        //忽略無關程式碼......
        Future<?> f;
        try {
            if (delayTime <= 0) {
                //通過執行緒池來開啟一個子執行緒執行
                f = executor.submit((Callable<Object>)sr);
            } else {
                f = executor.schedule((Callable<Object>)sr, delayTime, unit);
            }
            sr.setFuture(f);
        } catch (RejectedExecutionException ex) {
            if (parent != null) {
                parent.remove(sr);
            }
            RxJavaPlugins.onError(ex);
        }

        return sr;
    }

通過上面的分析我們可以知道, subscribeOn(Schedulers.io()) 本質上是通過執行緒池開啟了一個執行緒,在這個新的執行緒中還是通過呼叫 source.subscribe(observer) 來訂閱事件。

subscribeOn(Schedulers.io()) 分析過了,我們再來看一下 subscribeOn(AndroidSchedulers.mainThread()) 這個執行緒切換。
AndroidSchedulers.mainThread() 返回的是一個 HandlerScheduler 型別的執行緒排程器,它的 scheduleDirect 方法定義如下:

    public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
        if (run == null) throw new NullPointerException("run == null");
        if (unit == null) throw new NullPointerException("unit == null");

        run = RxJavaPlugins.onSchedule(run);
        ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
        
        //本質是通過handler機制實現執行緒切換
        handler.postDelayed(scheduled, Math.max(0L, unit.toMillis(delay)));
        return scheduled;
    }

從它的方法內部可以看到,它是通過 handler.postDelayed() 來實現切換到主執行緒的功能的,這個handler是定義在主執行緒的handler。其他型別的執行緒排程器就不再分析了,本質都是大同小異。

接下來我們來看一下 observeOn(AndroidSchedulers.mainThread()) ,額,其實它的套路也是一樣的,建立並返回了一個 ObservableObserveOn ,我們主要關注它的 subscribeActual 方法:

    protected void subscribeActual(Observer<? super T> observer) {
            //忽略無關程式碼......
            
            Scheduler.Worker w = scheduler.createWorker();
            //訂閱包裝類ObserveOnObserver
            source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
    }
    
    static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
    implements Observer<T>, Runnable {
        //下一級觀察者
        final Observer<? super T> actual;
        //對應Scheduler裡的Worker
        final Scheduler.Worker worker;
        //上一級Observable傳送的資料佇列
        SimpleQueue<T> queue;
        Disposable s;
        //如果onError了,儲存對應的異常
        Throwable error;
        //是否完成
        volatile boolean done;
        //是否取消
        volatile boolean cancelled;
        // 代表同步傳送 非同步傳送 
        int sourceMode;
        ....
        @Override
        public void onSubscribe(Disposable s) {
            if (DisposableHelper.validate(this.s, s)) {
                this.s = s;
                //忽略無關程式碼......
        
                queue = new SpscLinkedArrayQueue<T>(bufferSize);
                actual.onSubscribe(this);
            }
        }

        @Override
        public void onNext(T t) {
            //執行過error / complete ,直接返回
            if (done) {
                return;
            }
            //如果資料來源型別不是非同步的, 預設不是
            if (sourceMode != QueueDisposable.ASYNC) {
                //將上一級Observable傳送的資料加入佇列中
                queue.offer(t);
            }
            //這句程式碼是執行緒排程的核心,進入相應Worker執行緒,在相應執行緒傳送資料佇列中的資料
            schedule();
        }

        @Override
        public void onError(Throwable t) {
            if (done) {
                RxJavaPlugins.onError(t);
                return;
            }
            error = t;
            done = true;
            //這句程式碼是執行緒排程的核心,進入相應Worker執行緒,在相應執行緒傳送資料佇列中的資料
            schedule();
        }

        @Override
        public void onComplete() {
            if (done) {
                return;
            }
            done = true;
            //開始排程
            schedule();
        }

        void schedule() {
            if (getAndIncrement() == 0) {
                worker.schedule(this);
            }
        }
        
        @Override
        public void run() {
            //預設是false
            if (outputFused) {
                drainFused();
            } else {
                //取出佇列中的資料併傳送
                drainNormal();
            }
        }


        void drainNormal() {
            int missed = 1;

            final SimpleQueue<T> q = queue;
            final Observer<? super T> a = actual;

            for (;;) {
                //如果已經結束或佇列為空,跳出迴圈
                if (checkTerminated(done, q.isEmpty(), a)) {
                    return;
                }

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

                    try {
                        //從佇列裡取出一個值
                        v = q.poll();
                    } catch (Throwable ex) {
                        //異常處理並跳出函式
                        Exceptions.throwIfFatal(ex);
                        s.dispose();
                        q.clear();
                        a.onError(ex);
                        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;
                }
            }
        }

        boolean checkTerminated(boolean d, boolean empty, Observer<? super T> a) {
            //如果已經disposed 
            if (cancelled) {
                queue.clear();
                return true;
            }
            // 如果已經結束
            if (d) {
                Throwable e = error;
                //如果是延遲傳送錯誤
                if (delayError) {
                    //如果空
                    if (empty) {
                        if (e != null) {
                            a.onError(e);
                        } else {
                            a.onComplete();
                        }
                        //停止worker(執行緒)
                        worker.dispose();
                        return true;
                    }
                } else {
                    //傳送錯誤
                    if (e != null) {
                        queue.clear();
                        a.onError(e);
                        worker.dispose();
                        return true;
                    } else
                    //傳送complete
                    if (empty) {
                        a.onComplete();
                        worker.dispose();
                        return true;
                    }
                }
            }
            return false;
        }
    }

可以看到 subscribeOnobserveOn 不同的是二者執行緒切換的地方。 subscribeOn 是將 source.subscribe(observer) 整個部分切換到了相應的執行緒,而 observeOn 則是在 ObserveOnObserver 這個包裝類中的 onNextonError 方法中做了執行緒的切換,具體是在 onNext 等方法中建立了相關的 Worker 工作類,並呼叫了這個類的 schedule 方法,這個跟我們講 subscribeOn 是一樣的。

到此為止,RxJava2的原始碼就基本分析完了。最後根據最初的例子,用一幅圖來展示RxJava的流程:

相關文章

Android 開發 最新文章