RxAndroid 响应式框架介绍与使用说明

ReactiveX/RxJava 文档中文版 Android 模块

RxAndroid GitHub 主页

RxAndroid Wiki 主页

需要了解更多 RxJava 2 使用示例请看这里：

Android 异步编程探索 — 最喜爱 RxJava 2

一、介绍

RxAndroid 是 RxJava 2 的 Android 特定绑定。该模块为 RxJava 添加了最低限度的类，使得在 Android 中编写响应式组件变得简单而轻松。更具体地说，它提供了一个 Scheduler，可以在主线程或任何给定 Looper 的线程上进行调度。

• 它提供了一个可以在给定的 Android Handler 上调度 Observable 的调度器 Scheduler，特别是在 UI 主线程上。

• 它提供了一些操作符，让你可以更容易的处理 Fragment 和 Activity 的生命周期方法。

• 它提供了很多 Android 消息和通知组件的包装类，用于与 Rx 的调用链搭配使用。

• 针对常见的 Android 用例和重要的 UI，它提供了可复用的、自包含的响应式组件。

1.1 正式版本

implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'
// Because RxAndroid releases are few and far between, it is recommended you also
// explicitly depend on RxJava's latest version for bug fixes and new features.
// (see https://github.com/ReactiveX/RxJava/releases for latest 2.x.x version)
implementation 'io.reactivex.rxjava2:rxjava:2.2.4'

• RxAndroid：maven-central v2.1.0

• RxJava：maven-central v2.2.4

要获取其他二进制文件和依赖项信息，请访问 http://search.maven.org。

二、使用示例

2.1 在 UI 线程观察 (Observing)

在 Android 上，通常处理异步任务时你会在主线程上等待 (observing) 处理结果，一般情况下你使用 AsyncTask 达到这个目的。使用 RxJava，你可以使用 observeOn(AndroidSchedulers.mainThread()) 操作符声明你要在主线程等待 Observable 的结果：

Observable.just("one", "two", "three", "four", "five")
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(/* an Observer */);

这个例子中，Observable 在一个新的线程执行，结果通过 onNext() 在主线程发射。

2.2 在任意线程观察 (Observing)

前面的例子是一个普遍概念的特殊版本：Android 使用一个叫 Handler 的类将异步通信绑定到消息循环 (Looper)。为了在任意给定 Looper 的线程上观察一个 Observable，需要调用 AndroidSchedulers.from() 创建一个关联的 Scheduler：

Looper backgroundLooper = // ...
Observable.just("one", "two", "three", "four", "five")
        .observeOn(AndroidSchedulers.from(backgroundLooper))
        .subscribe(/* an Observer */)

这将在新线程上执行 Observable，并在运行 backgroundLooper的任何线程上通过 onNext() 发出结果。

2.3 记得取消订阅

在 Android 上，要在异步操作中访问框架中的对象有些棘手，那是因为 Andoid 系统可以决定销毁 (destroy) 一个 Activity，例如，当一个后台线程还在运行的时候，如果这个线程尝试访问一个已经死掉的 Activity 中的 View 对象，会导致异常退出 (Crash)。（这也会导致内存泄露，因为 Activity 已经销毁了，你的后台线程还持有它的引用。）

这仍然是在 Android 上使用 RxJava 需要关注的一个问题，但是通过使用 CompositeDisposable，你可以优雅地解决这个问题。通常来说，当你在 Activity 中订阅一个 Observable 的结果时（无论是直接的还是通过一个内部类），你必须在 onDestroy() 里取消订阅，就像下面例子里展示的那样：

    private CompositeDisposable mCompositeDisposable = new CompositeDisposable();


    @Override
    protected void onDestroy() {
        super.onDestroy();
        mCompositeDisposable.clear(); // do not send event after activity has been destroyed
    }


    private void setBitmapFromNet() {
        getObservable()
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(getObserver());
    }


    private Observable<Bitmap> getObservable() {
        return Observable.create(new ObservableOnSubscribe<Bitmap>() {
            @Override
            public void subscribe(final ObservableEmitter<Bitmap> emitter) throws Exception {
                if (!emitter.isDisposed()) {
                    Request request = new Request.Builder()
                            .url("https://img3.doubanio.com/img/musician/large/35000.jpg")
                            .build();

                    client.newCall(request).enqueue(new Callback() {
                        @Override
                        public void onFailure(Call call, IOException e) {
                            Log.d(TAG, "onFailure: " + e.getMessage());
                        }

                        @Override
                        public void onResponse(Call call, Response response) throws IOException {
                            InputStream inputStream = response.body().byteStream();
                            emitter.onNext(BitmapFactory.decodeStream(inputStream));
                            emitter.onComplete();
                        }
                    });
                }
            }
        });
    }


    private Observer<Bitmap> getObserver() {
        return new Observer<Bitmap>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "onSubscribe: 1 ");
                mCompositeDisposable.add(d);
            }

            @Override
            public void onNext(Bitmap bitmap) {
                Log.d(TAG, "onNext: 2 ");
                mImageView.setImageBitmap(bitmap);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "onError: 3 " + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete: 3 ");
            }
        };
    }

这样确保所有指向订阅者 (这个 Activity) 的引用尽快释放，不会再有通知通过 onNext() 发射给这个订阅者。

有一个问题，如果由于屏幕方向的变化导致这个 Activity 被销毁，在 onCreate() 中这个 Observable 会再次启动。你可以使用 cache 或 replay 操作符阻止它发生，这些操作符保证 Observable 在 Activity 的生命周期内存在（你可以在一个全局的缓存中保存它，比如放在 Fragment 中）。你可以使用任何操作符，只要能保证：当订阅者订阅一个已经在运行的 Observable 时，在它与 Activity 解除关联的这段时间里发射的数据都会被回放，并且来自这个 Observable 的任何离线通知都会正常分发。

三、升级须知

从 0.25.0 版本到 1.x 版本的转变很大，以下是这些转变的概述。

没有使用过老版本的可忽略这部分内容。

3.1 RxAndroid

虽然一些方法签名已经更改，但 AndroidSchedulers 仍然存在于 RxAndroid 中。

3.2 转移 Moved

WidgetObservable 和 ViewObservable 已经被移至（并改进） RxBinding 中。

LifecycleObservable 已移至 RxLifecycle。此外，它们的行为已经被重构，因此请务必检查更改日志。

已经通过 rx-preferences 转移（并改进） ContentObservable.fromSharedPreferencesChanges()。

3.3 例外 Orphaned

• ContentObservable 的其余部分尚未被移动。

• ReactiveDialog 尚未移至新项目。

3.4 移除 Removed

AppObservable 及其绑定方法已被彻底移除。因为它有很多问题：

1. 它试图自动取消订阅，但只有在 Activity 或 Fragment 暂停后序列发射一个数据项时才会这样做。因此，永不结束的序列可能永远不会取消订阅。

2. 它被设计为在暂停后抵制通知，但似乎只是由于 HandlerScheduler 中的一个微妙的逻辑问题而发生。

3. 它自动调用 observeOn(AndroidSchedulers.mainThread())，无论你是否需要。

当移除它时，请确保：

1. 使用手动订阅处理（或 RxLifecycle）正确取消订阅序列。

2. 检查是否需要将 observeOn(AndroidSchedulers.mainThread()) 添加到序列中。

四、RxJava 2 对比 RxJava 1

• RxJava 2 将 onCompleted 更改为 onComplete。

• RxJava 2 使用 Function 替代 Func1（Func1 已经被移除）。

• RxJava 2 使用 BiFunction 替代 Func2（Func2 已经被移除）。

• RxJava 2 使用 CompositeDisposable 替代 CompositeSubscription 和 Subscription（CompositeSubscription 和 Subscription 已经被移除）。

• RxJava 2 使用 take 操作符替代 limit 操作符（limit 操作符已经被删除）。