RxAndroid 學習

 鏈接：https://medium.com/@kurtisnusbaum/rxandroid-basics-part-1-c0d5edcf6850

如果你在閱讀這篇文章，相信你一定很想了解RxJava以及如何在Android應用中使用它。可能你已經見過RxJava的代碼了，但仍然有些疑惑，願你能在這篇文章里找到答案。

當我第一次使用RxJava的時候我只是在照搬代碼，這些代碼能跑起來，但是我對RxJava的基礎部分仍然存在誤解，而且我找不到好的源碼來學習。所以為了理解RxJava，我不得不一點一點學習，踩了不少坑。

為了不讓你把我踩過的坑再踩一遍，我會基於我的學習成果寫一些例子出來，目的就是讓你能夠對RxJava有足夠的了解，並能在你的Android應用中使用它。

源碼可以在這里找到。在每個例子的開始，我會寫清每個代碼段是屬於哪個Activity的。我會將本文分為兩個部分，在前三個例子里，我會着重講解如何用RxJava異步加載數據；在后三個例子里，我會探索一些更高級的用法。

幾個概念

在開始說代碼之前，先澄清幾個概念。RxJava最核心的東西就是Observable和Observer。Observable會發出數據，而與之相對的Observer則會通過訂閱Observable來進行觀察。

Observer可以在Observable發出數據、報錯或者聲明沒有數據可以發送時進行相應的操作。這三個操作被封裝在Observer接口中，相應的方法為onNext()，onError()和onCompleted()。

明確了這些概念以后，讓我們來看一些例子。

案例1：基礎

現在要寫一個用來展示一個顏色列表的Activity。 我們要寫一個能發送一個字符串列表、然后結束的Observeable。而后我們會通過這個字符串列表來填充顏色列表，這里要使用到 Observable.just()方法。由這個方法創建的Observable對象的特點是：所有Observer一旦訂閱這個Observable就 會立即調用onNext()方法並傳入Observable.just()的參數，而后因為Observable沒有數據可以發送 了，onComplete()方法會被調用。

Observable<List<String>> listObservable = Observable.just(getColorList());

注意這里的getColorList()是一個不耗時的方法。雖然現在看來這個方法無足輕重，但一會我們會回到這個方法。

下一步，我們寫一個Observer來觀察Observable。

listObservable.subscribe(new Observer<List<String>>() { 

    @Override 
    public void onCompleted() { } 

    @Override 
    public void onError(Throwable e) { } 

    @Override
    public void onNext(List<String> colors) {
        mSimpleStringAdapter.setStrings(colors);
    }
});

而后神奇的事情就發生了。如我剛才所說，一旦通過subscribe()方法訂閱Observable，就會發生一系列事情：

1. onNext()方法被調用，被發送的顏色列表會作為參數傳入。

2. 既然不再有數據可以發送（我們在Observable.just()中只讓Observable發送一個數據），onComplete()方法會被調用。

請記住：通過Observable被訂閱后的行為來區分它們。

在這個例子中我們不關心Observable何時完成數據的傳輸，所以我們不用在onComplete()方法里寫代碼。而且在這里不會有異常拋出，所以我們也不用管onError()方法。

寫了這么多你可能覺得很多余，畢竟我們本可以在adapter中直接設置作為數據源的顏色列表。請帶着這個疑問，和我看下面這個更有趣一些的例子。

案例2：異步加載

在這里我們要寫一個顯示電視劇列表的Activity。在Android中RxJava的主要用途就在於異步數據加載。首先讓我們寫一個Observable：

Observable<List<String>> tvShowObservable = Observable.fromCallable(new Callable<List<String>>() { 

    @Override 
    public List<String> call() { 
        return mRestClient.getFavoriteTvShows(); 
    }
});

在剛才的例子中，我們使用Observable.just()來創建Observable，你可能認為在這里可以通過Observable.just(mRestClient.getFavoriteTvShows())來創建Observable。

但在這里我們不能這么做，因為mRestClient.getFavoriteTvShows()會發起網絡請求。如果在這里我們使用Observable.just()，mRestClient.getFavoriteTvShows()會被立即執行並阻塞UI線程。

使用Observable.fromCallable()方法有兩點好處：

1. 獲取要發送的數據的代碼只會在有Observer訂閱之后執行。

2. 獲取數據的代碼可以在子線程中執行。

這兩點好處有時可能非常重要。現在讓我們訂閱這個Observable。

mTvShowSubscription = tvShowObservable
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<List<String>>() {

        @Override 
        public void onCompleted() { }

        @Override 
        public void onError(Throwable e) { }

        @Override 
        public void onNext(List<String> tvShows){
            displayTvShows(tvShows);
        }
    });

讓我們一個方法一個方法地來看這段代碼。subscribeOn會修改我們剛剛創建的Observable。在默認情況下Observable的所 有代碼，包括剛才說到的只有在被訂閱之后才會執行的代碼，都會在執行subscribe()方法的線程中運行。而通過subscribeOn()方法，這 些代碼可以在其他線程中執行。但具體是哪個線程呢？

在這個例子中我們讓代碼在”IO Scheduler”中執行（Schedulers.io()）。現在我們可以只把Scheduler當做一個可以工作的子線程，這個描述對於現在的我們已經足夠了，不過這其中還有更深層次的內容。

不過我們的確遇到了一個小障礙。既然Observable會在IO Scheduler中運行，那么它與Observer的連接也會在IO Scheduler中完成。這就意味着Observer的onNext()方法也會在IO Scheduler中運行，而onNext()方法會操作UI中的View，但View只能在UI主線程中操作。

事實上解決這個問題也很簡單，我們可以告訴RxJava我們要在UI線程中觀察這個Observable，也就是，我們想讓onNext()方法在 UI線程中執行。這一點我們可以通過在observeOn()方法中指定另一個Scheduler來完成，在這里也就是 AndroidSchedules.mainThread()所返回的Scheduler(UI線程的Scheduler)。

而后我們調用subscribe()方法。這個方法最重要，因為Callable只會在有Observer訂閱后運行。還記得剛才我說Observable通過其被訂閱后的行為來區分嗎？這就是一個很好的例子。

還有最后一件事。這個mTvShowSubscription到底是什么？每當Observer訂閱Observable時就會生成一個 Subscription對象。一個Subscription代表了一個Observer與Observable之間的連接。有時我們需要操作這個連接， 這里拿在Activity的onDestroy()方法中的代碼舉個例子：

if (mTvShowSubscription != null && !mTvShowSubscription.isUnsubscribed()) {
    mTvShowSubscription.unsubscribe();
}

如果你與多線程打過交道，你肯定會意識到一個大坑：當Activity執行onDestroy()后線程才結束（甚至永不結束）的話，就有可能發生內存泄漏與NullPointerException空指針異常。

Subscription就可以解決這個問題，我們可以通過調用unsubscribe()方法告訴Observable它所發送的數據不再被 Observer所接收。在調用unsubscribe()方法后，我們創建的Observer就不再會收到數據了，同時也就解決了剛才說的問題。

說到這里難點已經過去，讓我們來總結一下：

• Observable.fromCallable()方法可以拖延Observable獲取數據的操作，這一點在數據需要在其他線程獲取時尤其重要。

• subscribeOn()讓我們在指定線程中運行獲取數據的代碼，只要不是UI線程就行。

• observeOn()讓我們在合適的線程中接收Observable發送的數據，在這里是UI主線程。

• 記住要讓Observer取消訂閱以免Observable異步加載數據時發生意外。

案例3：使用Single

這次我們還是寫一個展示電視劇列表的Activity，但這次我們走一種更簡單的風格。Observable挺好用的，但在某些情況下過於重量級。比如說，你可能一經發現在過去的兩個方法中我們只是讓Observable發送一個數據，而且我們從來也沒寫過onComplete()回調方法。

其實呢，Observable還有一個精簡版，叫做Single。Single幾乎和Observable一模一樣，但其回調方法不是onComplete()/onNext()/onError()，而是onSuccess()/onError()。

我們現在把剛才寫過的Observable用Single重寫一遍。首先我們要創建一個Single:

Single<List<String>> tvShowSingle = Single.fromCallable(new Callable<List<String>>() { 
    @Override
    public List<String> call() throws Exception {
        mRestClient.getFavoriteTvShows(); 
    }
});

然后訂閱一下：

mTvShowSubscription = tvShowSingle
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new SingleSubscriber<List<String>>() {

        @Override 
        public void onSuccess(List<String> tvShows) {
            displayTvShows(tvShows); 
        }

        @Override 
        public void onError(Throwable error) {
            displayErrorMessage(); 
        } 
    });

這段代碼和剛才很像，我們調用subscribeOn()方法以確保getFavoriteTvShows()在子線程中執行。而后我們調用observeOn()以確保Single的數據被發送到UI線程。

但這次我們不再使用Observer，而是使用一個叫SingleSubscriber的類。這個類和Observer非常像，只不過它只有上述兩 個方法：onSuccess()和onError()。SingleSubscriber之於Single就如Observer之於 Observable。

訂閱一個Single的同時也會自動創建一個Subscription對象。這里的Subscription和案例2中沒有區別，一定要在onDestroy()中解除訂閱。

最后一點：在這里我們添加了處理異常的代碼，所以如果mRestClient出了問題，onError()就會被調用。建議你親手寫一個案例玩一玩，體驗一下有異常時程序是怎么運行的。

案例4：Subjects

現在我們寫一個Activity，里面要展示一個數字並有一個自增按鈕。在看代碼之前，先介紹另一個有關RxJava的概念，Subject。Subject這個對象既是Observable又是Observer，我會把Subject想象成一個管道：從一端把數據注入，結果就會從另一端輸出。

Subject有好幾類，在這里我們使用最簡單的：PublishSubject。使用PublishSubject時，一旦數據從一端注入，結果會立即從另一端輸出。

首先我們要寫這個管道的輸出端。剛才說了Subject也是Observable，也就是說我們可以像觀察任何一個Observable一樣觀察 它。這段代碼的功能就是觀察管道的輸出端到底輸出了什么。我們在這里寫一個很簡單的Observer來更新mCounterDisplay控件。

mCounterEmitter = PublishSubject.create(); 
mCounterEmitter.subscribe(new Observer<Integer>() {

    @Override
    public void onCompleted() { } 

    @Override
    public void onError(Throwable e) { } 

    @Override
    public void onNext(Integer integer) { 
        mCounterDisplay.setText(String.valueOf(integer));
    } 
});

與前面的幾個例子不同，在這個例子中onNext()會被調用多次。每次發送新的數據時，mCounterDisplay都會展示新的數據。但是PublishSubject怎么發送數據呢？讓我們看一下mIncrementButton的監聽代碼。

mIncrementButton.setOnClickListener(new View.OnClickListener() {

    @Override 
    public void onClick(View v) { 
        mCounter++;
        mCounterEmitter.onNext(mCounter);
    }
});

可以看到mIncrementButton在onClick()回調方法中做了兩件事情：

1. 讓mCounter變量自增。

2. 調用mCounterEmitter的onNext()方法並傳入mCounter。

由於Subject同時也是Observer，所以它也有onNext()方法，因此我們可以通過調用onNext()方法把數據注入管道的輸入端，可以理解為同我們在一端中觀察自增按鈕是否被點擊，然后把信息告知管道另一端的Observer。

案例5：Map()

我們現在要寫一個只顯示一個數字的Activity。這將是一個很簡單的Activity，因為我們要在這里使用map方法。如果你接觸過函數式編程，你可能對map並不陌生。你可以把map當做一個方法，它接收一個數據，然后輸出另一個數據，當然輸入輸出的兩個數據之間是有聯系的。

我們先寫一個只發送一個數字4的Single對象。

Single.just(4).map(new Func1<Integer, String>() { 

    @Override 
    public String call(Integer integer) { 
        return String.valueOf(integer);
    } 
}).subscribe(new SingleSubscriber<String>() { 

    @Override 
    public void onSuccess(String value) { 
        mValueDisplay.setText(value); 
    } 

    @Override 
    public void onError(Throwable error) { } 
}); 

我們最終要顯示Single所發送的數據，但首先我們需要將這個數據從Integer轉為String，而這里的解決方法就是使用map()函數。 正如剛才所說，map接收一個數據，進行處理而后輸出它，這正是我們需要的。現在Single會發送數字4，我們使用map()方法將其轉為 String，而后交給Observer去展示它。

這個例子中對於map方法的使用很輕量，不過map可是非常強大的，在下一個例子中你可以看到，map可以被用來執行任意代碼，在處理數據方面起到很重要的作用。

案例6：綜合使用

現在我們要寫一個用來根據名字搜索城市的Activity。在這個Activity中，我們要使用在這兩篇文章中所學的所有知識並寫一個比較大的例子。同時還要介紹一個新的概念：deboundce。開始。

現在我們要寫一個PublishSubject，並能接收用書輸入進輸入框的數據，而后根據輸入獲取符合的列表，並展示。

mTextWatchSubscription = mSearchResultsSubject
    .debounce(400, TimeUnit.MILLISECONDS)
    .observeOn(Schedulers.io())
    .map(new Func1<String, List<String>>() {

        @Override 
        public List<String> call(String s) { 
            return mRestClient.searchForCity(s); 
        } 
    })
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<List<String>>() { 

        @Override 
        public void onCompleted() { }

        @Override 
        public void onError(Throwable e) { } 

        @Override
        public void onNext(List<String> cities) {
            handleSearchResults(cities); 
        }
    });

mSearchInput.addTextChangedListener(new TextWatcher() {

    @Override 
    public void beforeTextChanged(CharSequence s, int start, int count, int after) { }

    @Override 
    public void onTextChanged(CharSequence s, int start, int before, int count) {
        mSearchResultsSubject.onNext(s.toString()); 
    }

    @Override 
    public void afterTextChanged(Editable s) { } 
});

這段代碼有不少內容，讓我們一點一點分析。

首先你會看到debounce()方法。這是啥？有啥用？如果你看一下我們是如何給輸入框添加監聽器的，你會發現每當輸入的內容改變時都會有輸入發 送到mSearchResultsSubject，不過我們不想讓用戶每點一個鍵都向服務器請求一次。我們想等一會，等用戶停止輸入（代表差不多輸完）的 時候再請求服務器。

而debounce()方法就是做這個的。這個方法告訴mSearchResultsSubject在沒有數據傳入達400毫秒時才發送數據。意思 就是，僅當用戶400ms都沒有改變輸入內容時，Subject才會發送最新的搜索字符串。這樣以來我們就不會進行無意義的網絡請求了，UI也不會每輸入 一個字符都更新。

我們想通過RestClient來訪問服務器，而因為RestClient涉及IO操作，我們需要在IO Scheduler中進行這個操作，所以要寫observeOn(Schedulers.io())。

好了，現在我們會把搜索字段發送到IO Scheduler中，在這里map就要發揮作用了，我們在map方法中通過關鍵字獲取搜索結果的列表。在map中我們可以調用任意外部方法，在這里使用RestClient獲取搜索結果。

因為map方法會在IO Scheduler中運行，而我們又要用其返回值填充View，所以要重新切換到UI線程，所以要寫 observeOn(AndroidSchedulers.mainThread())。現在搜索結果會被發送到UI線程。要注意兩個 observeOn()方法的順序，這一點至關重要。現在我們總結一下數據發送的順序。

mSearchResultsSubject 
            |
            |
            V
        debounce
          ||| 
          |||
           V 
          map 
           | 
           | 
           V
        observer

一個豎杠代表數據在UI線程中發送，三個豎杠代表數據在IO Scheduler中發送。

最終，我們獲得搜索結果，並展示給用戶。

結語

有關RxJava就說這么多了，希望這篇文章能幫你了解RxJava的基礎。強烈建議你自己探索有關RxJava的其他方面。如果你有問題或者只是想說點什么，歡迎在下方留言。

歡迎關注我的公眾號，將零碎時間都用在刷干貨上！

原文地址： 鏈接：https://medium.com/@kurtisnusbaum/rxandroid-basics-part-1-c0d5edcf6850