Android

前言

RxJava 是什么

RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.

以上是RxJava在Github上的介紹坡慌，大概意思是状原，針對于JVM(Java虛擬機(jī)）的響應(yīng)式擴(kuò)展實(shí)現(xiàn)斜脂，一個在Java VM上使用可觀察的序列來組合實(shí)現(xiàn)異步的抓艳、基于事件編程的庫。

RxJava現(xiàn)在大家用的都應(yīng)該已經(jīng)很溜了帚戳，用法這里就不再多說了玷或。我們都知道RxJava是對觀察者模式的擴(kuò)展，下面就從觀察者模式的實(shí)現(xiàn)機(jī)制出發(fā)片任，了解一下RxJava2的實(shí)現(xiàn)邏輯偏友。只有真正了解了RxJava 的實(shí)現(xiàn)原理，我們才能在遇到問題的時(shí)候对供，更快速更準(zhǔn)確的定位的到問題位他。

此次源碼分析基于 RxJava Release 2.1.7

觀察者模式

這里簡單回顧一下觀察者模式的組成及使用方式，通過之前觀察者模式一文中的分析犁钟，我們知道觀察者模式中有四個重要的角色：

• 抽象主題：定義添加和刪除觀察者的功能棱诱，也就是注冊和解除注冊的功能
• 抽象觀察者：定義觀察者收到主題通知后要做什么事情
• 具體主題：抽象主題的實(shí)現(xiàn)
• 具體觀察者：抽象觀察者的實(shí)現(xiàn)

當(dāng)我們創(chuàng)建好了具體主題和觀察者類，就可以使用觀察者模式了涝动，下面是一個最簡單的測試demo迈勋。

public class TestObservePattern {

    public static void main(String[] args) {
        // 創(chuàng)建主題(被觀察者)
        ConcreteSubject concreteSubject = new ConcreteSubject();
        // 創(chuàng)建觀察者
        ObserverOne observerOne=new ObserverOne();
        // 為主題添加觀察者
        concreteSubject.addObserver(observerOne);        
        //主題通知所有的觀察者
        concreteSubject.notifyAllObserver("wake up,wake up");
    }

}

以上就是觀察者模式的使用方式，很簡單是吧〈姿冢現(xiàn)在就讓我們帶著以下幾個問題靡菇，看看RxJava是如何使用觀察者模式的重归。

用RxJava這么久了，你可以思考一下如下幾個問題：

1. RxJava 中上面提到的四個重要角色是如何定義的厦凤？
2. RxJava 中具體的主題鼻吮，具體的觀察者是如何實(shí)例化的？
3. RxJava 中觀察者和主題是如何實(shí)現(xiàn)訂閱的较鼓？
4. RxJava 中上游是怎么發(fā)送事件的椎木，下游又是怎樣接收到的?
5. RxJava 中對常規(guī)的觀察者模式做了怎樣調(diào)整，帶來了什么好處博烂？

如果對以上幾個問題香椎，你有明確的答案，恭喜你禽篱，以下內(nèi)容你就不用再看了畜伐，O(∩_∩)O哈哈~。

很多開發(fā)者對RxJava的學(xué)習(xí)可能是從上游和下游的角度開始躺率，這里可以認(rèn)為這樣的敘述更偏重RxJava 事件序列的特征玛界。本文從被觀察者（主題）和觀察者的角度出發(fā)，可以說是更偏向于RxJava 觀察者模式的特征悼吱。這里的主題就是上游慎框，觀察者就是下游。無論從哪個角度出發(fā)去理解舆绎，源碼就那么一份鲤脏，無所謂對錯，只是每個人的認(rèn)知角度不同而已吕朵，選擇一種自己更容易了解的方式即可。

好了窥突，如果你看到了這里努溃，說明你對以上幾個問題，還有些許疑問阻问，那么我們就從這幾個問題出發(fā)梧税，了解一下RxJava的源碼實(shí)現(xiàn)。

RxJava2 的觀察者模式實(shí)現(xiàn)

我們就帶著上述幾個問題称近，依次來看看RxJava到底是怎么一回事兒第队。為了方便敘述和記憶，我們首先看一段RxJava2 最最基礎(chǔ)的使用方式刨秆。

private void basicRxjava2() {
        Observable mObservable = Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(ObservableEmitter e) throws Exception {
                e.onNext("1");
                e.onNext("2");
                e.onNext("3");
                e.onNext("4");
                e.onComplete();
            }
        });

        Observer mObserver = new Observer() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe: d=" + d);
                sb.append("\nonSubcribe: d=" + d);
            }

            @Override
            public void onNext(Object s) {
                Log.e(TAG, "onNext: " + s);
                sb.append("\nonNext: " + s);
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: " + e);
                sb.append("\nonError: " + e.toString());
                logContent.setText(sb.toString());
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete");
                sb.append("\nonComplete: ");
                logContent.setText(sb.toString());
            }
        };

        mObservable.subscribe(mObserver);
    }

上面這段代碼凳谦，應(yīng)該很容易理解了，輸出結(jié)果大家閉著眼睛也能想出來吧衡未。我們就以這段代碼為基礎(chǔ)尸执，結(jié)合上面提到的問題依次展開對RxJava的分析家凯。

四個重要的角色

• 抽象主題

首先可以看看這個Observable類。

public abstract class Observable<T> implements ObservableSource<T> {
……
}

他實(shí)現(xiàn)了ObservableSource接口如失，接著看ObservableSource

public interface ObservableSource<T> {

    /**
     * Subscribes the given Observer to this ObservableSource instance.
     * @param observer the Observer, not null
     * @throws NullPointerException if {@code observer} is null
     */
    void subscribe(@NonNull Observer<? super T> observer);
}

這里很明顯了绊诲，ObservableSource 就是抽象主題（被觀察者）的角色。按照之前觀察者模式中約定的職責(zé)褪贵，subscribe 方法就是用來實(shí)現(xiàn)訂閱觀察者（Observer）角色的功能掂之。從這里我們也可以看出，抽象觀察者的角色就是Observer了脆丁。

這里板惑，你也許會有疑問愕难，這么簡單掰茶？抽象主題（上游）不是需要發(fā)送事件嗎？onNext(),onComplete()以及onError()跑哪兒去了蛤奥？別著急晒夹，我們后面慢慢看裆馒。

• 具體主題

回過頭來繼續(xù)看Observable，他實(shí)現(xiàn)了ObservableSource接口丐怯，并且實(shí)現(xiàn)了其subscribe方法喷好，但是它并沒有真正的去完成主題和觀察者之間的訂閱關(guān)系，而是把這個功能读跷，轉(zhuǎn)接給了另一個抽象方法subscribeActual（具體細(xì)節(jié)后面分析）梗搅。

因此，Observable依舊是一個抽象類效览，我們知道抽象類是不能被實(shí)例化的无切，因此從理論上來說，他好像不能作為具體主題的角色丐枉。其實(shí)不然哆键，Observable內(nèi)部提供了create,defer,fromXXX,repeat,just等一系列創(chuàng)建型操作符， 用來創(chuàng)建各種Observable瘦锹。

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

在RxJava內(nèi)有很多他的子類籍嘹。

Observable的子類

誠然，你可以認(rèn)為弯院，這些子類其實(shí)才是真正的具體主題辱士。但是，換一個角度听绳，從代理模式的角度出發(fā)颂碘，我們可以把Observable當(dāng)做是一個代理類，客戶端你只管調(diào)用create 方法辫红，想要什么樣的
Observable告訴我一聲就可以凭涂，不同Observeable之間的差異你不用管祝辣，包在我身上，保證給你返回你想要的Observeable實(shí)例切油。

同時(shí)蝙斜，Observable另一個巨大的貢獻(xiàn)，就是定義了很多的操作符澎胡，我們平時(shí)常用的map,flatMap,distinct等孕荠，也是在這里定義。并且這些方法都是final類型的攻谁，因此他的所有子類都會繼承同時(shí)也無法改變這些操作符的實(shí)現(xiàn)稚伍。

因此，Observable 就是具體主題戚宦。

• 抽象觀察者

在抽象主題里已經(jīng)提過了个曙，Observer就是抽象觀察者的角色。

public interface Observer<T> {

    void onSubscribe(@NonNull Disposable d);

    void onNext(@NonNull T t);

    void onError(@NonNull Throwable e);

    void onComplete();

}

非常符合觀察者模式中抽象觀察者的職責(zé)描述受楼，Observer 定義了觀察者（下游）收到主題（上游）通知后該做什么事情垦搬。這里需要注意的是onSubscribe 也是定義在這里的。

• 具體的觀察者

這個具體的觀察者艳汽，o(╯□╰)oo(╯□╰)o猴贰，就不多說了吧。大家平時(shí)使用應(yīng)該都是直接用new一個Observer的實(shí)例河狐。RxJava內(nèi)部有很多Observer的子類米绕，有興趣的同學(xué)可以具體了解一下。這里其實(shí)可以引申出一個有意思的問題馋艺，同樣是抽象類栅干，為什么接口可以直接實(shí)例化，而用abstract修飾過的類就不可以丈钙？

具體的觀察者是如何實(shí)例化的

我們看一下這段代碼：

    Observable mObservable = Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(ObservableEmitter e) throws Exception {
            }
        });

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        ObjectHelper.requireNonNull(source, "source is null");
        return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
    }
    
    
    public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
        Function<? super Observable, ? extends Observable> f = onObservableAssembly;
        // 是否有別的其他操作符運(yùn)算非驮，有的話，在此Observable上執(zhí)行一遍
        if (f != null) {
            return apply(f, source);
        }
        return source;
    }

RxJava的代碼里雏赦，很多時(shí)候會有ObjectHelper.requireNonNull這種空檢查的地方，一律都是為了最大程度的防止NPE的出現(xiàn)芙扎，后面出現(xiàn)就不再贅述了.

我們使用create操作符創(chuàng)建Observable的過程中星岗，看似經(jīng)歷了很多方法，在不考慮任何其他操作符的前提下戒洼，整個過程簡化一下的話就這么一句代碼

  Observable mObservable=new ObservableCreate(new ObservableOnSubscribe())

從之前的分析俏橘，我們也看到了ObservableCreate 就是Observeable抽象類的一個子類。我們簡單看一下他的實(shí)現(xiàn)圈浇。

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) {
        ……
    }
}

可以看到寥掐，他唯一的構(gòu)造函數(shù)需要一個ObservableOnSubscribe實(shí)例靴寂，同時(shí)他實(shí)現(xiàn)subscribeActual方法，說明他真正處理主題和觀察者之間實(shí)現(xiàn)訂閱的邏輯召耘。

看了半天百炬，你可能一直很好奇，這個ObservableOnSubscribe是個什么東西呢污它？他其實(shí)很簡單剖踊。

/**
 * A functional interface that has a {@code subscribe()} method that receives
 * an instance of an {@link ObservableEmitter} instance that allows pushing
 * events in a cancellation-safe manner.
 *
 * @param <T> the value type pushed
 */
public interface ObservableOnSubscribe<T> {

    /**
     * Called for each Observer that subscribes.
     * @param e the safe emitter instance, never null
     * @throws Exception on error
     */
    void subscribe(@NonNull ObservableEmitter<T> e) throws Exception;
}

ε=(′ο｀*)))唉，怎么又一個subscribe衫贬，這又是啥德澈？不要慌，看注釋固惯。意思是說梆造，這里的subscribe 接收到一個ObservableEmitter實(shí)例后，就會允許他以一種可以安全取消（也就是一定能取消）的形式發(fā)送事件葬毫。

就是說會有某個對象镇辉，給他一個ObservableEmitte的實(shí)例，沒給他之前他是不會主動發(fā)送事件的供常，會一直憋著摊聋。，到這里栈暇，你是不是想到了什么麻裁，我們知道在RxJava 中只有觀察者（下游）訂閱(subscribe)了主題（上游），主題才會發(fā)送事件源祈。這就是和普通的觀察者模式有區(qū)別的地方之一煎源。

好了，最后再來看看這個神秘的ObservableEmitter是個什么鬼香缺？

public interface ObservableEmitter<T> extends Emitter<T> {

    void setDisposable(@Nullable Disposable d);


    void setCancellable(@Nullable Cancellable c);


    boolean isDisposed();

    ObservableEmitter<T> serialize();

      /**
     * Attempts to emit the specified {@code Throwable} error if the downstream
     * hasn't cancelled the sequence or is otherwise terminated, returning false
     * if the emission is not allowed to happen due to lifecycle restrictions.
     * <p>
     * Unlike {@link #onError(Throwable)}, the {@code RxJavaPlugins.onError} is not called
     * if the error could not be delivered.
     * @param t the throwable error to signal if possible
     * @return true if successful, false if the downstream is not able to accept further
     * events
     * @since 2.1.1 - experimental
     */
    boolean tryOnError(@NonNull Throwable t);
}

這里可以關(guān)注一下tryOnError這個方法手销，可以看到他會把某些類型的error傳遞到下游。

o(╥﹏╥)o图张，又是一個接口锋拖，而且還繼承了另一個接口，什么情況祸轮？繼續(xù)看

public interface Emitter<T> {

    void onNext(@NonNull T value);

  
    void onError(@NonNull Throwable error);

   
    void onComplete();
}

驚不驚喜兽埃，意不意外？ 哈哈适袜，終于找到你了柄错，熟悉的onNext，onError,onComplete.原來在這里。

這里有個問題可以思考一下售貌，在抽象觀察者中给猾，定義了四個處理事件的方法，這里只有三個颂跨，按照對應(yīng)關(guān)系來說似乎缺了一個onSubscribe敢伸，這又是怎么回事呢？后面會有分析毫捣，可以自己先想想

這兩個接口的含義很明顯了详拙，總結(jié)一下：

• Emitter 定義了可以發(fā)送的事件的三種機(jī)制
• ObservableEmitter 在Emitter 做了擴(kuò)展，添加了Disposable相關(guān)的方法蔓同，可以用來取消事件的發(fā)送饶辙。

好了，繞了一大圈斑粱，就為了一行代碼：

  Observable mObservable=new ObservableCreate(new ObservableOnSubscribe())

總結(jié)一下具體主題（上游）的到底干了啥：

• 創(chuàng)建了一個ObservableCreate 的實(shí)例對象
• ObservableCreate 內(nèi)持有ObservableOnSubscribe 對象的引用
• ObservableOnSubscribe 是一個接口弃揽，內(nèi)部有一個subscribe方法，調(diào)用他之后则北，會用其ObservableEmitter實(shí)例開始發(fā)送事件矿微。
• ObservableEmitter 繼承自Emitte。

如何實(shí)現(xiàn)訂閱尚揣、發(fā)送事件和接收事件

為了方便敘述涌矢，把問題3和4連在一起說了。

通過上面的敘述快骗，現(xiàn)在具體主題和具體的觀察者都創(chuàng)建好了娜庇，接下來就是實(shí)現(xiàn)二者的訂閱關(guān)系。

mObservable.subscribe(mObserver);

這里需要明確的一點(diǎn)是方篮，是觀察者（下游）訂閱了主題（上游）名秀，雖然從代碼上看好像了前者訂閱了后者，不要搞混了藕溅。

我們看Observable的subscribe() 方法：

    public final void subscribe(Observer<? super T> observer) {
        ObjectHelper.requireNonNull(observer, "observer is null");
        try {
            observer = RxJavaPlugins.onSubscribe(this, observer);

            ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");

            subscribeActual(observer);
        } catch (NullPointerException e) { // NOPMD
            throw e;
        } catch (Throwable e) {
         ……
        }
    }

這個前面已經(jīng)提到過了匕得，Observable并沒有真正的去實(shí)現(xiàn)subscribe,而是把他轉(zhuǎn)接給了subscribeActual()方法。

前面已經(jīng)說過巾表，Observable的實(shí)例是一個ObservableCreate對象汁掠，那么我們就到這個類里去看看subscribeActual()的實(shí)現(xiàn)。

    // 為了方便集币，順便再看一眼構(gòu)造函數(shù)
    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);
        }
    }

CreateEmitter 實(shí)現(xiàn)了之前提到的ObservableEmitter接口调塌。這里有一句關(guān)鍵的代碼：

observer.onSubscribe(parent);

之前在看到Emitter的定義時(shí)，我們說缺少了onSubscribe方法惠猿，到這里就明白了。onSubscribe并不是由主題（上游）主動發(fā)送的事件，而是有觀察者（下游）自己調(diào)用的一個事件偶妖，只是為了方便獲取Emitter的實(shí)例對象姜凄，準(zhǔn)確的說應(yīng)該是Disposable的實(shí)例對象，這樣下游就可以控制上游了趾访。

接下來就更簡單了态秧，source 是ObservableOnSubscribe，按照之前的邏輯扼鞋，調(diào)用其subscribe方法申鱼，給他一個ObservableEmitter對象實(shí)例，ObservableEmitter就會開始發(fā)送事件序列云头。這樣捐友，一旦開始訂閱了，主題（上游）就開始發(fā)送事件了溃槐。

接著看看CreateEmitter的實(shí)現(xiàn)匣砖。

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);
        observer.onSubscribe(parent);
        ……
    }

    static final class CreateEmitter<T>
    extends AtomicReference<Disposable>
    implements ObservableEmitter<T>, Disposable {


        private static final long serialVersionUID = -3434801548987643227L;

        final Observer<? super T> 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(t);
            }
        }

        @Override
        public void onError(Throwable t) {
            if (!tryOnError(t)) {
                RxJavaPlugins.onError(t);
            }
        }

        @Override
        public boolean tryOnError(Throwable t) {
            if (t == null) {
                t = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
            }
            if (!isDisposed()) {
                try {
                    observer.onError(t);
                } finally {
                    dispose();
                }
                return true;
            }
            return false;
        }

        @Override
        public void onComplete() {
            if (!isDisposed()) {
                try {
                    observer.onComplete();
                } finally {
                    dispose();
                }
            }
        }

        @Override
        public void setDisposable(Disposable d) {
            DisposableHelper.set(this, d);
        }

        @Override
        public void setCancellable(Cancellable c) {
            setDisposable(new CancellableDisposable(c));
        }

        @Override
        public ObservableEmitter<T> serialize() {
            return new SerializedEmitter<T>(this);
        }

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

        @Override
        public boolean isDisposed() {
            return DisposableHelper.isDisposed(get());
        }
    }
}

• 他的構(gòu)造函數(shù)，需要一個觀察者的實(shí)例昏滴；
• 他實(shí)現(xiàn)了ObservableEmitter接口猴鲫，并依次實(shí)現(xiàn)他的三個方法；
  • 在每一次的onNext事件中谣殊，他不再接受參數(shù)為null的類型拂共，在事件序列沒有中斷的情況下會把主題（上游）發(fā)送的事件T原封不動的傳遞給觀察者(下游）。
  • onComplete事件發(fā)生時(shí)姻几，他也會通知下游宜狐，如果發(fā)生異常，則中斷事件序列
  • onError 事件發(fā)生時(shí)鲜棠，并沒有直接傳遞到下游肌厨，而是在其內(nèi)部處理
  • tryOnError 事件發(fā)生時(shí)，才會把某些特定類型的錯誤傳遞到下游豁陆。
• 他實(shí)現(xiàn)了Disposable接口柑爸，下游根據(jù)獲取到的Emitter的實(shí)例對象，可以方便的獲取事件序列的信息盒音，甚至是可以主動關(guān)閉事件序列表鳍，及斷開觀察者模式中主題和觀察者間的訂閱關(guān)系。

RxJava 中對常規(guī)的觀察者模式做了怎樣調(diào)整祥诽，帶來了什么好處譬圣？

最后再來簡單說一下，RxJava中對常規(guī)的觀察者模式做了怎樣的調(diào)整雄坪，有什么值得借鑒的地方厘熟。大部分優(yōu)點(diǎn)在上面已經(jīng)提及了，這里就來總結(jié)一下。

• 觀察者訂閱主題后绳姨，主題才會開始發(fā)送事件
• RxJava中Observer通過onSubscribe獲取了發(fā)送事件中的Disposable對象登澜，這樣他就可以主動的獲取訂閱關(guān)系中二者的狀態(tài)，甚至是控制或者是中斷事件序列的發(fā)送飘庄。在常規(guī)的觀察者模式中脑蠕，主題有權(quán)利添加訂閱者，但也能是由他移除特定的訂閱者跪削，因?yàn)橹挥兴钟兴杏嗛喺叩募?/li>
• 抽象主題（上游）并沒有直接控制onNext谴仙，onComplete,onError事件的發(fā)送，而是只關(guān)注Emitter 實(shí)例的發(fā)送碾盐，ObservableOnSubscribe接口監(jiān)聽ObservableEmitter對象的發(fā)送晃跺，一旦接受到此對象就會通過他開始發(fā)送具體的事件，這里可以有點(diǎn)觀察者模式嵌套的意味廓旬。

好了哼审，以上就是從觀察者模式的角度出發(fā)，對RxJava的一次解讀孕豹，有什么疏漏或理解錯誤的地方涩盾，歡迎讀者指出，共同進(jìn)步励背！