RxJava2和Retrofit2封装教程（整洁、简单、实用）

package com.lin.netrequestdemo.data;


import android.util.Log;

import io.reactivex.Observable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

public class RxNet {
 /**
 * 统一处理单个请求
 *
 * @param observable
 * @param callBack
 * @param 
 */
 public static  Disposable request(Observable> observable, final RxNetCallBack callBack) {
 return observable.subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .onErrorReturn(new Function>() {
   @Override
   public BaseResponse apply(Throwable throwable) {
   Log.e("LinNetError", throwable.getMessage());
   callBack.onFailure(ExceptionHandle.handleException(throwable));
   return null;
   }
  })
  .subscribe(new Consumer>() {
   @Override
   public void accept(BaseResponse tBaseResponse) {
   if (tBaseResponse.getCode().equals("200")) {
    callBack.onSuccess(tBaseResponse.getData());

   } else {
    callBack.onFailure(tBaseResponse.getMsg());
   }
   }
  }, new Consumer() {
   @Override
   public void accept(Throwable throwable) {
   Log.e("LinNetError", "单个请求的错误" + throwable.getMessage());
   }
  });
 }

 /**
 * 统一处理单个请求
 * 返回数据没有body
 */
 public static Disposable requestWithoutBody(Observable observable,
      final RxNetCallBack callBack) {
 return observable.subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .onErrorReturn(new Function() {
   @Override
   public BaseResponse apply(Throwable throwable) {
   Log.v("LinNetError", throwable.getMessage());
   callBack.onFailure(ExceptionHandle.handleException(throwable));
   return null;
   }
  })
  .subscribe(new Consumer() {
   @Override
   public void accept(BaseResponse baseResponse) {
   if (baseResponse.getCode().equals("200")) {
    callBack.onSuccess(baseResponse.getMsg());
   } else {
    callBack.onFailure(baseResponse.getMsg());
   }
   }
  }, new Consumer() {
   @Override
   public void accept(Throwable throwable) {
   Log.v("LinNetError", "单个请求的错误:没有body" + throwable.getMessage());
   }
  });

 }
}

package com.lin.netrequestdemo.data;


public interface RxNetCallBack {
 /**
 * 数据请求成功
 *
 * @param data 请求到的数据
 */
 void onSuccess(T data);

 /**
 * 数据请求失败
 */
 void onFailure(String msg);
}

package com.lin.netrequestdemo.data;

import android.net.ParseException;

import com.google.gson.JsonParseException;


import org.apache.http.conn.ConnectTimeoutException;
import org.json.JSONException;

import java.net.ConnectException;

import retrofit2.HttpException;


public class ExceptionHandle {

 private static final int UNAUTHORIZED = 401;
 private static final int FORBIDDEN = 403;
 private static final int NOT_FOUND = 404;
 private static final int REQUEST_TIMEOUT = 408;
 private static final int INTERNAL_SERVER_ERROR = 500;
 private static final int BAD_GATEWAY = 502;
 private static final int SERVICE_UNAVAILABLE = 503;
 private static final int GATEWAY_TIMEOUT = 504;

 public static String handleException(Throwable e) {
 String errorMsg;
 if (e instanceof HttpException) {
  HttpException httpException = (HttpException) e;
  switch (httpException.code()) {
  case UNAUTHORIZED:
  case FORBIDDEN:
  case NOT_FOUND:
  case REQUEST_TIMEOUT:
  case GATEWAY_TIMEOUT:
  case INTERNAL_SERVER_ERROR:
  case BAD_GATEWAY:
  case SERVICE_UNAVAILABLE:
  default:
   errorMsg = "网络错误";
   break;
  }
  return errorMsg + ":" + httpException.code();
 } else if (e instanceof JsonParseException || e instanceof JSONException || e instanceof ParseException) {
  return "解析错误";
 } else if (e instanceof ConnectException) {
  return "连接失败";
 } else if (e instanceof javax.net.ssl.SSLHandshakeException) {
  return "证书验证失败";
 } else if (e instanceof ConnectTimeoutException) {
  return "连接超时";
 } else if (e instanceof java.net.SocketTimeoutException) {
  return "连接超时";
 } else {
  return "未知错误";
 }
 }

}

package com.lin.netrequestdemo.data.api;

import android.util.Log;

import com.lin.netrequestdemo.data.AppConstants;

import java.util.concurrent.TimeUnit;

import okhttp3.OkHttpClient;
import okhttp3.logging.HttpLoggingInterceptor;
import retrofit2.Retrofit;
import retrofit2.adapter.rxjava2.RxJava2CallAdapterFactory;
import retrofit2.converter.gson.GsonConverterFactory;


public class ApiManager {

 private Retrofit client;

 private ApiManager() {
 client = new Retrofit.Builder()
  .baseUrl(AppConstants.Base_Url_Test)
  .client(initClient())
  .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
  .addConverterFactory(GsonConverterFactory.create())
  .build();
 }

 private static volatile MallApi INSTANCE;

 public static MallApi getInstance() {
 if (INSTANCE == null) {
  synchronized (ApiManager.class) {
  if (INSTANCE == null) {
   INSTANCE = new ApiManager().getMallApi();
  }
  }
 }
 return INSTANCE;
 }

 private MallApi getMallApi() {
 return client.create(MallApi.class);
 }

 private static OkHttpClient initClient() {
 OkHttpClient.Builder builder = new OkHttpClient.Builder();
 //声明日志类
 HttpLoggingInterceptor httpLoggingInterceptor = new HttpLoggingInterceptor(new HttpLoggingInterceptor.Logger() {
  @Override
  public void log(String message) {
  Log.v("LinNet", message);
  }
 });
 //设定日志级别
 httpLoggingInterceptor.setLevel(HttpLoggingInterceptor.Level.BODY);
 //延时
 builder.addInterceptor(httpLoggingInterceptor)
  .connectTimeout(10, TimeUnit.SECONDS)
  .readTimeout(10, TimeUnit.SECONDS)
  .writeTimeout(10, TimeUnit.SECONDS);
 return builder.build();
 }

}

showLoading();
 Map map = new ArrayMap<>();
 map.put("action", "pricetrend");
 addCompositeDisposable(RxNet.request(ApiManager.getInstance().getCat(map), new RxNetCallBack>() {
  @Override
  public void onSuccess(List data) {
  hideLoading();
  showToast("获取列表成功" + data.get(0).toString());
  }

  @Override
  public void onFailure(String msg) {
  hideLoading();
  showToast(msg);
  }
 }));