RxJava2+Retrofit2实现网络请求和解析封装

半年多前写过一篇用Retrofit2请求网络和解析的博客，Retrofit2的简单应用与封装，不过当时其实还是遗留了不少细节问题没有处理，比如如果有公共参数放Header里面怎么处理，请求过程中想显示进度框怎么处理，退出时要退出网络请求怎么处理等等，这两天看了下RxJava，主要是看了这篇文章RxJava详解，虽然并不是说的RxJava2，但原理差不多，讲得非常清楚，觉得有点意思，就想用RxJava来重新改进一下之前的这个封装

首先声明，这篇博客侧重是讲如何使用Retrofit2和RxJava2封装网络请求和解析，并不会重点介绍Retrofit2和RxJava2的知识，如对基础知识不了解，请先查看上面的链接学习，另外，Retrofit和Retrofit2，RxJava和RxJava2还是有一些不同的地方，本篇博客是基于Retrofit2和RxJava2

1. 添加依赖

项目中用到Retrofit2，RxJava2，还用到Jackson，我们首先要在build.gradle中添加相关的依赖

compile 'com.squareup.retrofit2:retrofit:2.0.0'
compile 'com.squareup.retrofit2:converter-jackson:2.0.0'
compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0-RC3'
compile 'io.reactivex.rxjava2:rxjava:2.0.0-RC3'
compile 'io.reactivex.rxjava2:rxandroid:2.0.0-RC1'

2. 定义Retrofit访问的接口

public interface RetrofitService {
@FormUrlEncoded
@POST("getUser")
Observable<BaseEntity<User>> getUser(@FieldMap Map<String, String> map);
}

指定使用POST方式，访问服务器方法名为getUser，参数@FieldMap可以用来传递多个参数，当然如果参数较少，也可以直接使用@Field

这里唯一要注意的是方法的返回值，首先如果我们记得的话，在直接使用Retrofit，不使用RxJava时，我们的返回值都是Call，而如果我们想跟RxJava结合，这里的返回值对象就应该为Observable，Observable的泛型这里为BaseEntity<User>，这个在上面的博客中也讲过，这里再说明下

public class BaseEntity<E> implements Serializable {
private int code;
private String message;
private E data;

public int getCode() {
return code;
}

public void setCode(int code) {
this.code = code;
}

public String getMessage() {
return message;
}

public void setMessage(String message) {
this.message = message;
}

public E getData() {
return data;
}

public void setData(E data) {
this.data = data;
}
}

BaseEntity是服务器返回值的通用格式，它由三个部分组成，code表示成功还是失败，0为成功，非0为失败，message是提示内容，而主要的内容都封装在data里面，data为泛型，可以指定为任何内容，我们这个例子中就是一个User对象

public class User implements Serializable{
private String name;
private int age;

public String getName() {
return name;
}

public void setName(String name) {
this.name = name;
}

public int getAge() {
return age;
}

public void setAge(int age) {
this.age = age;
}
}

就是一个简单的实体类，这个根据实际情况大家可以随意定义

3. 对Retrofit2的基本设置

网络请求部分主要使用Retrofit2来实现，我们先看下基础的设置，直接上代码

public class RetroFactory {
private static String baseUrl = "http://192.168.0.107:8082/MyWeb/";

private RetroFactory() {
}

private static OkHttpClient httpClient = new OkHttpClient.Builder()
.addInterceptor(new Interceptor() {
@Override
public Response intercept(Chain chain) throws IOException {
Request.Builder builder = chain.request().newBuilder();
builder.addHeader("token", "abc");
return chain.proceed(builder.build());
}
}).connectTimeout(30, TimeUnit.SECONDS)
.readTimeout(30, TimeUnit.SECONDS)
.build();

private static RetrofitService retrofitService = new Retrofit.Builder()
.baseUrl(baseUrl)
.addConverterFactory(JacksonConverterFactory.create())
.addCallAdapterFactory(RxJava2CallAdapterFactory.create())
.client(httpClient)
.build()
.create(RetrofitService.class);

public static RetrofitService getInstance() {
return retrofitService;
}
}

这里baseUrl就是我们服务器的项目运行的地址，我是在我本机启动了一个Tomcat，192.168.0.107是我本机的IP，8082是Tomcat的端口号，MyWeb是我服务器项目的名称。

接着看，Retrofit2内部是使用OkHttp3的，我们对网络访问的一些设置都可以通过OkHttp来进行，这里首先调用了一个addInterceptor方法，用来增加一个拦截器，而拦截器的内容是给每个网络请求增加了一个通用的Header字段，名为token，值为abc，这在实际项目中是非常常见的，每个请求都通过token来识别是否是有效的请求，防止恶意请求，当然，实际token应该是动态生成的，我这里只是演示如何在Header中添加通用内容，就直接赋值为abc了。

下面接着设置了connectTimeout和readTimeout的超时时间为30秒，实际网络访问中，存在各种异常情况，掉线，不通，时断时续等等，那设置超时时间就非常必要了，肯定不能无限等待，其中connectTimeout是连接超时时间，在指定时间内还没有连接到服务器就会报SocketTimeout异常，而readTimeout是读取超时时间，是连接后在指定时间还没有获取到数据就超时。

设置为OkHttp，我们再来看Retrofit本身的设置，这里baseUrl就是我们上面讲的公用的链接，addConverterFactory是指定使用Jackson来解析Json数据，当然，你也可以使用Gson或者FastJson，不过数据量大的时候，Gson的效率不高，推荐使用Jackson和FastJson，而addCallAdapterFactory，通过这个转换，才能将服务器的返回值从Retrofit默认的Call变为Observable

最后，提供一个方法返回RetrofitService对象，这整个其实就是一个懒汉式单例模式

4. 定义网络请求Activity的公共基类

public class NetworkBaseActivity extends AppCompatActivity {
public ProgressDialog pd;
public Function<Observable, ObservableSource> composeFunction;
private final long RETRY_TIMES = 1;
private boolean showLoading = true;

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);

init();
}

private void init() {
pd = new ProgressDialog(this);

composeFunction = new Function<Observable, ObservableSource>() {
@Override
public ObservableSource apply(Observable observable) throws Exception {
return observable.retry(RETRY_TIMES)
.subscribeOn(Schedulers.io())
.doOnSubscribe(new Consumer<Disposable>() {
@Override
public void accept(Disposable disposable) throws Exception {
if (NetworkUtil.isNetworkAvailable(NetworkBaseActivity.this)) {
if (showLoading) {
if(pd != null && !pd.isShowing()){
pd.show();
}
}
} else {
Toast.makeText(NetworkBaseActivity.this, "网络连接异常，请检查网络", Toast.LENGTH_LONG).show();
}
}
})
.observeOn(AndroidSchedulers.mainThread());
}
};
}

public void setLoadingFlag(boolean show) {
showLoading = show;
}

@Override
protected void onStop() {
super.onStop();

if (pd != null && pd.isShowing()) {
pd.dismiss();
}
}
}

这里的ProgressDialog是一个简单的进度框，因为有的网络请求可能耗时较长，如果界面不提供任何互动的话，用户会误以为程序卡死，用户体验较差，提供一个进度框就可以解决这个问题。Function是对Observable的一些基础设置，等会再具体看，RETRY_TIMES，顾名思义，就是重试的次数，网络环境较差或出现其它异常情况的时候，我们希望程序可以自动进行重试，最后一个showLoading用来设置是否显示进度框，原则上一般的请求都要显示，所以默认值是true，但也有例外，举个例子，进应用的时候需要检测是否有版本更新，这个操作我们肯定是希望在后台进行而用户感知不到的，如果这里出现一个进度框就会莫名其妙，所以我们提供这样一个设置。

下面我们具体看下这个composeFunction，retry方法就是刚才说到的重试次数，不指定默认为0，subscribeOn用来指定网络请求所在的线程，这里用IO线程，doOnSubscribe是在事件发送前进行的操作，所以我们可以做一些初始化的工作，isNetworkAvailable用来检测网络是否是连接的

public class NetworkUtil {
public static boolean isNetworkAvailable(Activity activity) {
Context context = activity.getApplicationContext();
ConnectivityManager connectivityManager = (ConnectivityManager) context.getSystemService(Context.CONNECTIVITY_SERVICE);

if (connectivityManager == null) {
return false;
} else {
NetworkInfo[] networkInfo = connectivityManager.getAllNetworkInfo();

if (networkInfo != null && networkInfo.length > 0) {
for (int i = 0; i < networkInfo.length; i++) {

if (networkInfo[i].getState() == NetworkInfo.State.CONNECTED) {
return true;
}
}
}
}
return false;
}
}

这都是套路，不多说，如果网络连接正常，而且showLoading也为true，我们就显示一个进度框，否则提示用户网络连接异常。

observeOn是用来指定Observer操作的线程，也就是我们得到服务器返回的结果后的线程，因为我们需要操作控件，所以只能在UI线程进行。这里多说一句，doOnSubscribe的线程是什么呢？它既不是在subscribeOn指定的线程，更不是在observeOn指定的线程，而是执行subscribe时所在的线程，subscribe我们现在还没用到，后面会看到，本程序subscribe是在主线程执行，所以doOnSubscribe也就是在主线程了，我们这里显示进度框就是要在主线程进行，所以不用特意去指定，如果subscribe不在主线程，那可以在doOnSubscribe后通过subscribeOn指定它所在的线程。

setLoadingFlag方法就是提供一个设置是否显示进度框的途径。

最后的onStop，保险起见，我们判断一下，进度框是否关闭，如果没关闭要关掉，后面我们还会看到，进度框关闭时我们也会取消订阅，防止已经退出后还在处理请求。

5. 封装Observer

上面我们对Observable进行了封装，那现在我们再来封装Observer

public abstract class BaseObserver<T> implements Observer<BaseEntity<T>> {
private Context mContext;
private ProgressDialog mDialog;
private Disposable mDisposable;
private final int SUCCESS_CODE = 0;

public BaseObserver(Context context, ProgressDialog dialog) {
mContext = context;
mDialog = dialog;

mDialog.setOnCancelListener(new DialogInterface.OnCancelListener() {
@Override
public void onCancel(DialogInterface dialog) {
mDisposable.dispose();
}
});
}

@Override
public void onSubscribe(Disposable d) {
mDisposable = d;
}

@Override
public void onNext(BaseEntity<T> value) {
if (value.getCode() == SUCCESS_CODE) {
T t = value.getData();
onHandleSuccess(t);
} else {
onHandleError(value.getCode(), value.getMessage());
}
}

@Override
public void onError(Throwable e) {
Log.d("gesanri", "error:" + e.toString());

if(mDialog != null && mDialog.isShowing()){
mDialog.dismiss();
}

Toast.makeText(mContext, "网络异常，请稍后再试", Toast.LENGTH_LONG).show();
}

@Override
public void onComplete() {
Log.d("gesanri", "onComplete");

if(mDialog != null && mDialog.isShowing()){
mDialog.dismiss();
}
}

abstract void onHandleSuccess(T t);

void onHandleError(int code, String message) {
Toast.makeText(mContext, message, Toast.LENGTH_LONG).show();
}
}

Observer是一个接口，它提供了4个方法，onSubscribe用来随时取消和Observable的连接，onNext用来处理Observable的返回，也就是网络连接的返回，onComplete在onNext后被调用，表示完成，onError表示发生了错误，onComplete和onError两个方法中只会并且肯定会有一个方法被调用。

onSubscribe提供了一个Disposable参数，我们调用它的dispose方法就可以终止订阅，在dialog的setOnCancelListener中，我们调用它来取消订阅，这样如果用户在请求的过程中觉得等待时间过长，点击返回键关闭进度框或者退出应用时，我们就可以取消订阅而不继续进行处理了。不过这里有一点要注意的是，这个dispose方法并不会影响到服务器端，如果请求已经发送到服务器端，那就算客户端调用了dispose方法，服务器端的代码依然会继续执行，在一些服务端接口涉及插入数据库的操作时，就要特别注意，考虑客户端在dispose方法后，调用服务器端的方法去执行一个类似回滚的操作，否则客户端取消了订阅，服务器端依然会执行完成，这个要根据项目的实际情况来具体对待。

onNext方法，是我们要处理的主要方法，在这之中，我们通过判断返回值中的code，来判断要做的操作，如果code为0，也就是成功，我们就执行onHandleSuccess方法，如果code不为0，也就是失败，我们就执行onHandleError方法，注意，一般情况下，成功是要单独处理，而失败只给用户提示就可以了，所以这里我将onHandleSuccess声明为抽象的，也就是子类必须要实现，而onHandleError不是抽象的，子类可以选择实现或就用默认的实现即可。

最后，不管是进入了onComplete还是onError方法，都要记得关闭进度框

6. 定义调用网络请求的Activity

public class MainActivity extends NetworkBaseActivity {
private TextView name;
private TextView age;
private Observable observable;

@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);

name = (TextView) findViewById(R.id.name);
age = (TextView) findViewById(R.id.age);

getUsers();
}

private void getUsers() {
Map<String, String> map = new HashMap<String, String>();
map.put("id", "123");
map.put("name", "gesanri");

observable = RetroFactory.getInstance().getUser(map);
observable.compose(composeFunction).subscribe(new BaseObserver<User>(MainActivity.this, pd) {
@Override
void onHandleSuccess(User user) {
name.setText("姓名：" + user.getName());
age.setText("年龄：" + user.getAge());
}
});
}
}

这个类继承了我们之前第4步中定义的公共基类，布局文件也很简单

<?xml version="1.0" encoding="utf-8"?>
<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
android:layout_width="match_parent"
android:layout_height="match_parent"
android:orientation="vertical"
>

<TextView
android:id="@+id/name"
android:layout_width="wrap_content"
android:layout_height="wrap_content" />

<TextView
android:id="@+id/age"
android:layout_width="wrap_content"
android:layout_height="wrap_content" />
</LinearLayout>

这里功能很简单，就是假设传了2个参数，id和name，完后接收服务器返回的数据，显示姓名和年龄。可以看到，经过我们前面几步的封装，Activity的实现已经非常干净了，短短几行代码就实现了网络的请求和返回数据的解析

7.服务器端的实现

@WebServlet(name = "getUser", value = "/getUser")
public class GetUserServlet extends HttpServlet {
private static final long serialVersionUID = 1L;

@Override
protected void doPost(HttpServletRequest req, HttpServletResponse resp)
throws ServletException, IOException {

System.out.println("token:" + req.getHeader("token"));

System.out.println("id:" + req.getParameter("id"));
System.out.println("name:" + req.getParameter("name"));

resp.setCharacterEncoding("utf-8");

if (req.getHeader("token").equals("abc")) {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

resp.getWriter()
.write("{\"code\":0, \"message\":\"获取用户成功！\", \"data\":{\"name\":\"张三\", \"age\":23}}");
} else {
resp.getWriter()
.write("{\"code\":1, \"message\":\"获取用户失败！\", \"data\":\"\"}");
}
}
}

最后我们简单看下服务器端的实现，我们接收了三个参数，一个是Header里面的Token， 这个是每个请求都有的，另外两个是这个getUser请求特定的参数，我们判断如果token不为abc，就返回错误，如果是abc，就返回成功，实际项目肯定是要从数据库获取，这里主要是演示客户端，就偷懒直接返回数据了。这里为了在客户端演示进度框，就休眠了5秒。

整个封装过程就是这样，还是比较基础的，当然，不管是Retrofit还是RxJava，能做的工作远远不止这些，我个人了解的也比较少和浅，大家可以自己继续深入学习，根据项目的实际需要来不断优化自己的框架。

源码下载