在Android开发中使用RxJava

发表于 5年以前 | 总阅读数：3474 次

ReactiveX是专注于异步工作的API，它将异步事件的处理与观察者模式、迭代器模式及函数式编程相结合了起来。实时地处理返回数据是在工程中经常出现的情景，所以使用高效、可拓展的方式来解决这种问题非常重要。ReactiveX通过观察者模式以及操作符来提供灵活地处理异步通信的方式，你不用再去关注线程创造与同步这些繁琐的事情。

RxJava介绍

RxJava是一个开源的实现ReactiveX的工具。这里面有两种主要的类：Observalbe和Subscriber。在RxJava中，Observable类产生异步数据或事件，Subscriber类对这些数据和事件进行操作。正常的工作流程就是Observable产生一系列的数据或事件，然后完成或者产生异常。一个Observable可以拥有多个Subscriber，每一个被生成的事件都会触发被绑定的Subscriber的onNext()方法。当一个Observable生成完所有事件后，所有绑定的Subscriber的onCompleted()方法会被调用（在发生异常时会调用onError()方法）。现在我们对这两个类有了初步的认识，可以开始了解如何创建与订阅一个Observer了：

Observable integerObservable = Observable.create(new Observable.OnSubscribe() {
   @Override
   public void call(Subscriber subscriber) {
       subscriber.onNext(1);
       subscriber.onNext(2);
       subscriber.onNext(3);
       subscriber.onCompleted();
   }
});

我们创建的这个Observable产生了1、2、3三条数据，现在需要创建Subscriber来处理它们：

Subscriber integerSubscriber = new Subscriber() {
   @Override
   public void onCompleted() {
       System.out.println("Complete!");
   }

   @Override
   public void onError(Throwable e) {

   }

   @Override
   public void onNext(Integer value) {
       System.out.println("onNext: " + value);
   }
};

你可以看到，这个Subscriber将每一个收到的数据打印了出来。当你创建完Observable与Subscriber后，你需要将它们用Observable.subscribe()方法连接起来。

integerObservable.subscribe(integerSubscriber);
// 输出:
// onNext: 1
// onNext: 2
// onNext: 3
// Complete!

这整个过程，可以使用Observable.just()来简化数据的生成过程，再创建Subscriber的匿名内部类来简化处理过程：

Observable.just(1, 2 ,3).subscribe(new Subscriber() {
   @Override
   public void onCompleted() {
       System.out.println("Complete!");
   }

   @Override
   public void onError(Throwable e) {}

   @Override
   public void onNext(Integer value) {
       System.out.println("onNext: " + value);
   }
});

操作符

只是创建与订阅Observable非常简单，然而这只是RxJava的冰山一角。任何Observable都可以将它的输出通过“操作符”来进行预处理，一个Observable可以绑定多个操作符来进行链式处理。举个例子，前面使用的Observable只是输出简单的数字，但是我们只想对奇数进行处理，可以通过设置filter来实现：

Observable.just(1, 2, 3, 4, 5, 6) // add more numbers
       .filter(new Func1() {
           @Override
           public Boolean call(Integer value) {
               return value % 2 == 1;
           }
       })
       .subscribe(new Subscriber() {
            @Override
            public void onCompleted() {
                System.out.println("Complete!");
            }

            @Override
                public void onError(Throwable e) {
            }

            @Override
            public void onNext(Integer value) {
                System.out.println("onNext: " + value);
            }
        });


// Outputs:
// onNext: 1
// onNext: 3
// onNext: 5
// Complete!

filter()操作符定义了一个函数，它只会在输出为奇数的时候返回true，输出为偶数的时候返回false。返回false的输出是不会发送到Subscriber的。值得注意的是filter函数返回了一个Observable，我们可以进行链式处理。比如现在希望找到所有奇数的平方根，你可以在Subscriber获取到数字的时候来处理这个逻辑，但是这样的话就不能对做平方根后的数字进行进一步处理了。你可以在filter操作符后加一个map操作符来实现这个逻辑：

Observable.just(1, 2, 3, 4, 5, 6) // add more numbers
       .filter(new Func1() {
           @Override
           public Boolean call(Integer value) {
               return value % 2 == 1;
           }
       })
       .map(new Func1() {
           @Override
           public Double call(Integer value) {
               return Math.sqrt(value);
           }
       })
       .subscribe(new Subscriber() { // notice Subscriber type changed to 
           @Override
           public void onCompleted() {
               System.out.println("Complete!");
           }

           @Override
           public void onError(Throwable e) { }

           @Override
           public void onNext(Double value) {
               System.out.println("onNext: " + value);
           }
       });

// Outputs:
// onNext: 1.0
// onNext: 1.7320508075688772
// onNext: 2.23606797749979
// Complete!

链式操作符是RxJava不可分割的一块内容，它简便、可拓展，功能强大。现在你对Observable与Subscriber的联系有了进一步的认识，我们可以开始下一个话题：将RxJava与Android开发结合起来。

简化Android多线程开发

在Android开发中，我们常常需要在后台线程中完成一些耗时工作，当后台线程完成工作之后通知主线程来显示结果。Android Sdk中有很多类可以完成多这种工作，Asynctask、Loader、Service等等。然而它们有时候并不是最好的选择。Asynctask经常会造成内存泄露；与ContentProvider搭配的CursorLoader使用起来非常繁琐，并且需要大量样板代码；Service通常是用于执行需要长时间在后台运行的任务。那么我们来看看RxJava怎么来解决这些问题。

在下面这个布局中，有一个按钮，它会启动一个耗时任务，并且将任务的进度展示到ProgressBar上。

<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
   xmlns:app="http://schemas.android.com/apk/res-auto"
   android:id="@+id/root_view"
   android:layout_width="match_parent"
   android:layout_height="match_parent"
   android:fitsSystemWindows="true"
   android:orientation="vertical">

   <android.support.v7.widget.Toolbar
       android:id="@+id/toolbar"
       android:layout_width="match_parent"
       android:layout_height="?attr/actionBarSize"
       android:background="?attr/colorPrimary"
       app:popupTheme="@style/AppTheme.PopupOverlay"
       app:theme="@style/ThemeOverlay.AppCompat.Dark.ActionBar" />

   <Button
       android:id="@+id/start_btn"
       android:layout_width="wrap_content"
       android:layout_height="wrap_content"
       android:layout_gravity="center_horizontal"
       android:text="@string/start_operation_text" />

   <ProgressBar
       android:layout_width="wrap_content"
       android:layout_height="wrap_content"
       android:layout_gravity="center_horizontal"
       android:indeterminate="true" />

</LinearLayout>

一旦按钮按下之后，我们禁用按钮，同时在后台线程中开始做耗时任务，当任务结束之后会弹出一个SnackBar。现在使用一个简单的AsyncTask来完成如下这个耗时任务longRunningOperation()。

public String longRunningOperation() {
   try {
       Thread.sleep(2000);
   } catch (InterruptedException e) {
       // error
   }
   return "Complete!";
}

private class SampleAsyncTask extends AsyncTask {

   @Override
   protected String doInBackground(Void... params) {
       return longRunningOperation();
   }

   @Override
   protected void onPostExecute(String result) {
       Snackbar.make(rootView, result, Snackbar.LENGTH_LONG).show();
       startAsyncTaskButton.setEnabled(true);
   }
}

现在要怎么用RxJava替换掉呢？首先需要在app模块的gradle脚本中添加依赖：io.reactivex:rxjava:1.0.14。然后创建一个Observable来运行耗时任务，调用Observable.create方法来实现这个功能：

final Observable operationObservable = Observable.create(new Observable.OnSubscribe() {
   @Override
   public void call(Subscriber subscriber) {
       subscriber.onNext(longRunningOperation());
       subscriber.onCompleted();
   }
});

现在创建的这个Observable会执行longRunningOperation()，在执行完毕后通知Subscriber，并且调用Subscriber的onCompleted()方法（注意：耗时任务在Subscriber与Observable绑定之前不会被执行）。接下来，我们需要在按钮点击事件上将Observable与Subscriber绑定。

startRxOperationButton.setOnClickListener(new View.OnClickListener() {
   @Override
   public void onClick(final View v) {
       v.setEnabled(false);
       operationObservable.subscribe(new Subscriber() {
           @Override
           public void onCompleted() {
               v.setEnabled(true);
           }

           @Override
           public void onError(Throwable e) {}

           @Override
           public void onNext(String value) {
               Snackbar.make(rootView, value, Snackbar.LENGTH_LONG).show();
           }
       });
   }
});

接下来我们就可以运行这个应用了。但是你会发现，点击按钮之后ProgressBar并不会有反应！（这是因为UI主线程被耗时任务阻塞了）。我们还没有指定Observable的运行线程，也没有指定Subscriber的运行线程。RxJava调度的功能就在这里体现：我们可以指定两个不同的线程，一个是运行线程，一个是监视线程。使用Observable.observeOn()函数，你可以定义一个线程用来监视Observable运行，并且查看是否Observable有新的输出（Subscriber的onNext，onCompleted， onError方法都是在这个监视线程中执行的）。调用Observable.subscribeOn()方法可以指定一个线程运行耗时任务。RxJava原本将两个任务都安排在一个线程中执行，而你可以通过调用observeOn与subcribeOn方法来进行多线程操作。RxJava封装了很多调度工具，比如Schedulers.io()（阻塞I/O操作），Schedulers.computation()（进行计算工作），Schedulers.newThread()（创建新线程）。不过在Android开发中，你应该关心的是怎么将代码执行在主线程中。我们可以通过RxAndroid库来完成这个工作！

RxAndroid是RxJava的轻量级拓展工具，它提供了运行在主线程上的Scheduler，或者运行在任意Handler线程上的Scheduler！这样一来，我们就可以在后台线程中运行耗时工作，同时在主线程中处理结果。要使用RxAndroid，你需要添加依赖io.reactivex:rxandroid:1.0.1。

final Observable operationObservable = Observable.create(new Observable.OnSubscribe() {
   @Override
   public void call(Subscriber subscriber) {
       subscriber.onNext(longRunningOperation());
       subscriber.onCompleted();
   }
}).subscribeOn(Schedulers.io()) // subscribeOn the I/O thread
  .observeOn(AndroidSchedulers.mainThread()); // observeOn the UI Thread

以上经过修改代码会在Schedulers.io线程上运行耗时工作，Subscriber会在主线程中处理输出结果。现在我们运行应用，点击按钮，我们的耗时任务没有阻塞UI线程。RxJava 1.0.13同时提供了Single类，它只会产生一个结果，之后马上执行onComplete()方法。示例如下：

Subscription subscription = Single.create(new Single.OnSubscribe() {
           @Override
           public void call(SingleSubscriber singleSubscriber) {
               String value = longRunningOperation();
               singleSubscriber.onSuccess(value);
           }
       })
       .subscribeOn(Schedulers.io())
       .observeOn(AndroidSchedulers.mainThread())
       .subscribe(new Action1() {
           @Override
           public void call(String value) {
               // onSuccess
               Snackbar.make(rootView, value, Snackbar.LENGTH_LONG).show();
           }
       }, new Action1() {
           @Override
           public void call(Throwable throwable) {
               // handle onError
           }
       });

在订阅Single的时候，只有onSuccess()、onError()可以监听。同时你可以调用Single.mergeWith()操作符来将多个Single合成一个Observable，这个被合成的Observable会依次输出所有Single的结果。

防止内存泄露

我们之前提到，使用AsyncTask的一大缺点就是它可能会造成内存泄露。当它们持有的Activity/Fragment的引用没有正确处理时就会这样。不幸的是，RxJava并不会自动防止这种情况发生，好在它可以很容易地防止内存泄露。Observable.subscribe()方法会返回一个Subscription对象，这个对象仅仅有两个方法：isSbscribed()与unsubscribe()。你可以在Activity/Fragment的onDestroy方法中调用Subscription.isSubscribed()检测是否这个异步任务仍在进行。如果它仍在进行，则调用unsubscribe()方法来结束任务，从而释放其中的强引用，防止内存泄露。如果你使用了多个Observable与Subscriber，那么你可以将它们添加到CompositeSubscription中，并调用CompositeSubscription.unsubscribe()结束所有的任务。

结束语

RxJava为Android提供了一个非常棒的多线程同步的解决方案。在Android开发中，让耗时任务在后台线程运行，并将结果在主线程中展示，是非常重要的。同时它的操作符模式为这个工具起到了锦上添花的作用。使用RxJava的时候，你需要对整个工具库有一定理解，才能将其发挥最大的作用。所以花时间去系统地学习这个库是非常值得的。未来我还将会在我的博客中讨论RxJava的几个特性：冷热Observable的比较、处理反压力、Rx子类。本节中的样本代码可以从我的Github中找到。

特别介绍：RetroLambda

Java 8 引入了Lambda表达式，可惜的是Android不支持Java 8，所以我们不能使用RxJava与这一特性结合的功能。幸运的是，有一个工具叫RetroLambda可以将Lambda表达式与低版本的Java兼容。同时还有一个Gradle插件可以在Android开发中使用。使用了Lambda表达式，你的RxJava代码会更加简洁：

final Observable operationObservable = Observable.create(
       (Subscriber subscriber) -> {
           subscriber.onNext(longRunningOperation());
           subscriber.onCompleted();
       })
       .subscribeOn(Schedulers.io())
       .observeOn(AndroidSchedulers.mainThread());

startRxOperationButton = (Button) findViewById(R.id.start_rxjava_operation_btn);
startRxOperationButton.setOnClickListener(v -> {
   v.setEnabled(false);
   operationObservable.subscribe(
           value -> Snackbar.make(rootView, value, Snackbar.LENGTH_LONG).show(),
           error -> Log.e("TAG", "Error: " + error.getMessage()),
           () -> v.setEnabled(true));
});

Lambda表达式减少了很多RxJava的样本代码，我强烈推荐将RetroLambda引入工程中使用。它给开发带来的好处不仅仅于RxJava上（你会注意到OnCliclListener也是用Lambda表达式设置的）!