RxJava2 实战知识梳理(1) 后台执行耗时操做，实时通知 UI 更新

1、前言

接触RxJava2已经好久了，也看了网上的不少文章，发现基本都是在对RxJava的基本思想介绍以后，再去对各个操做符进行分析，可是看了以后感受过了不久就忘了。

偶然的机会看到了开源项目 RxJava-Android-Samples，这里一共介绍了十六种RxJava2的使用场景，它从实际的应用场景出发介绍RxJava2的使用，特别适合对于RxJava2已经有初步了解的开发者进一步地去学习如何将其应用到实际开发当中。

所以，我打算跟着这个项目的思路编写一系列实战的介绍并完成示例代码编写，并对该实例中用到的知识进行介绍，作到学以至用。下面，就开始第一个例子的学习，源码的仓库为：RxSample。

2、示例

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、示例解析

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.immediate( )：在当前线程执行任务
• 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方法，就能避免内存泄漏的发生。

4、小结

这个系列的第一篇文章，咱们介绍了如何使用subscribeOn/observeOn来实现后台执行耗时任务，并通知主线程更新进度。

---

更多文章，欢迎访问个人 Android 知识梳理系列：

• Android 知识梳理目录：www.jianshu.com/p/fd82d1899…
• 我的主页：lizejun.cn
• 我的知识总结目录：lizejun.cn/categories/