RxJava2 實(shí)戰(zhàn)系列文章

RxJava2 實(shí)戰(zhàn)知識(shí)梳理(1) - 后臺(tái)執(zhí)行耗時(shí)操作星持，實(shí)時(shí)通知 UI 更新
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(2) - 計(jì)算一段時(shí)間內(nèi)數(shù)據(jù)的平均值
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(3) - 優(yōu)化搜索聯(lián)想功能
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(4) - 結(jié)合 Retrofit 請(qǐng)求新聞資訊
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(5) - 簡(jiǎn)單及進(jìn)階的輪詢(xún)操作
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(6) - 基于錯(cuò)誤類(lèi)型的重試請(qǐng)求
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(7) - 基于 combineLatest 實(shí)現(xiàn)的輸入表單驗(yàn)證
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(8) - 使用 publish + merge 優(yōu)化先加載緩存碳想，再讀取網(wǎng)絡(luò)數(shù)據(jù)的請(qǐng)求過(guò)程
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(9) - 使用 timer/interval/delay 實(shí)現(xiàn)任務(wù)調(diào)度
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(10) - 屏幕旋轉(zhuǎn)導(dǎo)致 Activity 重建時(shí)恢復(fù)任務(wù)
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(11) - 檢測(cè)網(wǎng)絡(luò)狀態(tài)并自動(dòng)重試請(qǐng)求
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(12) - 實(shí)戰(zhàn)講解 publish & replay & share & refCount & autoConnect
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(13) - 如何使得錯(cuò)誤發(fā)生時(shí)不自動(dòng)停止訂閱關(guān)系
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(14) - 在 token 過(guò)期時(shí)钻趋，刷新過(guò)期 token 并重新發(fā)起請(qǐng)求
RxJava2 實(shí)戰(zhàn)知識(shí)梳理(15) - 實(shí)現(xiàn)一個(gè)簡(jiǎn)單的 MVP + RxJava + Retrofit 應(yīng)用

一川陆、前言

接觸RxJava2已經(jīng)很久了，也看了網(wǎng)上的很多文章蛮位，發(fā)現(xiàn)基本都是在對(duì)RxJava的基本思想介紹之后较沪，再去對(duì)各個(gè)操作符進(jìn)行分析鳞绕，但是看了之后感覺(jué)過(guò)了不久就忘了。

偶然的機(jī)會(huì)看到了開(kāi)源項(xiàng)目 RxJava-Android-Samples尸曼，這里一共介紹了十六種RxJava2的使用場(chǎng)景们何，它從實(shí)際的應(yīng)用場(chǎng)景出發(fā)介紹RxJava2的使用，特別適合對(duì)于RxJava2已經(jīng)有初步了解的開(kāi)發(fā)者進(jìn)一步地去學(xué)習(xí)如何將其應(yīng)用到實(shí)際開(kāi)發(fā)當(dāng)中控轿。

因此冤竹，我打算跟著這個(gè)項(xiàng)目的思路編寫(xiě)一系列實(shí)戰(zhàn)的介紹并完成示例代碼編寫(xiě)，并對(duì)該實(shí)例中用到的知識(shí)進(jìn)行介紹解幽，做到學(xué)以致用贴见。下面烘苹，就開(kāi)始第一個(gè)例子的學(xué)習(xí)躲株，源碼的倉(cāng)庫(kù)為：RxSample。

二镣衡、示例

2.1 應(yīng)用場(chǎng)景

當(dāng)我們需要進(jìn)行一些耗時(shí)操作霜定，例如下載、訪問(wèn)數(shù)據(jù)庫(kù)等廊鸥，為了不阻塞主線程望浩，往往會(huì)將其放在后臺(tái)進(jìn)行處理，同時(shí)在處理的過(guò)程中惰说、處理完成后通知主線程更新UI磨德，這里就涉及到了后臺(tái)線程和主線程之間的切換。首先回憶一下吆视，在以前我們一般會(huì)用以下兩種方式來(lái)實(shí)現(xiàn)這一效果：

• 創(chuàng)建一個(gè)新的子線程典挑，在其run()方法中執(zhí)行耗時(shí)的操作，并通過(guò)一個(gè)和主線程Looper關(guān)聯(lián)的Handler發(fā)送消息給主線程更新進(jìn)度顯示啦吧、處理結(jié)果您觉。
• 使用AsyncTask，在其doInBackground方法中執(zhí)行耗時(shí)的操作授滓，調(diào)用publishProgress方法通知主線程琳水，然后在onProgressUpdate中更新進(jìn)度顯示，在onPostExecute中顯示最終結(jié)果般堆。

那么在孝，讓我們看一些在RxJava中如何完成這一需求。

2.2 示例代碼

我們的界面上有一個(gè)按鈕mTvDownload淮摔，點(diǎn)擊之后會(huì)發(fā)起一個(gè)耗時(shí)的任務(wù)私沮，這里我們用Thread.sleep來(lái)模擬耗時(shí)的操作，每隔500ms我們會(huì)將當(dāng)前的進(jìn)度通知主線程噩咪，在mTvDownloadResult中顯示當(dāng)前處理的進(jìn)度顾彰。

public class BackgroundActivity extends AppCompatActivity {

    private TextView mTvDownload;
    private TextView mTvDownloadResult;
    private CompositeDisposable mCompositeDisposable = new CompositeDisposable();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_background);
        mTvDownload = (TextView) findViewById(R.id.tv_download);
        mTvDownloadResult = (TextView) findViewById(R.id.tv_download_result);
        mTvDownload.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                startDownload();
            }
        });
    }

    private void startDownload() {
        final Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0; i < 100; i++) {
                    if (i % 20 == 0) {
                        try {
                            Thread.sleep(500); //模擬下載的操作极阅。
                        } catch (InterruptedException exception) {
                            if (!e.isDisposed()) {
                                e.onError(exception);
                            }
                        }
                        e.onNext(i);
                    }
                }
                e.onComplete();
            }

        });
        DisposableObserver<Integer> disposableObserver = new DisposableObserver<Integer>() {

            @Override
            public void onNext(Integer value) {
                Log.d("BackgroundActivity", "onNext=" + value);
                mTvDownloadResult.setText("Current Progress=" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d("BackgroundActivity", "onError=" + e);
                mTvDownloadResult.setText("Download Error");
            }

            @Override
            public void onComplete() {
                Log.d("BackgroundActivity", "onComplete");
                mTvDownloadResult.setText("Download onComplete");
            }
        };
        observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(disposableObserver);
        mCompositeDisposable.add(disposableObserver);
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        mCompositeDisposable.clear();
    }
}

實(shí)際的運(yùn)行結(jié)果如下：

三、示例解析

3.1 線程切換

在上面的例子中涨享，涉及到了兩種類(lèi)型的操作：

• 需要在后臺(tái)執(zhí)行的耗時(shí)操作筋搏，對(duì)應(yīng)于subscribe(ObservableEmitter<Integer> e)中的代碼。
• 需要在主線程進(jìn)行UI更新的操作厕隧，對(duì)應(yīng)于DisposableObserver的所有回調(diào)奔脐，具體的是在onNext中進(jìn)行進(jìn)度的更新；在onComplete和onError中展示最終的處理結(jié)果吁讨。

那么髓迎，這兩種類(lèi)型操作所運(yùn)行的線程是在哪里指定的呢，關(guān)鍵是下面這句：

observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(disposableObserver);

• subscribeOn(Schedulers.io())：指定observable的subscribe方法運(yùn)行在后臺(tái)線程建丧。
• observeOn(AndroidSchedulers.mainThread())：指定observer的回調(diào)方法運(yùn)行在主線程排龄。

這兩個(gè)函數(shù)剛開(kāi)始的時(shí)候很有可能弄混，我是這么記的翎朱，subscribeOn以s開(kāi)頭橄维，可以理解為“上游”開(kāi)頭的諧音，也就是上游執(zhí)行的線程拴曲。

關(guān)于這兩個(gè)函數(shù)争舞，還有一點(diǎn)說(shuō)明：多次調(diào)用subscribeOn，會(huì)以第一次的為準(zhǔn)澈灼；而多次調(diào)用observeOn則會(huì)以最后一次的為準(zhǔn)竞川，不過(guò)一般我們都不會(huì)這么干，就不舉例子了叁熔。

3.2 線程的類(lèi)型

subscribeOn/observeOn都要求傳入一個(gè)Schedulers的子類(lèi)委乌，它就代表了運(yùn)行線程類(lèi)型，下面我們來(lái)看一下都有哪些選擇：

• Schedulers.computation()：用于計(jì)算任務(wù)者疤，默認(rèn)線程數(shù)等于處理器的數(shù)量福澡。
• Schedulers.from(Executor executor)：使用Executor作為調(diào)度器，關(guān)于Executor框架可以參考這篇文章：多線程知識(shí)梳理(5) - 線程池四部曲之 Executor 框架驹马。
• Schedulers.io(?)：用于IO密集型任務(wù)革砸，例如訪問(wèn)網(wǎng)絡(luò)、數(shù)據(jù)庫(kù)操作等糯累，也是我們最常使用的算利。
• Schedulers.newThread(?)：為每一個(gè)任務(wù)創(chuàng)建一個(gè)新的線程。
• Schedulers.trampoline(?)：當(dāng)其它排隊(duì)的任務(wù)完成后泳姐，在當(dāng)前線程排隊(duì)開(kāi)始執(zhí)行效拭。
• Schedulers.single()：所有任務(wù)共用一個(gè)后臺(tái)線程。

以上是在io.reactivex.schedulers包中，提供的Schedulers缎患，而如果我們導(dǎo)入了下面的依賴(lài)慕的，那么在io.reactivex.android.schedulers下，還有額外的兩個(gè)Schedulers可選：

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

• AndroidSchedulers.mainThread()：運(yùn)行在應(yīng)用程序的主線程挤渔。
• AndroidSchedulers.from(Looper looper)：運(yùn)行在該looper對(duì)應(yīng)的線程當(dāng)中肮街。

3.3 使用 CompositeDisposable 對(duì)下游進(jìn)行管理

如果Activity要被銷(xiāo)毀時(shí)，我們的后臺(tái)任務(wù)沒(méi)有執(zhí)行完判导，那么就會(huì)導(dǎo)致Activity不能正臣蹈福回收，而對(duì)于每一個(gè)Observer眼刃，都會(huì)有一個(gè)Disposable對(duì)象用于管理绕辖，而RxJava提供了一個(gè)CompositeDisposable類(lèi)用于管理這些Disposable，我們只需要將其將入到該集合當(dāng)中擂红，在Activity的onDestroy方法中仪际，調(diào)用它的clear方法，就能避免內(nèi)存泄漏的發(fā)生篮条。

四弟头、小結(jié)

這個(gè)系列的第一篇文章，我們介紹了如何使用subscribeOn/observeOn來(lái)實(shí)現(xiàn)后臺(tái)執(zhí)行耗時(shí)任務(wù)涉茧，并通知主線程更新進(jìn)度。

更多文章疹娶，歡迎訪問(wèn)我的 Android 知識(shí)梳理系列：

• Android 知識(shí)梳理目錄：http://www.reibang.com/p/fd82d18994ce
• 個(gè)人主頁(yè)：http://lizejun.cn
• 個(gè)人知識(shí)總結(jié)目錄：http://lizejun.cn/categories/