RxJava2-后台執行耗時操作，實時通知 UI 更新(一)

一、前言

接觸RxJava2已經很久了，也看了網上的很多文章，發現基本都是在對RxJava的基本思想介紹之后，再去對各個操作符進行分析，但是看了之后感覺過了不久就忘了。

偶然的機會看到了開源項目 RxJava-Android-Samples，這里一共介紹了十六種RxJava2的使用場景，它從實際的應用場景出發介紹RxJava2的使用，特別適合對於RxJava2已經有初步了解的開發者進一步地去學習如何將其應用到實際開發當中。

因此，我打算跟着這個項目的思路編寫一系列實戰的介紹並完成示例代碼編寫，並對該實例中用到的知識進行介紹，做到學以致用。下面，就開始第一個例子的學習，源碼的倉庫為：RxSample。

二、示例

2.1 應用場景

當我們需要進行一些耗時操作，例如下載、訪問數據庫等，為了不阻塞主線程，往往會將其放在后台進行處理，同時在處理的過程中、處理完成后通知主線程更新UI，這里就涉及到了后台線程和主線程之間的切換。首先回憶一下，在以前我們一般會用以下兩種方式來實現這一效果：

• 創建一個新的子線程，在其run()方法中執行耗時的操作，並通過一個和主線程Looper關聯的Handler發送消息給主線程更新進度顯示、處理結果。
• 使用AsyncTask，在其doInBackground方法中執行耗時的操作，調用publishProgress方法通知主線程，然后在onProgressUpdate中更新進度顯示，在onPostExecute中顯示最終結果。

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

2.2 示例代碼

我們的界面上有一個按鈕mTvDownload，點擊之后會發起一個耗時的任務，這里我們用Thread.sleep來模擬耗時的操作，每隔500ms我們會將當前的進度通知主線程，在mTvDownloadResult中顯示當前處理的進度。

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

實際的運行結果如下：

三、示例解析

3.1 線程切換

在上面的例子中，涉及到了兩種類型的操作：

• 需要在后台執行的耗時操作，對應於subscribe(ObservableEmitter<Integer> e)中的代碼。
• 需要在主線程進行UI更新的操作，對應於DisposableObserver的所有回調，具體的是在onNext中進行進度的更新；在onComplete和onError中展示最終的處理結果。

那么，這兩種類型操作所運行的線程是在哪里指定的呢，關鍵是下面這句：

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

• subscribeOn(Schedulers.io())：指定observable的subscribe方法運行在后台線程。
• observeOn(AndroidSchedulers.mainThread())：指定observer的回調方法運行在主線程。

這兩個函數剛開始的時候很有可能弄混，我是這么記的，subscribeOn以s開頭，可以理解為“上游”開頭的諧音，也就是上游執行的線程。

關於這兩個函數，還有一點說明：多次調用subscribeOn，會以第一次的為准；而多次調用observeOn則會以最后一次的為准，不過一般我們都不會這么干，就不舉例子了。

3.2 線程的類型

subscribeOn/observeOn都要求傳入一個Schedulers的子類，它就代表了運行線程類型，下面我們來看一下都有哪些選擇：

• Schedulers.computation()：用於計算任務，默認線程數等於處理器的數量。
• Schedulers.from(Executor executor)：使用Executor作為調度器，關於Executor框架可以參考這篇文章：多線程知識梳理(5) - 線程池四部曲之 Executor 框架。
• Schedulers.io( )：用於IO密集型任務，例如訪問網絡、數據庫操作等，也是我們最常使用的。
• Schedulers.newThread( )：為每一個任務創建一個新的線程。
• Schedulers.trampoline( )：當其它排隊的任務完成后，在當前線程排隊開始執行。
• Schedulers.single()：所有任務共用一個后台線程。

以上是在io.reactivex.schedulers包中，提供的Schedulers，而如果我們導入了下面的依賴，那么在io.reactivex.android.schedulers下，還有額外的兩個Schedulers可選：

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

• AndroidSchedulers.mainThread()：運行在應用程序的主線程。
• AndroidSchedulers.from(Looper looper)：運行在該looper對應的線程當中。

3.3 使用 CompositeDisposable 對下游進行管理

如果Activity要被銷毀時，我們的后台任務沒有執行完，那么就會導致Activity不能正常回收，而對於每一個Observer，都會有一個Disposable對象用於管理，而RxJava提供了一個CompositeDisposable類用於管理這些Disposable，我們只需要將其將入到該集合當中，在Activity的onDestroy方法中，調用它的clear方法，就能避免內存泄漏的發生。

四、小結

這個系列的第一篇文章，我們介紹了如何使用subscribeOn/observeOn來實現后台執行耗時任務，並通知主線程更新進度。