Android RxJava与Retrofit结合使用详解

public class HttpRequest {
    public static final long CONNECTTIME = 30000;
    public static final String BASE_URL = "http://jxhdapi.ooowin.com/";
    private  ApiService apiService;
    public HttpRequest() {
        //添加日志拦截器
        HttpLoggingInterceptor interceptor = new HttpLoggingInterceptor(new HttpLoggingInterceptor.Logger() {
            @Override
            public void log(String message) {
                Log.d("TAG", "==========" + message);
            }
        }).setLevel(HttpLoggingInterceptor.Level.BODY);
        //获取OkHttpClient
        OkHttpClient client = new OkHttpClient.Builder()
                .connectTimeout(CONNECTTIME, TimeUnit.MICROSECONDS)
                .readTimeout(CONNECTTIME,TimeUnit.MICROSECONDS)
                .writeTimeout(CONNECTTIME,TimeUnit.MICROSECONDS)
                .addInterceptor(interceptor)
                .addNetworkInterceptor(new HttpHeaderInterceptor())
                .build();
        //初始化Retrofit
        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl(BASE_URL)
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .client(client)
                .build();
        apiService = retrofit.create(ApiService.class);
    }
    //  创建单例
    private static class SingletonHolder {
        private static final HttpRequest INSTANCE = new HttpRequest();
    }
    public static ApiService getApi(){
        return SingletonHolder.INSTANCE.apiService;
    }
}

public class HttpHeaderInterceptor implements Interceptor {
    @Override
    public Response intercept(Chain chain) throws IOException {
        Request request = chain.request();
        Request build = request.newBuilder()
//                .addHeader("","")   添加header
                .build();
        return chain.proceed(build);
    }
}

{
    "code": 1,
    "msg": "操作成功",
    "data": {······}
}

public class BaseBean<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;
    }
}

public interface ApiService {
    //获取省列表
    @GET("common/areas")
    Flowable<BaseBean<List<Province>>> province();
}

HttpRequest.getApi()
            .province()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<BaseBean<List<Province>>>() {
                @Override
                public void onSubscribe(Subscription s) {
                }
                @Override
                public void onNext(BaseBean<List<Province>> listBaseBean) {
                }
                @Override
                public void onError(Throwable t) {
                }
                @Override
                public void onComplete() {
                }
            });
}

public class SchedulersHelper implements FlowableTransfORMer{
    @Override
    public Publisher apply(Flowable upstream) {
        return  upstream.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }
}

public abstract class MySubscriber<T> implements Subscriber<T>{
    @Override
    public void onSubscribe(Subscription s) {
        s.request(Long.MAX_VALUE);
        showLoading();
    }
    @Override
    public void onNext(T t) {
        //code为1代表服务器返回成功的数据
        if (((BaseBean)t).getCode() == 1) {
            //成功后返回data数据进行处理即可
            onSuccess((T) ((BaseBean) t).getData());
        }else {
            //处理服务器返回错误code
        }
    }
    @Override
    public void onComplete() {
        finishLoading();
    }
    @Override
    public void onError(Throwable t) {
        finishLoading();
        //处理网络异常
        Log.d("TAG","=========" + t);
    }
    protected abstract void onSuccess(T t);
    protected abstract void showLoading();
    protected abstract void finishLoading();
}

HttpRequest.getApi().province()
        .compose(new SchedulersHelper())
        .subscribe(new MySubscriber() {
            @Override
            protected void onSuccess(Object o) {
            }
            @Override
            protected void showLoading() {
            }
            @Override
            protected void finishLoading() {
            }
        });