理解 RxJava 中的 Single 和 Completable

原文链接

在大多数 RxJava 示例代码和教程中出现最为频繁的一个类 —— Observable，它是产生响应式编程魔力的关键。它的用法很简单，只需要跟踪 3 个事件 —— onNext，onError和onCompleted就可以应用上百个操作符来实现自己的表达式。那么为什么你还需要了解其他东西？

但是你仔细思考下，你真的需要每次都知道这 3 个事件吗？实际上，在大多数情况下并不需要。ReactiveX 文档中讲述的基本都是关于连续的事件流，因此我们经常忘记通常我们关心的只是监听单一事件或者只监听 completed or failed 事件。

在这种情况下我们应该考虑用 RxJava 的两个绝妙的设计 —— Single<T> 和 Completable，在分析两者之前，让我们先看看他们应用场景的示例。

本文中所有代码都是基于 RxJava 2.x ，不是 1.x 版本。如果你还没升级 RxJava 到最新的 2.x 版本， 强烈建议你马上升级。

Single

在 Android 中使用 RxJava 最常见的场景就是网络请求，你可能使用 Retrofit 作为项目的 Http client。假设你有一个 GET HTTP 请求返回一些数据，同时使用 RxJavaAdapter 你大概会这么写：

public interface APIClient {    @GET("my/api/path")    Observable<MyData> getMyData();
}

上面的代码没什么问题，当调用它时：

apiClient.getMyData()
    .subscribe(new Consumer<MyData myData>() {        @Override
        public void accept(MyData myData) throws Exception {            // handle data fetched successfully
        }
    }, new Consumer<Throwable>() {        @Override
        public void accept(Throwable throwable) throws Exception{            // handle error event
        }
    }, new Action() {        @Override
        public void run() throws Exception {            // handle on complete event
        }
    });

仔细思考下，其实这个网络请求并不是一个连续事件流，你只会发起一次 Get 请求返回数据并且只收到一个事件。我们都知道这种情况下 onComplete 会紧跟着 onNext 被调用，那为什么不把它们合二为一呢？

在上面这种情况下为了更清楚的体现请求的意图，应该用Single<MyData>替换 Observable。从官方文档中对 Single 的说明可以发现为什么它是最恰当的选择：A Single is something like an Observable, but instead of emitting a series of values — anywhere from none at all to an infinite number — it always either emits one value or an error notification。所以修改后 API client 是这样的：

public interface APIClient {    @GET("my/api/path")    Single<MyData> getMyData();
}

同时请求的调用也可以简化：

apiClient.getMyData()
    .subscribe(new Consumer<MyData>() {        @Override
        public void accept(MyData myData) throws Exception {            // handle data fetched successfully and API call completed
        }
    }, new Consumer<Throwable>() {        @Override
        public void accept(Throwable throwable) throws Exception{            // handle error event
        }
    });

最值得高兴的是 Single 基本上实现了 Observable 所有的操作符 —— map、flatMap、filter、zip等，如果你发现需要用到一个 Observable 的操作符而 Single 并不支持，你可以用toObservable操作符把Single<T>转换为Observable<T>。

apiClient.getMyData()
    .toObservable()    // This is an Observable<MyData> now

如果你有 Observable 表现地像 Single 一样，也可以通过singleOrError操作符转换为 Single。

Completable

继续讨论 Retrofit 的例子，再看看另外一种常用场景 —— 通过 PUT 请求更新数据。我们修改了 MyData 类型对象的一些属性，把它发送到服务器更新服务器数据库。大部分服务器 API 设计都是成功后返回更新后的对象，所以你的 API client 的实现是：

public interface APIClient {    @PUT("my/api/updatepath")    Observable<MyData> updateMyData(@Body MyData data);
}

同样的，跟之前的例子类似，应该这样调用：

apiClient.updateMyData(myUpdatedData)
    .subscribe(new Consumer<MyData myData>() {        @Override
        public void accept(MyData myData) throws Exception {            // handle data fetched successfully and API call completed
        }
    }, new Consumer<Throwable>() {        @Override
        public void accept(Throwable throwable) throws Exception{            // handle error event
        }
    }, new Action() {        @Override
        public void run() throws Exception {            // handle completion - what we actually care about
        }
    });

你可能会说这里我们可以同样用 Single 来简化代码，是的没错。在这种情况下我们仍然需要 MyData 结果，确定？服务器返回给我们更新后的数据是良好的设计，当时实际上仅仅是返回给我们之前发送给它的对象。我们真正需要的只是更新成功了，这意味着，我只关心 onComplete 事件。

这也是引入Completable的原因，官方文档对它的描述是：Represents a computation without any value but only indication for completion or exception。使用 Completable 时我们忽略 onNext 事件，只处理 onComplete 和 onError 事件，API client 改写为：

public interface APIClient {    @PUT("my/api/updatepath")    Completable updateMyData(@Body MyData data);
}

调用为：

apiClient.updateMyData(myUpdatedData)
    .subscribe(new Action() {        @Override
        public void run() throws Exception {            // handle completion
        }
    }, new Consumer<Throwable>() {        @Override
        public void accept(Throwable throwable) throws Exception{            // handle error
        }
    });

Completable 本质上来说和 Observable 与 Single 不一样，因为它不发射数据。因此 Completable 的操作符也有所区别，最常用的是andThen。在这个操作符中你可以传任何Observable、Single、Flowable、Maybe或者其他Completable，它们会在原来的 Completable 结束后执行。例如。你想执行一些其他操作（Single）：

apiClient.updateMyData(myUpdatedData)
    .andThen(performOtherOperation()) // a Single<OtherResult>
    .subscribe(new Consumer<OtherResult>() {        @Override
        public void accept(OtherResult result) throws Exception {            // handle otherResult
        }
    }, new Consumer<Throwable>() {        @Override
        public void accept(Throwable throwable) throws Exception{            // handle error
        }
    });

跟 Single 不同的是 RxJava 不允许直接把 Observable 转换为 Completable，因为没办法知道一个 Observable 什么时候 complete。但是你可以把 Single 转换为 Completable，因为 Single 保证 onComplete 会被调用，这个操作符是toCompletable。

希望通过这篇简短的对 Single 和 Completable 的介绍能让你理解这两个概念从而写出更简洁的代码。