剛剛開通了微信公眾號：BaronTalk，之前專欄上的文章也陸續(xù)完成了搬遷。后續(xù)會持續(xù)保質(zhì)保量的輸出岳悟，覺得我的文章還有值得一讀那就關(guān)注一波吧！Ｗ旄摺竿音！ :-)

• RxJava系列1(簡介)
• RxJava系列2(基本概念及使用介紹)
• RxJava系列3(轉(zhuǎn)換操作符)
• RxJava系列4(過濾操作符)
• RxJava系列5(組合操作符)
• RxJava系列6(從微觀角度解讀RxJava源碼)
• RxJava系列7(最佳實踐)

前言

上一篇的示例代碼中大家一定發(fā)現(xiàn)了Observable這個類。從純Java的觀點看拴驮，Observable類源自于經(jīng)典的觀察者模式春瞬。RxJava的異步實現(xiàn)正是基于觀察者模式來實現(xiàn)的，而且是一種擴展的觀察者模式套啤。

觀察者模式

觀察者模式基于Subject這個概念宽气，Subject是一種特殊對象，又叫做主題或者被觀察者潜沦。當它改變時那些由它保存的一系列對象將會得到通知萄涯，而這一系列對象被稱作Observer(觀察者)。它們會對外暴漏了一個通知方法(比方說update之類的)唆鸡，當Subject狀態(tài)發(fā)生變化時會調(diào)用的這個方法涝影。

觀察者模式很適合下面這些場景中的任何一個：

1. 當你的架構(gòu)有兩個實體類，一個依賴另一個，你想讓它們互不影響或者是獨立復用它們時趁怔。
2. 當一個變化的對象通知那些與它自身變化相關(guān)聯(lián)的未知數(shù)量的對象時。
3. 當一個變化的對象通知那些無需推斷具體類型的對象時岩梳。

通常一個觀察者模式的類圖是這樣的：

Observer

如果你對觀察者模式不是很了解伯襟，那么強烈建議你先去學習下猿涨。關(guān)于觀察者模式的詳細介紹可以參考我之前的文章：設(shè)計模式之觀察者模式

擴展的觀察者模式

在RxJava中主要有4個角色：

• Observable
• Subject
• Observer
• Subscriber

Observable和Subject是兩個“生產(chǎn)”實體，Observer和Subscriber是兩個“消費”實體姆怪。說直白點Observable對應于觀察者模式中的被觀察者叛赚，而Observer和Subscriber對應于觀察者模式中的觀察者。Subscriber其實是一個實現(xiàn)了Observer的抽象類稽揭，后面我們分析源碼的時候也會介紹到俺附。Subject比較復雜，以后再分析淀衣。

上一篇文章中我們說到RxJava中有個關(guān)鍵概念：事件昙读。觀察者Observer和被觀察者Observable通過subscribe()方法實現(xiàn)訂閱關(guān)系。從而Observable 可以在需要的時候發(fā)出事件來通知Observer膨桥。

RxJava如何使用

我自己在學習一種新技術(shù)的時候通常喜歡先去了解它是怎么用的蛮浑，掌握了使用方法后再去深挖其原理。那么我們現(xiàn)在就來說說RxJava到底該怎么用只嚣。

第一步：創(chuàng)建觀察者Observer

Observer<Object> observer = new Observer<Object>() {

    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(Object s) {

    }
 };

這么簡單沮稚，一個觀察者Observer創(chuàng)建了!

大兄弟你等等...，你之前那篇觀察者模式中不是說觀察者只提供一個update方法的嗎册舞？這特么怎么有三個蕴掏？！调鲸！

少年勿急盛杰，且聽我慢慢道來。在普通的觀察者模式中觀察者一般只會提供一個update()方法用于被觀察者的狀態(tài)發(fā)生變化時藐石，用于提供給被觀察者調(diào)用即供。而在RxJava中的觀察者Observer提供了:onNext()、 onCompleted()和onError()三個方法于微。還記得嗎逗嫡？開篇我們講過RxJava是基于一種擴展的觀察這模式實現(xiàn)，這里多出的onCompleted和onError正是對觀察者模式的擴展株依。ps:onNext就相當于普通觀察者模式中的update

RxJava中添加了普通觀察者模式缺失的三個功能：

1. RxJava中規(guī)定當不再有新的事件發(fā)出時驱证，可以調(diào)用onCompleted()方法作為標示；
2. 當事件處理出現(xiàn)異常時框架自動觸發(fā)onError()方法恋腕；
3. 同時Observables支持鏈式調(diào)用抹锄，從而避免了回調(diào)嵌套的問題。

第二步：創(chuàng)建被觀察者Observable

Observable.create()方法可以創(chuàng)建一個Observable，使用crate()創(chuàng)建Observable需要一個OnSubscribe對象伙单，這個對象繼承Action1呆万。當觀察者訂閱我們的Observable時，它作為一個參數(shù)傳入并執(zhí)行call()函數(shù)车份。

Observable<Object> observable = Observable.create(new               Observable.OnSubscribe<Object>() {
    @Override
    public void call(Subscriber<? super Object> subscriber) {

    }
});

除了create()，just()和from()同樣可以創(chuàng)建Observable牡彻∩ㄕ樱看看下面兩個例子：

just(T...)將傳入的參數(shù)依次發(fā)送

Observable observable = Observable.just("One", "Two", "Three");
//上面這行代碼會依次調(diào)用
//onNext("One");
//onNext("Two");
//onNext("Three");
//onCompleted();

from(T[])/from(Iterable<? extends T>)將傳入的數(shù)組或者Iterable拆分成Java對象依次發(fā)送

String[] parameters = {"One", "Two", "Three"};
Observable observable = Observable.from(parameters);
//上面這行代碼會依次調(diào)用
//onNext("One");
//onNext("Two");
//onNext("Three");
//onCompleted();

第三步：被觀察者Observable訂閱觀察者Observer（ps:你沒看錯，不同于普通的觀察者模式庄吼，這里是被觀察者訂閱觀察者）

有了觀察者和被觀察者缎除，我們就可以通過subscribe()來實現(xiàn)二者的訂閱關(guān)系了。

observable.subscribe(observer);

observable.subscribe(observer)

連在一起寫就是這樣：

Observable.create(new Observable.OnSubscribe<Integer>() {

    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        for (int i = 0; i < 5; i++) {
            subscriber.onNext(i);
        }
        subscriber.onCompleted();
    }

}).subscribe(new Observer<Integer>() {

    @Override
    public void onCompleted() {
        System.out.println("onCompleted");
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("onError");
    }

    @Override
    public void onNext(Integer item) {
        System.out.println("Item is " + item);
    }
});

至此一個完整的RxJava調(diào)用就完成了总寻。

兄臺器罐，你叨逼叨叨逼叨的說了一大堆，可是我沒搞定你特么到底在干啥敖バ小轰坊？！祟印！不急肴沫，我現(xiàn)在就來告訴你們到底發(fā)生了什么。

首先我們使用Observable.create()創(chuàng)建了一個新的Observable<Integer>蕴忆，并為create()方法傳入了一個OnSubscribe颤芬，OnSubscribe中包含一個call()方法，一旦我們調(diào)用subscribe()訂閱后就會自動觸發(fā)call()方法套鹅。call()方法中的參數(shù)Subscriber其實就是subscribe()方法中的觀察者Observer站蝠。我們在call()方法中調(diào)用了5次onNext()和1次onCompleted()方法。一套流程周下來以后輸出結(jié)果就是下面這樣的：

Item is 0
Item is 1
Item is 2
Item is 3
Item is 4
onCompleted

看到這里可能你又要說了卓鹿，大兄弟你別唬我傲饽А！OnSubscribe的call()方法中的參數(shù)Subscriber怎么就變成了subscribe()方法中的觀察者Observer减牺？Ｍ阆啊！拔疚！這倆兒貨明明看起來就是兩個不同的類啊肥隆。

我們先看看Subscriber這個類：

public abstract class Subscriber<T> implements Observer<T>, Subscription {
    
    ...
}

從源碼中我們可以看到，Subscriber是Observer的一個抽象實現(xiàn)類稚失，所以我首先可以肯定的是Subscriber和Observer類型是一致的栋艳。接著往下我們看看subscribe()這個方法：

public final Subscription subscribe(final Observer<? super T> observer) {

    //這里的if判斷對于我們要分享的問題沒有關(guān)聯(lián)，可以先無視
    if (observer instanceof Subscriber) {
        return subscribe((Subscriber<? super T>)observer);
    }
    return subscribe(new Subscriber<T>() {

        @Override
        public void onCompleted() {
            observer.onCompleted();
        }

        @Override
        public void onError(Throwable e) {
            observer.onError(e);
        }

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

    });
}

我們看到subscribe()方法內(nèi)部首先將傳進來的Observer做了一層代理句各，將它轉(zhuǎn)換成了Subscriber吸占。我們再看看這個方法內(nèi)部的subscribe()方法：

public final Subscription subscribe(Subscriber<? super T> subscriber) {
    return Observable.subscribe(subscriber, this);
}

進一步往下追蹤看看return后面這段代碼到底做了什么晴叨。精簡掉其他無關(guān)代碼后的subscribe(subscriber, this)方法是這樣的：

private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {

    subscriber.onStart();
    try {
        hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);
        return hook.onSubscribeReturn(subscriber);
    } catch (Throwable e) {
        return Subscriptions.unsubscribed();
    }
}

我們重點看看hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber),前面這個hook.onSubscribeStart(observable, observable.onSubscribe)返回的是它自己括號內(nèi)的第二個參數(shù)observable.onSubscribe,然后調(diào)用了它的call方法。而這個observable.onSubscribe正是create()方法中的Subscriber矾屯，這樣整個流程就理順了兼蕊。看到這里是不是對RxJava的執(zhí)行流程清晰了一點呢件蚕？這里也建議大家在學習新技術(shù)的時候多去翻一翻源碼孙技，知其然還要能知其所以然不是嗎。

subscribe()的參數(shù)除了可以是Observer和Subscriber以外還可以是Action1排作、Action0牵啦；這是一種更簡單的回調(diào)，只有一個call(T)方法妄痪；由于太簡單這里就不做詳細介紹了哈雏！

異步

上一篇文章中開篇就講到RxJava就是來處理異步任務的。但是默認情況下我們在哪個線程調(diào)用subscribe()就在哪個線程生產(chǎn)事件衫生，在哪個線程生產(chǎn)事件就在哪個線程消費事件裳瘪。那怎么做到異步呢？RxJava為我們提供Scheduler用來做線程調(diào)度障簿，我們來看看RxJava提供了哪些Scheduler盹愚。

同時RxJava還為我們提供了subscribeOn()和observeOn()兩個方法來指定Observable和Observer運行的線程。

Observable.from(getCommunitiesFromServer())
            .flatMap(community -> Observable.from(community.houses))
            .filter(house -> house.price>=5000000).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this::addHouseInformationToScreen);

上面這段代碼大家應該有印象吧站故，沒錯正是我們上一篇文章中的例子皆怕。subscribeOn(Schedulers.io())指定了獲取小區(qū)列表、處理房源信息等一系列事件都是在IO線程中運行西篓，observeOn(AndroidSchedulers.mainThread())指定了在屏幕上展示房源的操作在UI線程執(zhí)行愈腾。這就做到了在子線程獲取房源，主線程展示房源岂津。

好了虱黄，RxJava系列的入門內(nèi)容我們就聊到這。下一篇我們再繼續(xù)介紹更多的API以及它們內(nèi)部的原理吮成。

如果你喜歡我的文章橱乱，就關(guān)注下我的公眾號 BaronTalk 、 知乎專欄 或者在 GitHub 上添個 Star 吧粱甫！

• 微信公眾號：BaronTalk
• 知乎專欄：https://zhuanlan.zhihu.com/baron
• GitHub：https://github.com/BaronZ88
• 個人博客：http://baronzhang.com