快速开始

RxJava 快速开始

在 Android 开发中，我们往往需要在后台线程中执行一些费时操作，然后传递到 UI 线程中进行界面设置。假设需要找出某个本地 url 列表中的图片本地目录，并且加载对应的图片，展现在 UI 上。

基于多线程与回调的模型

new Thread() {
@Override
public void run() {
    super.run();
    for (String url : urls) {
        if (url.endsWith(".png")) {
            final Bitmap bitmap = getBitmap(url);
            getActivity().runOnUiThread(new Runnable() {
                @Override
                public void run() {
                    imageView.setImageBitmap(bitmap);
                }
            });
        }
    }
}
}.start();

基于 RxJava 的响应式模型

Observable.from(urls)
.filter(new Func1<String, Boolean>() {
    @Override
    public Boolean call(String url) {
        return url.endsWith(".png");
    }
})
.map(new Func1<String, Bitmap>() {
    @Override
    public Bitmap call(String url) {
        return getBitmap(url);
    }
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Bitmap>() {
    @Override
    public void call(Bitmap bitmap) {
        imageView.addImage(bitmap);
    }
});

Quick Start

Installation

笔者在 J2EE 领域还是倾向于使用 Maven，直接在 pom 文件中添加如下依赖即可：

<dependency>
    <groupId>io.reactivex</groupId>
    <artifactId>rxjava</artifactId>
    <version>1.1.3</version>
</dependency>

添加了 Pom 依赖项之后，即可以引入 Observable 以及 Subscribe 对象，源代码参考这里:

Hello World

package wx.toolkits.sysproc.concurrence.rxjava;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.concurrent.*;
import java.util.List;
import rx.Observable;

public class ObservableCreator {

  public static void fromArrayList() {
    List<Integer> arrayList = Arrays.asList(new Integer[] { 1, 2, 3, 4, 5 });

    /*
       * Example using single-value lambdas (Func1)
       */
      Observable
      .from(arrayList)
      .filter(
        v -> {
          return v < 4;
        }
      )
      .subscribe(
        value -> {
          System.out.println("Value: " + value);
        }
      );

    /*
       * Example with 'reduce' that takes a lambda with 2 arguments (Func2)
       */
      Observable
      .from(arrayList)
      .reduce(
        (seed, value) -> {
          // sum all values from the sequence
          return seed + value;
        }
      )
      .map(
        v -> {
          return "DecoratedValue: " + v;
        }
      )
      .subscribe(
        value -> {
          System.out.println(value);
        }
      );
  }

  public static void fromFuture() {
    Callable<List<Integer>> callable =
      () -> {
        List<Integer> arrayList = Arrays.asList(
          new Integer[] { 1, 2, 3, 4, 5 }
        );
        return arrayList;
      };

    ExecutorService executorService = Executors.newSingleThreadExecutor();

    Future future = executorService.submit(callable);

    Observable
      .from(future)
      .subscribe(
        value -> {
          System.out.println(value);
        }
      );
  }

  public static void main(String args[]) {
    //演示从ArrayList创建
    ObservableCreator.fromArrayList();

    //演示從Callable創建
    ObservableCreator.fromFuture();
  }
}

Terminology

RxJava 有四个基本概念：Observable (可观察者，即被观察者)、Observer (观察者)、subscribe (订阅)、事件。Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在需要的时候发出事件来通知 Observer。与传统观察者模式不同，RxJava 的事件回调方法除了普通事件 onNext() (相当于 onClick() / onEvent())之外，还定义了两个特殊的事件：onCompleted() 和 onError()。

onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的 onNext() 发出时，需要触发 onCompleted() 方法作为标志。
onError(): 事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。
在一个正确运行的事件序列中, onCompleted() 和 onError() 有且只有一个，并且是事件序列中的最后一个。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在队列中调用了其中一个，就不应该再调用另一个。

RxJava 的观察者模式大致如下图：

Observer/Subscriber(观察者)

Observer 即观察者，它决定事件触发的时候将有怎样的行为。RxJava 中的 Observer 接口的实现方式：

Observer<String> observer = new Observer<String>() {
    @Override
    public void onNext(String s) {
        Log.d(tag, "Item: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

除了 Observer 接口之外，RxJava 还内置了一个实现了 Observer 的抽象类：Subscriber。Subscriber 对 Observer 接口进行了一些扩展，但他们的基本使用方式是完全一样的：

Subscriber<String> subscriber = new Subscriber<String>() {
    @Override
    public void onNext(String s) {
        Log.d(tag, "Item: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

不仅基本使用方式一样，实质上，在 RxJava 的 subscribe 过程中，Observer 也总是会先被转换成一个 Subscriber 再使用。所以如果你只想使用基本功能，选择 Observer 和 Subscriber 是完全一样的。它们的区别对于使用者来说主要有两点：

onStart(): 这是 Subscriber 增加的方法。它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。这是一个可选方法，默认情况下它的实现为空。需要注意的是，如果对准备工作的线程有要求(例如弹出一个显示进度的对话框，这必须在主线程执行)，onStart() 就不适用了，因为它总是在 subscribe 所发生的线程被调用，而不能指定线程。要在指定的线程来做准备工作，可以使用 doOnSubscribe() 方法，具体可以在后面的文中看到。
unsubscribe(): 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。在这个方法被调用后，Subscriber 将不再接收事件。一般在这个方法调用前，可以使用 isUnsubscribed() 先判断一下状态。unsubscribe() 这个方法很重要，因为在 subscribe() 之后，Observable 会持有 Subscriber 的引用，这个引用如果不能及时被释放，将有内存泄露的风险。所以最好保持一个原则：要在不再使用的时候尽快在合适的地方(例如 onPause() onStop() 等方法中)调用 unsubscribe() 来解除引用关系，以避免内存泄露的发生。

最近更新于 0001-01-01