1.导入gradle

         //retrofit
        implementation 'com.squareup.retrofit2:retrofit:2.3.0'
        implementation 'com.squareup.retrofit2:converter-gson:2.3.0'
        implementation 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'
        //rxjava2
        implementation 'io.reactivex.rxjava2:rxjava:2.2.14'
        implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'
        //rxlifecycle
        implementation 'com.trello.rxlifecycle3:rxlifecycle:3.0.0'
        implementation 'com.trello.rxlifecycle3:rxlifecycle-android:3.0.0'
        implementation 'com.trello.rxlifecycle3:rxlifecycle-components:3.0.0'
        implementation 'androidx.lifecycle:lifecycle-extensions:2.2.0'

2.创建api类

    public IRetrofit iRetrofit;
    private  static Api mInstance =new Api();


    private Api() {
        initRetrofit();
    }

    public static Api getInstance() {
        return mInstance;
    }

    private void initRetrofit() {
        //日志拦截
        HttpLoggingInterceptor loggingInterceptor = new HttpLoggingInterceptor();
        loggingInterceptor.level(HttpLoggingInterceptor.Level.BODY);

        OkHttpClient okHttpClient = new OkHttpClient.Builder()
                .addInterceptor(loggingInterceptor)
                .connectTimeout(15,TimeUnit.SECONDS) //超时时间
                .readTimeout(15, TimeUnit.SECONDS)
                .writeTimeout(20, TimeUnit.SECONDS)
                .build();


        Gson gson = new GsonBuilder()
                .setLenient()
                .create();
        Retrofit mRetrofit = new Retrofit.Builder()
                .baseUrl("https://www.xnpool.com")
                .addConverterFactory(GsonConverterFactory.create(gson))
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .client(okHttpClient)
                .build();
        iRetrofit = mRetrofit.create(IRetrofit.class);
    }

3.创建接口类

    @FormUrlEncoded //post的请求必须的
    @Headers("Authorization:Basic dasdasdasdasdasdasdas") //这里是请求头
    @POST //请求类型
    fun getinfo(@Field("参数1") 参数1: String, @Field("参数2") 参数2: String): Observable<实体类>

4.配置 RxHelper 线程调度

  public class RxHelper {

    public static <T> ObservableTransformer<T, T> observableIO2Main(final Context context) {
        return upstream -> {
            Observable<T> observable = upstream.subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread());
            return composeContext(context, observable);
        };
    }

    public static <T> ObservableTransformer<T, T> observableIO2Main(final RxFragment fragment) {
        return upstream -> upstream.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()).compose(fragment.<T>bindToLifecycle());
    }

    public static <T> FlowableTransformer<T, T> flowableIO2Main() {
        return upstream -> upstream
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }

    private static <T> ObservableSource<T> composeContext(Context context, Observable<T> observable) {
        if(context instanceof RxActivity) {
            return observable.compose(((RxActivity) context).bindUntilEvent(ActivityEvent.DESTROY));
        } else if(context instanceof RxFragmentActivity){
            return observable.compose(((RxFragmentActivity) context).bindUntilEvent(ActivityEvent.DESTROY));
        }else if(context instanceof RxAppCompatActivity){
            return observable.compose(((RxAppCompatActivity) context).bindUntilEvent(ActivityEvent.DESTROY));
        }else {
            return observable;
        }
    }
}

4.调用方式（我这里进行了其他的处理，例如错误处理以及回掉的封装）

          Api.instance.init()
         Api.instance.service.getinfo(参数1，参数2)
               .compose(ApiSubscribe.compose())
               .subscribe(object : ApiResponse<BaseBeens<ResponseBody>>(context){
                   override fun success(data: BaseBeens<ResponseBody>) {
                       info.value=data
                   }
                   override fun failure(statusCode: Int, apiErrorModel: ApiErrorModel) {
                       Toast.makeText(context, apiErrorModel.message, Toast.LENGTH_SHORT).show()
                   }

               }
 )

5.封装回调

public abstract class ApiResponse<T> implements Observer<T> {
    private Context mContext;
    private Disposable d;
    private  Gloading.Holder holder;
    private ACache aCache;

    public ApiResponse(Context context, Gloading.Holder holder) {
        mContext=context;
        this.holder=holder;
        aCache=ACache.get(context);
    }

    public abstract void success(T data);


    @Override
    public void onSubscribe(final Disposable d) {
        this.d = d;
        if (!NetworkUtils.isConnected()) {
            if(holder!=null){
                holder.showLoadFailed();
            }
        }
    }

    @Override
    public void onNext(final T t) {
        success(t);
    }

    @RequiresApi(api = Build.VERSION_CODES.M)
    @Override
    public void onError(Throwable e) {
        ApiException.exceptionHandler(e);
    
        if (d.isDisposed()) {
            d.dispose();
        }
    }

    @Override
    public void onComplete() {
        if (d.isDisposed()) {
            d.dispose();
        }
    }

    /**
     * 取消订阅
     */
    public void cancleRequest(){
        if (d!=null&&d.isDisposed()) {
            d.dispose();
            if(holder!=null){
                holder.showLoadFailed();
           
        }
    }

}

6.错误处理

public class ApiException {
    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 SERVICE_UNAVAILABLE = 503;


    public static void exceptionHandler(Throwable e){
        LogUtils.e("Retrofit", e.getMessage());
        if (e instanceof HttpException) {     //   HTTP错误
            HttpException httpException = (HttpException) e;
            switch (httpException.code()) {
                case UNAUTHORIZED:
                    ToastUtils.showShort("账号密码错误");
                    break;
                case FORBIDDEN:
                    ToastUtils.showShort("请求被拒绝", Toast.LENGTH_SHORT);
                    break;
                case NOT_FOUND:
                    ToastUtils.showShort("资源不存在", Toast.LENGTH_SHORT);
                    break;
                case REQUEST_TIMEOUT:
                    ToastUtils.showShort("服务器执行超时", Toast.LENGTH_SHORT);
                    break;
                case INTERNAL_SERVER_ERROR:
                    ToastUtils.showShort("服务器内部错误", Toast.LENGTH_SHORT);
                    break;
                case SERVICE_UNAVAILABLE:
                    ToastUtils.showShort("服务器不可用", Toast.LENGTH_SHORT);
                    break;
                default:
                    ToastUtils.showShort("网络错误", Toast.LENGTH_SHORT);
                    break;
            }
        } else if (e instanceof ConnectException
                || e instanceof UnknownServiceException) {   //   连接错误
            ToastUtils.showShort("连接失败", Toast.LENGTH_SHORT);
        } else if (e instanceof SocketTimeoutException) {   //  连接超时
            ToastUtils.showShort("连接超时", Toast.LENGTH_SHORT);
        } else if (e instanceof JsonParseException
                || e instanceof JSONException
                || e instanceof ParseException) {   //  解析错误
            ToastUtils.showShort("解析错误", Toast.LENGTH_SHORT);
        }else {
            ToastUtils.showShort("未知错误", Toast.LENGTH_SHORT);
        }
    }

}

使用方式

        Api.getInstance().iRetrofit.poolstate(url)
.compose(RxHelper.observableIO2Main(context))
. subscribe(new ApiResponse<Respon<List<PoolState>>>(context,holder) {
                    public void success(Respon<List<PoolState>> data) {
                        allPoolStart.setValue(data);
                    }
                });

至此 使用即可。 java 也是一样的 包结构: