RxJava 与 Retrofit 结合网络请求，你值得拥有

此文旨在讲解RxJava+Retrofit联网请求的结合应用，我将尽我所能的详细讲解。文章的末尾将源码奉上 代码可实现如下功能： 1.网络请求带缓存 2.可取消网络请求 3.加载时显示圆形进度条,加载完毕后移除 4.简单的封装

纳尼!~你对RxJava 和 Retrofit 不甚了解,那么......

RxJava RxJava学习是一个曲折漫长的过程，但一旦掌握，妙用无穷。 1.给 Android 开发者的 RxJava 详解 2.关于RxJava最友好的文章 3.关于RxJava最友好的文章（进阶） Retrofit 1.Retrofit用法详解 2.Retrofit 2.0：有史以来最大的改进 3.Retrofit 2.0 官方文档 4. Retrofit2 完全解析 探索与okhttp之间的关系

一. 只用Retrofit

网络接口的使用为查询天气的接口 http://wthrcdn.etouch.cn/weather_mini?city=北京

返回的Json格式我就不详细介绍了，需要的转换下格式自行看一下 关于返回的结果我们封装成实体类 WeatherEntity

1.1添加所诉依赖build.gradle

compile 'com.squareup.retrofit2:retrofit:2.1.0' compile 'com.squareup.retrofit2:converter-gson:2.1.0' <!-- 别忘记访问网络权限 --> <uses-permission android:name="android.permission.INTERNET" />

1.2 Get请求

public interface GetWeatherService { @GET("weather_mini") Call<WeatherEntity> getMessage(@Query("city") String city); }

在MianActivity中 我们写在initOnlyRetrofit()方法中，如下：

/** * 只有Retrofit的联网请求 */ private void initOnlyRetrofit() { Retrofit retrofit = new Retrofit.Builder() .baseUrl("http://wthrcdn.etouch.cn/")//基础URL 建议以 / 结尾 .addConverterFactory(GsonConverterFactory.create())//设置 Json 转换器 .build(); GetWeatherService weatherService = retrofit.create(GetWeatherService.class); Call<WeatherEntity> call = weatherService.getMessage("北京"); call.enqueue(new Callback<WeatherEntity>() { @Override public void onResponse(Call<WeatherEntity> call, Response<WeatherEntity> response) { Log.e(TAG, "response == " + response.body().getData().getGanmao()); } @Override public void onFailure(Call<WeatherEntity> call, Throwable t) { Log.e(TAG, "Throwable : " + t); } }); }

运行一下我们可以看到打印结果如下:

二.RxJava 与 Retrofit

2.1 同样先添加依赖(新增的)

compile 'com.squareup.retrofit2:adapter-rxjava:2.1.0' compile 'io.reactivex:rxandroid:1.2.1' compile 'io.reactivex:rxjava:1.2.1'

2.2 这时将Retrofit与RxJava 链接起来只需这样修改. 在Retrofit Builder链表中如下调用addCallAdapterFactory

Retrofit retrofit = new Retrofit.Builder() .baseUrl("http://wthrcdn.etouch.cn/")//基础URL 建议以 / 结尾 .addConverterFactory(GsonConverterFactory.create())//设置 Json 转换器 .addCallAdapterFactory(RxJavaCallAdapterFactory.create())//RxJava 适配器 .build();

2.3 这样一来我们定义的service返回值就不在是一个Call了，而是一个Observable ,那我们需重新定义接口

public interface GetWeatherService { @GET("weather_mini") Observable<WeatherEntity> getRxMessage(@Query("city") String city); }

2.4 重新定以后你的Service接口现在可以作为Observable返回了.我们需要用RxJava方式实现网络请求,可以完全像RxJava那样使用它.

GetWeatherService weatherService = retrofit.create(GetWeatherService.class); weatherService.getRxMessage("北京") .subscribeOn(Schedulers.io())//IO线程加载数据 .observeOn(AndroidSchedulers.mainThread())//主线程显示数据 .subscribe(new Subscriber<WeatherEntity>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(WeatherEntity weatherEntity) { Log.e(TAG,"RxJava + Retrofit= " + weatherEntity.getData().getGanmao()); } });

subscribeOn(): 指定 subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。 observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。

2.5 这里实现基本的网络请求了,我们来看下打印日志.

嗯,联网实现了,我看可以考虑收工啦~~~ 纳尼? 说好的缓存呢?dialog呢?还有取消请求我该怎么做? 嗯,嗯,别捉急,我开个玩笑,马上带来重头戏.

三.缓存的实现

在Retrofit 2.0中，OkHttp 是必须的，并且自动设置为了依赖。也就说我们不用在build.gradle添加依赖了。下面的代码是从Retrofit 2.0的pom文件中抓取的。

<dependencies> <dependency> <groupId>com.squareup.okhttp</groupId> <artifactId>okhttp</artifactId> </dependency> ... </dependencies>

为了让OkHttp 的Call模式成为可能，在Retrofit 2.0中OkHttp 自动被用作HTTP 接口。此时我们要在OkHttpClient里设置缓存,添加拦截器.那OkHttpClient在Retrofit里怎么设置呢.其实只需添加一句,在retrofit链式调用中调用client(OkHttpClient client)方法即可，来看下代码：

OkHttpClient mClient = new OkHttpClient.Builder() .addInterceptor(mInterceptor)//应用程序拦截器 .addNetworkInterceptor(mNetInterceptor)//网络拦截器 .cache(mCache)//添加缓存 .build(); Retrofit retrofit = new Retrofit.Builder() .baseUrl(url) .client(mClient)//添加OK .addConverterFactory(GsonConverterFactory.create()) .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) .build();

OkHttpClient我们分别添加了3方法. .cache(mCache)用于添加缓存; .addInterceptor(mInterceptor)应用程序拦截器; .addNetworkInterceptor(mNetInterceptor)网络拦截器;

3.1 OkHttp本身是有缓存这个东西的，只是如果你不去设置，是不起作用的 我们需要设置.cache()去添加缓存

File mFile = new File(context.getCacheDir() + "http");//储存目录 long maxSize = 10 * 1024 * 1024; // 10 MB 最大缓存数 Cache mCache = new Cache(mFile, maxSize);

3.2 这里设置了缓存和目录还是不够的,我们还需要设置拦截器Interceptors

先来看看Interceptor本身的文档解释：观察，修改以及可能短路的请求输出和响应请求的回来。通常情况下拦截器用来添加，移除或者转换请求或者回应的头部信息。 拦截器接口中有intercept(Chain chain)方法，同时返回Response。 chain.proceed(request)是拦截器的关键部分。这个看似简单的方法是所有的HTTP工作发生的地方，产生满足要求的反应。 拦截器可以链接。假设你有一个压缩的拦截和校验拦截器：你需要决定数据是否被压缩，或者校验或校验然后压缩。okhttp使用列表来跟踪和拦截，拦截器会按顺序调用。

3.3 拦截器可以注册为应用程序拦截或网络拦截

每种拦截器chain有相对的优势。

Application interceptors（应用拦截器）

1.不必担心中间的responses，例如重定向和重连。2.总是调用一次，即使是从缓存HTTP响应。3.观察应用程序的原始意图。不关心OkHttp的注入headers，例如If-None-Match4.允许短路和不执行Chain.proceed().5.允许重连，多次调用proceed()。

.

Network Interceptors （网络拦截器）

1.能够操作中间反应，例如重定向和重连。2.不能被缓存响应，例如短路网络调用。3.观察数据，正如它将在网络上传输。4.有权使用携带request的Connection

具体2种选择器用哪种大家酌情考虑即可。说了这么多,拦截Interceptor到底该怎么写?下面个给出拦截器的代码片段。

private static final int NET_MAX = 30; //30秒 有网超时时间 private static final int NO_NET_MAX = 60 * 60 * 24 * 7; //7天 无网超时时间 //网络拦截器 Interceptor mNetInterceptor = new Interceptor() { @Override public Response intercept(Chain chain) throws IOException { Log.e("TAG", "拦截 应用 缓存"); Request request = chain.request(); if (!NetWorkUtils.networkIsAvailable(context)) {//判断网络状态 无网络时 request = request.newBuilder() //Pragma:no-cache。在HTTP/1.1协议中，它的含义和Cache-Control:no-cache相同。为了确保缓存生效 .removeHeader("Pragma") .header("Cache-Control", "private, only-if-cached, max-stale=" + NO_NET_MAX) .build(); } else { request = request.newBuilder() //Pragma:no-cache。在HTTP/1.1协议中，它的含义和Cache-Control:no-cache相同。为了确保缓存生效 .removeHeader("Pragma") .header("Cache-Control", "private, max-age=" + NET_MAX)//添加缓存请求头 .build(); } return chain.proceed(request); } };

这里NetWorkUtils.networkIsAvailable(context)是判断网络状态的工具类,大家可以去源码里找,这里就不在给出.

3.4 Okhttp的缓存由返回的header 来决定，用”Cache-control”来控制，常见的取值有private、no-cache、max-age、must-revalidate等，默认为private。

public 所有内容都将被缓存private 内容只缓存到私有缓存中no-cache 所有内容都不会被缓存no-store 所有内容都不会被缓存到缓存或 Internet 临时文件中must-revalidation/proxy-revalidation 如果缓存的内容失效，请求必须发送到服务器/代理以进行重新验证max-age=xxx (xxx is numeric) 缓存的内容将在 xxx 秒后失效, 这个选项只在HTTP 1.1可用, 并如果和Last-Modified一起使用时, 优先级较高 在某些情况下，如用户单击“刷新”按钮，就可能有必要跳过缓存，并直接从服务器获取数据。要强制刷新，添加无缓存指令："Cache-Control"： "no-cache"。如果缓存只是用来和服务器做验证，可是设置更有效的"Cache-Control"："max-age=0"。有时你会想显示可以立即显示的资源。这是可以使用的，这样你的应用程序可以在等待最新的数据下载的时候显示一些东西， 重定向request到本地缓存资源，添加"Cache-Control"："only-if-cached"。有时候过期的response比没有response更好，设置最长过期时间来允许过期的response响应：int maxStale = 30; // 30秒 "Cache-Control"："max-stale=" + maxStale。

缓存的实现到这里就讲完了,可能代码片段让你看的头晕目眩,那给大家一个简单封装了的完整代码

package com.aaron.rxjava_retrofit.net; import android.content.Context; import android.util.Log; import com.aaron.rxjava_retrofit.utils.NetWorkUtils; import java.io.File; import java.io.IOException; import okhttp3.Cache; import okhttp3.Interceptor; import okhttp3.OkHttpClient; import okhttp3.Request; import okhttp3.Response; import retrofit2.Retrofit; import retrofit2.adapter.rxjava.RxJavaCallAdapterFactory; import retrofit2.converter.gson.GsonConverterFactory; /** * 作者：哇牛Aaron * 作者简书文章地址: http://www.jianshu.com/users/07a8b5386866/latest_articles * 时间: 2016/11/24 * 功能描述: */ public class HttpUtils { private static final String URL_WEATHER = "http://wthrcdn.etouch.cn/"; private static final int NET_MAX = 30; //30秒 有网超时时间 private static final int NO_NET_MAX = 60 * 60 * 24 * 7; //7天 无网超时时间 public static Retrofit getRetrofit(String url, final Context context) { //应用程序拦截器 Interceptor mInterceptor = new Interceptor() { @Override public Response intercept(Chain chain) throws IOException { Log.e("TAG", "拦截 网络 缓存"); Request request = chain.request(); if (!NetWorkUtils.networkIsAvailable(context)) {//判断网络状态 无网络时 Log.e("TAG", "无网~~ 缓存"); request = request.newBuilder() //Pragma:no-cache。在HTTP/1.1协议中，它的含义和Cache-Control:no-cache相同。为了确保缓存生效 .removeHeader("Pragma") .header("Cache-Control", "private, only-if-cached, max-stale=" + NO_NET_MAX) .build(); } else {//有网状态 Log.e("TAG", "有网~~ 缓存"); request = request.newBuilder() //Pragma:no-cache。在HTTP/1.1协议中，它的含义和Cache-Control:no-cache相同。为了确保缓存生效 .removeHeader("Pragma") .header("Cache-Control", "private, max-age=" + NET_MAX)//添加缓存请求头 .build(); } return chain.proceed(request); } }; //网络拦截器 Interceptor mNetInterceptor = new Interceptor() { @Override public Response intercept(Chain chain) throws IOException { Log.e("TAG", "拦截 应用 缓存"); Request request = chain.request(); if (!NetWorkUtils.networkIsAvailable(context)) {//判断网络状态 无网络时 request = request.newBuilder() //Pragma:no-cache。在HTTP/1.1协议中，它的含义和Cache-Control:no-cache相同。为了确保缓存生效 .removeHeader("Pragma") .header("Cache-Control", "private, only-if-cached, max-stale=" + NO_NET_MAX) .build(); } else { request = request.newBuilder() //Pragma:no-cache。在HTTP/1.1协议中，它的含义和Cache-Control:no-cache相同。为了确保缓存生效 .removeHeader("Pragma") .header("Cache-Control", "private, max-age=" + NET_MAX)//添加缓存请求头 .build(); } return chain.proceed(request); } }; File mFile = new File(context.getCacheDir() + "http");//储存目录 long maxSize = 10 * 1024 * 1024; // 10 MB 最大缓存数 Cache mCache = new Cache(mFile, maxSize); OkHttpClient mClient = new OkHttpClient.Builder() .addInterceptor(mInterceptor)//应用程序拦截器 .addNetworkInterceptor(mNetInterceptor)//网络拦截器 .cache(mCache)//添加缓存 .build(); Retrofit retrofit = new Retrofit.Builder() .baseUrl(url) .client(mClient)//添加OK .addConverterFactory(GsonConverterFactory.create()) .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) .build(); return retrofit; } public static GetWeatherService createWeatherService(Context context) { return getRetrofit(URL_WEATHER, context).create(GetWeatherService.class); } }

在MainActivity中调用

HttpUtils.createWeatherService(MainActivity.this).getRxMessage("北京") .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber<WeatherEntity>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(WeatherEntity weatherEntity) { Log.e(TAG, "Cache RxJava + Retrofit= " + weatherEntity.getData().getGanmao()); } });

此时,缓存已经完成了.~~~嗯,写的我有点累了. 学习关于OkHttp Interceptor的知识，请到OkHttp Interceptors。

四.如何取消一个Http请求

当Activity结束后我们需要结束网络请求,为用户节省流量,或者其它需求等 Retrofit中你只需调用call.cancel()。就可以取消网络请求 可是前面已经说了 Call 此时 已经变成了 Observable.那么如何取消呢?我们能做的似乎只有解除对Observable对象的订阅，其他的什么也做不了。 好在Retrofit已经帮我们考虑到了这一点。 答案在RxJavaCallAdapterFactory这个类的源码中可以找到。

static final class CallOnSubscribe<T> implements Observable.OnSubscribe<Response<T>> { private final Call<T> originalCall; CallOnSubscribe(Call<T> originalCall) { this.originalCall = originalCall; } @Override public void call(final Subscriber<? super Response<T>> subscriber) { // Since Call is a one-shot type, clone it for each new subscriber. final Call<T> call = originalCall.clone(); // Attempt to cancel the call if it is still in-flight on unsubscription. subscriber.add(Subscriptions.create(new Action0() { @Override public void call() { call.cancel(); } })); try { Response<T> response = call.execute(); if (!subscriber.isUnsubscribed()) { subscriber.onNext(response); } } catch (Throwable t) { Exceptions.throwIfFatal(t); if (!subscriber.isUnsubscribed()) { subscriber.onError(t); } return; } if (!subscriber.isUnsubscribed()) { subscriber.onCompleted(); } } }

我们看到call方法中，给subscriber添加了一个Subscription对象，Subscription对象很简单，主要就是取消订阅用的，如果你查看Subscriptions.create的源码，发现是这样的。

public static Subscription create(final Action0 unsubscribe) { return BooleanSubscription.create(unsubscribe); }

利用了一个BooleanSubscription类来创建一个Subscription，如果你点进去看BooleanSubscription.create方法一切就清晰了，当接触绑定的时候，subscriber会调用Subscription的unsubscribe方法，然后触发创建Subscription时候的传递进来的Action0的call方法。RxJavaCallAdapterFactory帮我们给subscriber添加的是call.cancel()。

Subscription subscription = HttpUtils.createWeatherService(MainActivity.this).getRxMessage("北京") .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber<WeatherEntity>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(WeatherEntity weatherEntity) { Log.e(TAG, "Cache RxJava + Retrofit= " + weatherEntity.getData().getGanmao()); } }); /*if (subscription != null && !subscription.isUnsubscribed()) {//isUnsubscribed 是否取消订阅 subscription.unsubscribe();//取消网络请求 }*/

五.显示圆形进度条

要添加dialog 我们不得不提RxJava中的两个观察者Observer 和 Subscriber 你只想使用基本功能，这两个观察者基本是一致的它们的区别对于使用者来说主要有两点：

1.onStart(): 这是 Subscriber 增加的方法。它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。这是一个可选方法，默认情况下它的实现为空。需要注意的是，如果对准备工作的线程有要求（例如弹出一个显示进度的对话框，这必须在主线程执行）， onStart() 就不适用了，因为它总是在 subscribe 所发生的线程被调用，而不能指定线程。要在指定的线程来做准备工作，可以使用 doOnSubscribe() 方法.

2.unsubscribe(): 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。在这个方法被调用后，Subscriber 将不再接收事件。一般在这个方法调用前，可以使用 isUnsubscribed() 先判断一下状态。 unsubscribe() 这个方法很重要，因为在 subscribe() 之后， Observable 会持有 Subscriber 的引用，这个引用如果不能及时被释放，将有内存泄露的风险。所以最好保持一个原则：要在不再使用的时候尽快在合适的地方（例如 onPause() onStop() 等方法中）调用 unsubscribe() 来解除引用关系，以避免内存泄露的发生。

final ProgressDialog mDialog = new ProgressDialog(this); Retrofit retrofit = new Retrofit.Builder() .baseUrl("http://wthrcdn.etouch.cn/") .addConverterFactory(GsonConverterFactory.create()) .addCallAdapterFactory(RxJavaCallAdapterFactory.create())//RxJava 适配器 .build(); GetWeatherService weatherService = retrofit.create(GetWeatherService.class); weatherService.getRxMessage("北京") .subscribeOn(Schedulers.io())//指定网络请求在IO线程 .doOnSubscribe(new Action0() { @Override public void call() { mDialog.show(); } }) .subscribeOn(AndroidSchedulers.mainThread())//显示Dialog在主线程中 .observeOn(AndroidSchedulers.mainThread())//显示数据在主线程 .subscribe(new Subscriber<WeatherEntity>() { @Override public void onCompleted() { if (mDialog != null){ mDialog.cancel(); } } @Override public void onError(Throwable e) { if (mDialog != null){ mDialog.cancel(); } } @Override public void onNext(WeatherEntity weatherEntity) { Log.e(TAG, "RxJava + Retrofit= " + weatherEntity.getData().getGanmao()); } });

如上，在 doOnSubscribe()的后面跟一个 subscribeOn()，就能指定准备工作的线程了。

好，终于讲完了，喜欢的小伙伴希望不要吝惜点个喜欢吧~ 要转载的朋友，请注明出处，珍惜下我的劳动成果，谢谢。 有什么问题，欢迎大家留言 讨论。

源码点这里

补充:关于封装可以利用单例设计模式进行封装，这里给出一个简单的案例，源码就不再修改了。

/** * 作者：哇牛Aaron * 作者简书文章地址: http://www.jianshu.com/users/07a8b5386866/latest_articles * 时间: 2016/11/28 * 功能描述: */ public class HttpUtils { public static final String BASE_URL = "http://wthrcdn.etouch.cn/"; private Retrofit retrofit; private static final int NO_NET_MAX = 60 * 60 * 24 * 7; //7天 无网超时时间 private static final int NET_MAX = 30; //30秒 有网超时时间 private HttpUtils() { Interceptor mInterceptor = new Interceptor() { @Override public Response intercept(Chain chain) throws IOException { Request request = chain.request(); if (!NetWorkUtils.networkIsAvailable(AaronApplication.getObjectContext())) { request = request.newBuilder() .removeHeader("Pragma") .header("Cache-Control", "private, only-if-cached, max-stale=" + NO_NET_MAX) .build(); } else { request = request.newBuilder() //Pragma:no-cache。在HTTP/1.1协议中，它的含义和Cache-Control:no-cache相同。为了确保缓存生效 .removeHeader("Pragma") .header("Cache-Control", "private, max-age=" + NET_MAX)//添加缓存请求头 .build(); } return chain.proceed(request); } }; OkHttpClient mClient = new OkHttpClient.Builder() .addNetworkInterceptor(mInterceptor) .cache(new Cache(new File(AaronApplication.getObjectContext().getCacheDir() + "http") , 1024 * 1024 * 10)) .build(); retrofit = new Retrofit.Builder() .baseUrl(BASE_URL) .addConverterFactory(GsonConverterFactory.create()) .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) .client(mClient) .build(); } //单例设计模式 private static class singRetrofit { private static final HttpUtils instance = new HttpUtils(); } public static HttpUtils getInstance() { return singRetrofit.instance; } public static GetWeatherService createWeather() { //GetWeatherService getWeatherService = getInstance().retrofit.create(GetWeatherService.class); return getInstance().retrofit.create(GetWeatherService.class); } }