封装了线程的基本管理、网络错误的集中处理，数据返回结果的过滤等，满足基本开发；（后期会加上MVP的封装以及基于MVP）的线程管理等

依赖

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.1.0'
compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'
compile 'com.squareup.okhttp3:logging-interceptor:3.4.1'
compile 'com.squareup.retrofit2:retrofit:2.1.0'
compile 'com.squareup.retrofit2:converter-gson:2.2.0'

创建ApiService类

 public interface ApiService {
     @FormUrlEncoded
     @POST("api/recommend")
     Observable<WrapBean<RecBean>> getRecApi(@FieldMap Map<String, String> map);
   }

实体类

服务器返回数据的WrapBean

public class WrapBean<T> {
    private int code;
    private String msg;
    private T data;

    public int getCode() {
        return code;
    }

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

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }

    public T getData() {
        return data;
    }

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

具体data类：

public class RecBean {
    此处省略...
}

RetrofitManager

public class RetrofitManager {

    /**
     * OkHttpClient 的基本设置
     * @return
     */
    private static OkHttpClient getOkHttpClient() {
        OkHttpClient.Builder builder = new OkHttpClient.Builder();
        if (BuildConfig.DEBUG) {
            builder.addInterceptor(getHttpLoggingInterceptor());
        }
        builder.connectTimeout(10, TimeUnit.SECONDS);
        builder.readTimeout(15, TimeUnit.SECONDS);
        builder.writeTimeout(15, TimeUnit.SECONDS);
        builder.retryOnConnectionFailure(true);
        return builder.build();
    }

    /**
     * 日志控制
     *
     * @return
     */
    private static HttpLoggingInterceptor getHttpLoggingInterceptor() {
        HttpLoggingInterceptor loggingInterceptor = new HttpLoggingInterceptor();
        if (BuildConfig.DEBUG) {
            loggingInterceptor.setLevel(HttpLoggingInterceptor.Level.BODY);
        } else {
            loggingInterceptor.setLevel(HttpLoggingInterceptor.Level.NONE);
        }
        return loggingInterceptor;
    }

    public static ApiService getApiService() {
        return ApiServiceHolder.apiService;
    }

    private static class ApiServiceHolder {
        private static final ApiService apiService = new Retrofit.Builder()
                .baseUrl(Constant.BASE_URL)
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .client(getOkHttpClient())
                .build().create(ApiService.class);
    }

}

RxSchedulersHelper集中处理（网络错误、过滤等）

public class RxSchedulersHelper {
    public static final String TAG = RxSchedulersHelper.class.getSimpleName();
    public static <T> ObservableTransformer<T, T> compose() {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> observable) {
                return observable
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .filter(new Predicate<T>() {
                            @Override
                            public boolean test(@NonNull T t) throws Exception {
                                //这里可以对请求数据进行初步过滤，只处理code==200的结果，开发中结合实际处理，或许code==3001时你也需要做处理
                                int code = ((WrapBean) t).getCode();
//                                if (code == 3001) {
//                                    return true;
//                                } else {
//                                    return code == 200;
//                                }
                                return code == 200;
                            }
                        })
                        .onErrorResumeNext(new Function<Throwable, ObservableSource<? extends T>>() {
                            @Override
                            public ObservableSource<? extends T> apply(@NonNull Throwable throwable) throws Exception {
                                //网络异常处理
                                return Observable.error(NetException.throwable(throwable));
                            }
                        })
                        ;
            }
        };
    }
}

网络异常的处理

public class NetException {
    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;
    private static final int ACCESS_DENIED = 302;

    public static ResponseThrowable throwable(Throwable e) {
        ResponseThrowable ex;
        if (e instanceof HttpException) {
            HttpException httpException = (HttpException) e;
            ex = new ResponseThrowable(e, ERROR.HTTP_ERROR);
            switch (httpException.code()) {
                case REQUEST_TIMEOUT:
                case GATEWAY_TIMEOUT:
                    ex.message = "请求超时,请检查网络";
                    break;
                case UNAUTHORIZED:
                case FORBIDDEN:
                case NOT_FOUND:
                case INTERNAL_SERVER_ERROR:
                case BAD_GATEWAY:
                case SERVICE_UNAVAILABLE:
                case ACCESS_DENIED:
                    ex.message = "网络异常";
                    break;
                default:
                    ex.message = "网络错误";
                    break;
            }
            return ex;
        } else if (e instanceof RuntimeException) {
            ex = new ResponseThrowable(e, ERROR.RUNTIME);
            ex.message = "RuntimeException: " + e.getMessage();
            return ex;
//            throw new RuntimeException(e);
        } else if (
//                e instanceof JsonParseException||
                e instanceof JSONException
                        || e instanceof ParseException) {
            ex = new ResponseThrowable(e, ERROR.PARSE_ERROR);
            ex.message = "解析错误";
            return ex;
        } else if (e instanceof ConnectException) {
            ex = new ResponseThrowable(e, ERROR.NETWORD_ERROR);
            ex.message = "网络连接失败";
            return ex;
        } else if (e instanceof java.security.cert.CertPathValidatorException) {
            ex = new ResponseThrowable(e, ERROR.SSL_NOT_FOUND);
            ex.message = "证书路径没找到";
            return ex;
        } else if (e instanceof javax.net.ssl.SSLHandshakeException) {
            ex = new ResponseThrowable(e, ERROR.SSL_ERROR);
            ex.message = "证书验证失败";
            return ex;
        } else if (e instanceof java.net.SocketTimeoutException) {
            ex = new ResponseThrowable(e, ERROR.TIMEOUT_ERROR);
            ex.message = "连接超时,请稍后重试";
            return ex;
        } else {
            ex = new ResponseThrowable(e, ERROR.UNKNOWN);
            if (!SDNetWorkUtil.isNetWorkConnected()) {
                ex.message = "网络未连接";
            } else if (!SDNetWorkUtil.isAvailableByPing()) {
                ex.message = "网络不可用";
            } else {
                ex.message = "未知错误";
            }
            return ex;
        }
    }

    public static class ResponseThrowable extends Exception {
        public int code;
        public String message;

        public ResponseThrowable(Throwable cause) {
            super(cause);
        }

        public ResponseThrowable(Throwable throwable, int code) {
            super(throwable);
            this.code = code;

        }
    }

    /**
     * 约定异常
     */
    class ERROR {
        /**
         * 未知错误
         */
        public static final int UNKNOWN = 1000;
        /**
         * 解析错误
         */
        public static final int PARSE_ERROR = 1001;
        /**
         * 网络错误
         */
        public static final int NETWORD_ERROR = 1002;
        /**
         * 协议出错
         */
        public static final int HTTP_ERROR = 1003;

        /**
         * 证书出错
         */
        public static final int SSL_ERROR = 1005;

        /**
         * 连接超时
         */
        public static final int TIMEOUT_ERROR = 1006;

        /**
         * 证书未找到
         */
        public static final int SSL_NOT_FOUND = 1007;

        public static final int RUNTIME = 2017;

    }
}

很多时候我们的参数有公共参数，所以也可以封装到外层

public class Params {

    public static HashMap<String, String> baseParams(HashMap<String, String> map) {
//        map.put("base1", base1);
//        map.put("base2", base2);
        return map;
    }

    /**
     * @return
     */
    public static HashMap<String, String> getRecParams() {
        HashMap<String, String> map = new HashMap<>();
//        map.put("base", base);
        return baseParams(map);
    }
}

请求的封装

public class Request {

    public static Request getInstance() {
        return RequestHolder.request;
    }

    private static class RequestHolder {
        private static final Request request = new Request();
    }

    private static ApiService apiService = RetrofitManager.getApiService();

    public Observable<WrapBean> getCustom() {
        return apiService.getRecApi(Params.getCustomParams()).compose(RxSchedulersHelper.<WrapBean>compose());
    }
}

BaseActivity管理RXJava生命周期

public class BaseActivity extends Activity {
    private CompositeDisposable compositeDisposable = new CompositeDisposable();

    //管理RXJava生命周期
    public void addCompositeDisposable(Disposable disposable) {
        compositeDisposable.add(disposable);
    }
    //管理RXJava生命周期
    public void stopCompositeDisposable(Disposable disposable) {
        disposable.dispose();
    }
    @Override
    protected void onDestroy() {
        super.onDestroy();
        compositeDisposable.dispose();
    }
}

使用中：

public class MainActivity extends BaseActivity {
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        getData();
    }
    private void getData() {

        Request.getInstance().getCustom()
                .subscribe(new BaseObserver<WrapBean>() {
                    @Override
                    public void onError(NetException.ResponseThrowable e) {

                    }

                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        addCompositeDisposable(d);

                    }

                    @Override
                    public void onNext(@NonNull WrapBean wrapBean) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }
}