Android RxJava+Retrofit完美封装

今日科技快讯

前几天，北京地区发生重度雾霾，不少中小学都已经停课，一些公司也临时启用了在家远程办公的方式。据京东统计，在12月16日至12月20日五天时间内，京东一共卖出了1500万只口罩和11万台空气净化器。但具有戏剧性的一幕是，就在昨天，北京的空气突然爆好，AQI指数从400多骤降到20左右，晴空万里，雾霾散尽，看来很多人的口罩可能都白买了。

作者简介

本篇来自 小河马 的投稿，分享了自己是如何进行 RxJava+Retrofit 的封装。本文的技术点自然没话说，另外对于这种模块化的思路，希望能帮助到大家。最后提前祝大家周末愉快以及圣诞快乐！

小河马 的博客地址：

http://www.jianshu.com/users/14354bcb0e09

前言

Retrofit 和 RxJava 已经出来很久了，很多前辈写了很多不错的文章，在此不得不感谢这些前辈无私奉献的开源精神，能让我们站在巨人的肩膀上望得更远。对于 RxJava 不是很了解的同学推荐你们看扔物线大神的这篇文章：

给Android开发者的RxJava详解

http://gank.io/post/560e15be2dca930e00da1083

一遍看不懂就看第二遍。Retrofit的使用可以参考：

Android Retrofit2.0使用

http://wuxiaolong.me/2016/01/15/retrofit

本文内容是基于 Retrofit + RxJava 做的一些巧妙的封装。参考了很多文章加入了一些自己的理解，请多指教。源码地址：

https://github.com/Hemumu/RxSample

先放出 build.gradle：

本文是基于 RxJava1.1.0 和 Retrofit 2.0.0-beta4 来进行的。

初始化Retrofit

新建类Api，此类就是初始化 Retrofit，提供一个静态方法初始化 Retrofit 非常简单.

提供一个静态方法初始化 Retrofit，手动创建了 OkHttpClient 设置了请求的超时时间。并在 OkHttp 的拦截器中增加了请求头。注意这里是为所有的请求添加了请求头，你可以单独的给请求增加请求头，例如：

和 Retrofit 初始化不同的地方就在我们添加了这两句话：

service 的定义也从这样：

变成了：

返回值变成了 Observable，这个 Observable 不就是 RxJava 的可观察者（即被观察者）么。

封装服务器请求以及返回数据

用户在使用任何一个网络框架都只关心请求的返回和错误信息，所以对请求的返回和请求要做一个细致的封装。

我们一般请求的返回都是像下面这样：

如果你们的服务器返回不是这样的格式那你就只有坐下来请他喝茶，跟他好好说（把他头摁进显示器）了。大不了就献出你的菊花吧！

对于这样的数据我们肯定要对 code 做出一些判断，不同的 code 对应不同的错误信息。所以我们新建一个 HttpResult 类，对应上面的数据结构。

这算是所有实体的一个基类，data 可以为任何数据类型。

我们要对所以返回结果进行预处理，新建一个 RxHelper，预处理无非就是对 code 进行判断和解析，不同的错误返回不同的错误信息，这还不简单。Rxjava 的 map 操作符不是轻松解决。

哟，这不是轻松愉快 so seay么！对 code 进行了判断，code 为0 就做对应更新UI或者其他后续操作，不等于0就抛出异常，在 ApiException 中对 code 做处理，根据 message 字段进行提示用户。

然而。。。RxJava 永远比你想象的强大。RxJava 中那么多操作符看到我身体不适，有个操作符 compose。因为我们在每一个请求中都会处理 code 以及一些重用一些操作符，比如用 observeOn 和 subscribeOn 来切换线程。

RxJava提供了一种解决方案：Transformer（转换器），一般情况下就是通过使用操作符Observable.compose()来实现。具体可以参考：

避免打断链式结构：使用.compose( )操作符

http://www.jianshu.com/p/e9e03194199e

新建一个 RxHelper 对结果进行预处理，代码：

Transformer 实际上就是一个 Func1<Observable<T>, Observable<R>>，换言之就是：可以通过它将一种类型的 Observable 转换成另一种类型的 Observable，和调用一系列的内联操作符是一模一样的。

这里我们首先使用 flatMap 操作符把 Obserable<HttpResult<T>>,转换成为 Observable<T> 在内部对 code 进行了预处理。如果成功则把结果 Observable<T> 发射给订阅者。反之则把 code 交给 ApiException 并返回一个异常，ApiException 中我们对 code 进行相应的处理并返回对应的错误信息。

最后调用了频繁使用的 subscribeOn() 和 observeOn() 以及 unsubscribeOn()。

处理ProgressDialog

在 Rxjava 中我们什么时候来显示 Dialog 呢。一开始觉得是放在 Subscriber<T> 的 onStart 中。onStart 可以用作流程开始前的初始化。然而 onStart() 由于在 subscribe() 发生时就被调用了，因此不能指定线程，而是只能执行在 subscribe() 被调用时的线程。所以 onStart 并不能保证永远在主线程运行。

怎么办呢？

千万不要小看了 RxJava，与 onStart() 相对应的有一个方法 doOnSubscribe()，它和 onStart() 同样是在 subscribe() 调用后而且在事件发送前执行，但区别在于它可以指定线程。默认情况下， doOnSubscribe() 执行在 subscribe() 发生的线程；而如果在 doOnSubscribe() 之后有 subscribeOn() 的话，它将执行在离它最近的 subscribeOn() 所指定的线程。

可以看到在 RxHelper 中看到我们调用了两次 subscribeOn，最后一个调用也就是离doOnSubscribe() 最近的一次 subscribeOn 是指定的 AndroidSchedulers.mainThread() 也就是主线程。这样我们就就能保证它永远都在主线运行了。这里不得不感概 RxJava 的强大。

这里我们自定义一个类 ProgressSubscriber 继承 Subscriber<T>：

初始化 ProgressSubscriber 新建了一个我们自己定义的 ProgressDialog 并且传入一个自定义接口 ProgressCancelListener。此接口是在 SimpleLoadDialog 消失 onCancel 的时候回调的。用于终止网络请求。

ProgressSubscriber 其他就很简单了，在 onCompleted() 和 onError() 的时候取消 Dialog。需要的时候调用 showProgressDialog 即可。

处理数据缓存

服务器返回的数据我们肯定要做缓存，所以我们需要一个 RetrofitCache 类来做缓存处理。

几个参数注释上面已经写得很清楚了，不需要过多的解释。这里我们先取了一个 Observable<T> 对象 fromCache,里面的操作很简单，去缓存里面找个 key 对应的缓存，如果有就发射数据。

在 fromNetwork 里面做的操作仅仅是缓存数据这一操作。最后判断如果强制刷新就直接返回 fromNetwork 反之用 Observable.concat() 做一个合并。concat 操作符将多个 Observable 结合成一个 Observable 并发射数据。这里又用了 first()。fromCache 和 fromNetwork 任何一步一旦发射数据后面的操作都不执行。

最后我们新建一个 HttpUtil 用来返回用户关心的数据，缓存，显示Dialog在这里面进行：

Activity生命周期管理

基本的网络请求都是向服务器请求数据，客户端拿到数据后更新UI。但也不排除意外情况，比如请求回数据途中 Activity 已经不在了，这个时候就应该取消网络请求。

要实现上面的功能其实很简单，两部分

• 随时监听 Activity(Fragment) 的生命周期并对外发射出去； 在我们的网络请求中，接收生命周期
• 并进行判断，如果该生命周期是自己绑定的，如 Destory，那么就断开数据向下传递的过程

实现以上功能需要用到 Rxjava 的 Subject 的子类 PublishSubject

在你的 BaseActivity 中添加如下代码：

这样的话，我们把所有生命周期事件都传给了 PublishSubject 了，或者说 PublishSubject 已经接收到了并能够对外发射各种生命周期事件的能力了。

现在我们要让网络请求的时候去监听这个 PublishSubject，在收到相应的生命周期后取消网络请求，这又用到了我们神奇的 compose(),我们需要修改 handleResult 代码如下：

调用的时候增加了两个参数一个是 ActivityLifeCycleEvent 其实就是一些枚举表示 Activity 的生命周期

public enum ActivityLifeCycleEvent {
 CREATE,
 START,
 RESUME,
 PAUSE,
 STOP,
 DESTROY
}

另外一个参数就是我们在 BaseActivity 添加的 PublishSubject，这里用到了 takeUntil(),它的作用是监听我们创建的 compareLifecycleObservable

compareLifecycleObservable 中就是判断了如果当前生命周期和 Activity 一样就发射数据，一旦 compareLifecycleObservable 对外发射了数据，就自动把当前的Observable(也就是网络请求的Observable)停掉。

当然有个库是专门针对这种情况的，叫

RxLifecycle

https://github.com/trello/RxLifecycle

不过要继承他自己的 RxActivity，当然这个库不只是针对网络请求，其他所有的Rxjava都可以。有需要的可以去看看。

最后新建一个 ApiService 存放我们的请求:

使用起来就超级简单了:

具体很多东西都可以在使用的时候具体修改，比如缓存我用的 Hawk。Dialog 是我自己定义的一个 SimpleLoadDialog。源码已经给出请多指教！