前言

今年八月開始踏上RxJava的學(xué)習(xí)之路，當(dāng)時的RxJava版本已經(jīng)更新到3.x了恐似。學(xué)習(xí)的過程中參考了RxJava2的資料，demo的學(xué)習(xí)以官網(wǎng)為主。文章的內(nèi)容不會就RxJava的操作符來進(jìn)行講述擦剑，因為這方面的文章已經(jīng)很多了，參考官網(wǎng)提供的例子就好了芥颈。但要注意RxJava3和RxJava2是有些區(qū)別的惠勒。文章更偏向于源碼的分析。

資料

• RxJava官網(wǎng)
• RxJava文檔中文版（不是RxJava3爬坑，可以參考）
• 給 Android 開發(fā)者的 RxJava 詳解
• RxJava2 只看這一篇文章就夠了
• RxJava2源碼分析

在RxJava的官網(wǎng)的wiki詳細(xì)介紹了RxJava的各種操作符纠屋，理解的時候要結(jié)合官方給出的每個操作符對應(yīng)的模型圖，如果英文不好可以自行有道或者直接參考中文版的文檔盾计。資料三可以加深你對RxJava的認(rèn)識和理解售担，這個一定要看看；資料四可以當(dāng)字典用署辉，主要介紹了各種操作符的使用族铆。

如果以上資料不能滿足你，可以關(guān)注任玉剛哭尝、郭霖哥攘、劉望舒、何俊林材鹦、鴻洋這幾位大佬的博客或者公眾號逝淹，里面都有關(guān)于RxJava的文章，可以看看有沒有你要找的答案桶唐。

觀察者模式

RxJava是基于觀察者模式的實現(xiàn)的栅葡。觀察者模式也叫發(fā)布訂閱模式，其定義：定義對象間一種一對多的依賴關(guān)系莽红，使得每當(dāng)一個對象改變狀態(tài)妥畏，則所有依賴與它的對象都會得到通知并被自動更新邦邦。

• Subject被觀察者
  定義被觀察者必須實現(xiàn)的職責(zé)，它必須能夠動態(tài)增加醉蚁、取消觀察者燃辖。它一般是抽象類或者是實現(xiàn)類，僅僅完成作為被觀察者必須實現(xiàn)的職責(zé)：管理觀察者并通知觀察者
• Observer觀察者
  觀察者接收到消息后网棍，即進(jìn)行update（更新方法）操作黔龟，對接收到的信息進(jìn)行處理
• ConcreteSubject具體的被觀察者
  定義被觀察者自己的業(yè)務(wù)邏輯，同時定義對哪些事件進(jìn)行通知
• ConcreteObserver具體的觀察者
  每個觀察在接收到消息后的處理反應(yīng)是不同的滥玷，各個觀察者有自己的處理邏輯

以上關(guān)于觀察者的知識點都摘自《設(shè)計模式之禪》（鏈接氏身，提取碼: tbks），如果你想學(xué)習(xí)設(shè)計模式或者更深入地了解觀察者模式惑畴，可以下載下來讀一讀蛋欣。

下面是一個題目，感興趣的可以用觀察者模式來自己寫一寫：

練習(xí)題.png

源碼分析

源碼分析主要內(nèi)容是RxJava的工作過程如贷。

工作過程

在沒有操作符的情況下：

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Throwable {
                emitter.onNext("hello");
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                System.out.println(s);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                System.out.println("完成");
            }
        });

以上代碼是一個很簡單鏈?zhǔn)秸{(diào)用的寫法陷虎，當(dāng)然你也可以一步一步來寫。我們的源碼分析是基于這個demo出發(fā)的杠袱，在正式開始分析它的工作過程之前尚猿，需要先了解幾個重要的類：

被觀察者

Observable是一個抽象類，我們所用到的各種各樣的操作符的最上層api就封裝在這里楣富，它源碼太長了凿掂，就不貼出來了。ObservableSource纹蝴，是一個接口庄萎，Observable是它的一個實現(xiàn)類，其中subscribe和subscribeActual（抽象方法）方法需要重點關(guān)注骗灶，Observable是抽象類這一點也是需要注意的惨恭。

public interface ObservableSource<T> {
    void subscribe(@NonNull Observer<? super T> observer);
}

// Observable下的subscribe秉馏、subscribeActual
public final Disposable subscribe() {
    return subscribe(Functions.emptyConsumer(), Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
}

public final Disposable subscribe(Consumer<? super T> onNext) {
    return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION);
}

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {
    return subscribe(onNext, onError, Functions.EMPTY_ACTION);
}

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
            Action onComplete) {
    ObjectHelper.requireNonNull(onNext, "onNext is null");
    ObjectHelper.requireNonNull(onError, "onError is null");
    ObjectHelper.requireNonNull(onComplete, "onComplete is null");

    LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, Functions.emptyConsumer());

    subscribe(ls);

    return ls;
}

// 實現(xiàn)了ObservableSource的subscribe
public final void subscribe(Observer<? super T> observer) {
    ObjectHelper.requireNonNull(observer, "observer is null");
    try {
        observer = RxJavaPlugins.onSubscribe(this, observer);

        ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");

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

protected abstract void subscribeActual(Observer<? super T> observer);

subscribe可以簡單理解為將觀察者和被觀察者關(guān)聯(lián)起來耙旦，通過閱讀源碼可以知道，它們最終都會調(diào)用subscribe(Observer<? super T> observer)萝究。

subscribeActual可以將操作符之間關(guān)聯(lián)起來免都，而且每調(diào)用一個操作符都會返回一個新的Observable，那么它就必須去實現(xiàn)subscribeActual方法帆竹，并將處理好的數(shù)據(jù)發(fā)射出去绕娘。至于它們?nèi)绾谓邮盏缴弦粋€操作符發(fā)射過來的數(shù)據(jù)和如何發(fā)射處理好的數(shù)據(jù)會在后續(xù)的源碼分析中了解到，所以栽连，操作符之間的功能差異可以從這里看出來险领。

發(fā)射器

Emitter侨舆，翻譯過來就是發(fā)射的意思，它可以將生成或創(chuàng)建的數(shù)據(jù)發(fā)射出去（回調(diào)）绢陌。onNext方法是最常見的挨下，他還有很多子類，比如ObservableEmitter脐湾。

ObservableOnSubscribe接口也可以關(guān)注下臭笆，它的subscribe方法參數(shù)類型是Emitter。

public interface Emitter<T> {
    void onNext(@NonNull T value);

    void onError(@NonNull Throwable error);

    void onComplete();
}

public interface ObservableEmitter<T> extends Emitter<T> {
    void setDisposable(@Nullable Disposable d);

    void setCancellable(@Nullable Cancellable c);

    boolean isDisposed();

    @NonNull
    ObservableEmitter<T> serialize();

    boolean tryOnError(@NonNull Throwable t);
}

public interface ObservableOnSubscribe<T> {
    void subscribe(@NonNull ObservableEmitter<T> emitter) throws Throwable;
}

觀察者

Observer秤掌，可以處理最終的數(shù)據(jù)愁铺。

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

    void onNext(@NonNull T t);

    void onError(@NonNull Throwable e);

    void onComplete();
}

對這幾個關(guān)鍵的類有所了解后，對后續(xù)的源碼分析會有很大的幫助闻鉴。那么先從上述demo的create創(chuàng)建操作符開始：

public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    ObjectHelper.requireNonNull(source, "source is null");
    // 可以理解成返回一個Observable對象
    return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}

RxJava有很多地方都對傳遞過來的值做判斷茵乱，比如ObjectHelper.requireNonNull(source, "source is null")，如果為null孟岛，那么傳遞就會被中斷似将，這塊不用過多關(guān)注。我們重點關(guān)注new ObservableCreate<T>(source)蚀苛，至于RxJavaPlugins.onAssembly在验，在當(dāng)下，可以忽略堵未。

以下是ObservableCreate的部分源碼腋舌，為了方便閱讀我刪了一些：

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

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

    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 void onComplete() {
            if (!isDisposed()) {
                try {
                    observer.onComplete();
                } finally {
                    dispose();
                }
            }
        }

    }

}

可知，ObservableCreate是Observable的子類渗蟹，其中subscribeActual抽象方法被它實現(xiàn)了块饺。總的來說雌芽，create操作符只是創(chuàng)建了一個Observable對象授艰。它的構(gòu)造方法的參數(shù)類型是ObservableOnSubscribe，也就是發(fā)射器Emitter世落，在前面也對它進(jìn)行了簡單的介紹了淮腾，而泛型T對發(fā)射的數(shù)據(jù)類型做了限制。

其實屉佳，ObservableCreate最關(guān)鍵的點是對subscribeActual的實現(xiàn)以及被調(diào)用的時機(jī)谷朝，而CreateEmitter則對demo里的emitter.onNext("hello")作了具體的實現(xiàn)，說白了就是怎么把數(shù)據(jù)發(fā)射出去武花。

我們暫時不對這塊作深入分析圆凰，先往下走。接著就執(zhí)行到subscribe方法体箕，在這里貼一下它的關(guān)鍵代碼：

subscribeActual(observer);

剛剛就提到了subscribeActual的調(diào)用時機(jī)了专钉，它就是在Observable的subscribe方法里回調(diào)ObservableCreate的subscribeActual方法的挑童。以下是它的源碼：

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

observer就是demo里的Observer對象；CreateEmitter是一個發(fā)射器跃须，它持有Observer對象的引用炮沐，在其構(gòu)造方法里初始化；observer.onSubscribe會回調(diào)demo里的onSubscribe方法回怜，source.subscribe會回調(diào)demo里的subscribe方法大年。看到這里玉雾，我們已經(jīng)清楚地知道了onSubscribe和subscribe方法的參數(shù)是誰了翔试。

現(xiàn)在，我在回頭看一下demo里的一句代碼：

emitter.onNext("hello")

當(dāng)這一句被調(diào)用的時候复旬，在CreateEmitter內(nèi)部類里的onNext方法就會被調(diào)用垦缅，也就是說demo里的onNext方法會被回調(diào)，發(fā)射出來的數(shù)據(jù)也通過參數(shù)的傳遞到達(dá)了觀察者Observer驹碍。

// CreateEmitter類的onNext
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在前面已經(jīng)提過了
        observer.onNext(t);
    }
}

現(xiàn)在我們寫一個更加復(fù)雜一點的：

Observable.just(1, 2, 3)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Throwable {
                        return integer % 2 != 0;
                    }
                })
                .mergeWith(Observable.just(4, 5, 6))
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        System.out.println(integer);
                    }
                });
// 打印的數(shù)據(jù)
// 1壁涎，3，4志秃，5怔球，6

在前面的基礎(chǔ)上，我們可以很好理解浮还。大體上是沒啥區(qū)別的竟坛，在這里，我們將關(guān)注點的中心放在操作符的subscribeActual方法上钧舌。以下是filter和mergeWith操作符的subscribeActual方法的源碼担汤，其它源碼就不貼，到最后還是回歸到第一個demo的分析思路洼冻。

// ObservableFlatMap(mergeWith操作符)
public void subscribeActual(Observer<? super U> t) {

    if (ObservableScalarXMap.tryScalarXMapSubscribe(source, t, mapper)) {
        return;
    }

    source.subscribe(new MergeObserver<T, U>(t, mapper, delayErrors, maxConcurrency, bufferSize));
}

// ObservableFilter(filter操作符)
public void subscribeActual(Observer<? super T> observer) {
    source.subscribe(new FilterObserver<T>(observer, predicate));
}

這里的source是Observable對象（可以debug查看）崭歧，每一個操作符都會返回一個新的Observable對象。

大概的執(zhí)行流程：根據(jù)操作符just撞牢、filter率碾、mergeWith的執(zhí)行順序先后創(chuàng)建了各自的Observable對象，然后調(diào)起subscribe方法普泡，而操作符之間通過subscribe關(guān)聯(lián)起來播掷，依次執(zhí)行了megeWith审编、filter撼班、just操作符的subscribeActual方法，最后回歸到前面分析的流程垒酬。

總結(jié)

整體上砰嘁，RxJava的源碼還是很繞的件炉，還需要多一點耐心。在后續(xù)也會把它的線程調(diào)度過程不上矮湘，目前還需要補(bǔ)一補(bǔ)線程的知識斟冕。