RxJava 与 Retrofit 结合

文章
给 Android 开发者的 RxJava 详解
 Android Retrofit 2.0 使用-补充篇

RxJava 与 Retrofit 封装
1.创建一个对象HttpMethods

public class HttpMethods {    
    public static final String BASE_URL = "https://api.douban.com/v2/movie/";
    private static final int DEFAULT_TIMEOUT = 5;
    public final static int READ_TIMEOUT=5;
    public final static int WRITE_TIMEOUT=5;
    private static Retrofit retrofit;
    public static HttpService mHttpService;
    //构造方法私有
    private HttpMethods() {
        //手动创建一个OkHttpClient并设置超时时间
        OkHttpClient.Builder builder = new OkHttpClient.Builder()
                .connectTimeout(DEFAULT_TIMEOUT, TimeUnit.SECONDS)//设置连接超时时间
                .readTimeout(READ_TIMEOUT,TimeUnit.SECONDS)//设置读取超时时间
                .writeTimeout(WRITE_TIMEOUT,TimeUnit.SECONDS)//设置写的超时时间
                .retryOnConnectionFailure(true);//错误重连
        retrofit = new Retrofit.Builder()
                .client(builder.build())
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                .baseUrl(BASE_URL)
                .build();
        mHttpService = retrofit.create(HttpService.class);
    }
    //在访问HttpMethods时创建单例
    private static class SingletonHolder{
        private static final HttpMethods INSTANCE = new HttpMethods();
    }
    //获取单例
    public static HttpMethods getInstance(){
        return SingletonHolder.INSTANCE;
    }
}

2.封装相同格式的Http请求数据
我们可以创建一个HttpResult类,以我的demo为例：

public class HttpResult<T> {
 //用来模仿resultCode和resultMessage
 private int count;
 private int start;
 private int total;
 private String title;
 //用来模仿Data
 private T subjects;
}

这样泛型的时候就要写为：

Observable<HttpResult<List<Subject>>>

3.相同格式的Http请求数据统一进行预处理

/**
 * 用来统一处理Http的resultCode,并将HttpResult的Data部分剥离出来返回给subscriber
 * @param <T>   Subscriber真正需要的数据类型，也就是Data部分的数据类型
 */
public class HttpResultFunc<T> implements Func1<HttpResult<T>, T> {
    @Override    public T call(HttpResult<T> httpResult) {
        if (httpResult.getCount() == 0) {
            throw new ApiException(100);
        }
        return httpResult.getSubjects();
    }
}

4.如果你觉得写更改线程的代码觉得也很烦的话，可以把订阅这部分也封装起来：

public class AppAction {
    public static HttpMethods mHttpMethods;
    public AppAction() {
        mHttpMethods = HttpMethods.getInstance();
    }
    /**
     * 用于获取豆瓣电影Top250的数据
     * @param subscriber  由调用者传过来的观察者对象
     * @param start 起始位置
     * @param count 获取长度
     */
    public void getTopMovie(Subscriber<List<Subject>> subscriber, int start, int count){
       Observable observable = HttpMethods.mHttpService.getTopMovie(start, count)
                .map(new HttpResultFunc<List<Subject>>());
        toSubscribe(observable, subscriber);
    }
    private <T> void toSubscribe(Observable<T> o, Subscriber<T> s){
        o.subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(s);
    }
}

取消一个Http请求

如果没有使用Rxjava，那么Service返回的是一个Call，而这个Call对象有一个cancel方法可以用来取消Http请求。那么用了Rxjava之后，如何来取消一个请求呢？因为返回值是一个Observable。我们能做的似乎只有解除对Observable对象的订阅，其他的什么也做不了。

好在Retrofit已经帮我们考虑到了这一点。答案在RxJavaCallAdapterFactory这个类的源码中可以找到

static final class CallOnSubscribe<T> implements Observable.OnSubscribe<Response<T>> {
 private final Call<T> originalCall; CallOnSubscribe(Call<T> originalCall) {
 this.originalCall = originalCall;
 }
 @Override
 public void call(final Subscriber<? super Response<T>> subscriber) {
 // Since Call is a one-shot type, clone it for each new subscriber.
 final Call<T> call = originalCall.clone();
 // Attempt to cancel the call if it is still in-flight on unsubscription.
 subscriber.add(Subscriptions.create(new Action0() {
 @Override
 public void call() {
 call.cancel();
 }
 }));
 try { Response<T> response = call.execute(); if (!subscriber.isUnsubscribed()) {
 subscriber.onNext(response);
 }
 } catch (Throwable t)
 {
 Exceptions.throwIfFatal(t);
 if (!subscriber.isUnsubscribed()) {
 subscriber.onError(t);
 }
 return;
 }
 if (!subscriber.isUnsubscribed()) {
 subscriber.onCompleted();
 } 
}}

我们看到call方法中，给subscriber添加了一个Subscription对象，Subscription对象很简单，主要就是取消订阅用的，如果你查看Subscriptions.create的源码，发现是这样的

public static Subscription create(final Action0 unsubscribe) {
 return BooleanSubscription.create(unsubscribe);
}

总结起来就是说，我们在Activity或者Fragment中创建subscriber对象，想要取消请求的时候调用subscriber的unsubscribe方法就可以了。

6.封装ProgressDialog的Subscriber样子

我们先来创建一个类，就叫ProgressSubscriber，让他继承Subscriber。
Subscriber给我们提供了onStart、onNext、onError、onCompleted四个方法。
其中只有onNext方法返回了数据，那我们自然希望能够在onNext里面处理数据相关的逻辑。
onStart方法我们用来启动一个ProgressDialog。 onError方法我们集中处理错误，同时也停止ProgressDialog onComplated方法里面停止ProgressDialog

处理onNext
我们先来定义一个接口，命名SubscriberOnNextListener

public interface SubscriberOnNextListener<T> { void onNext(T t);}

我们希望当cancel掉ProgressDialog的时候，能够取消订阅，也就取消了当前的Http请求。所以我们先来创建个接口来处理这件事情。

public interface ProgressCancelListener { void onCancelProgress();}

然后我们用ProgressSubscriber来实现这个接口，这样ProgressSubscriber就有了一个onCancelProgress方法，在这里面取消订阅。

@Override
public void onCancelProgress() {
 if (!this.isUnsubscribed()) {
 this.unsubscribe();
 }
}

然后我用了一个Handler来封装了ProgressDialog。

public class ProgressDialogHandler extends Handler {
 public static final int SHOW_PROGRESS_DIALOG = 1;
 public static final int DISMISS_PROGRESS_DIALOG = 2;
 private ProgressDialog pd;
 private Context context;
 private boolean cancelable;
 private ProgressCancelListener mProgressCancelListener;
 public ProgressDialogHandler(Context context, ProgressCancelListener mProgressCancelListener, boolean cancelable) {
 super();
 this.context = context;
 this.mProgressCancelListener = mProgressCancelListener;
 this.cancelable = cancelable;
 }
 private void initProgressDialog(){
 if (pd == null) {
 pd = new ProgressDialog(context);
 pd.setCancelable(cancelable);
 if (cancelable) {
 pd.setOnCancelListener(new DialogInterface.OnCancelListener() {
 @Override
 public void onCancel(DialogInterface dialogInterface) {
 mProgressCancelListener.onCancelProgress();
 }});
 }
 if (!pd.isShowing()) {
 pd.show();
 }
 } }
 private void dismissProgressDialog(){
 if (pd != null) {
 pd.dismiss();
 pd = null;
 } }
 @Override
 public void handleMessage(Message msg) {
 switch (msg.what) {
 case SHOW_PROGRESS_DIALOG:
 initProgressDialog();
 break;
 case DISMISS_PROGRESS_DIALOG:
 dismissProgressDialog();
 break;
 } }
}

再来看一下ProgressSubscriber的代码

public class ProgressSubscriber<T> extends Subscriber<T> implements ProgressCancelListener {
    private SubscriberOnNextListener mSubscriberOnNextListener;
    private ProgressDialogHandler mProgressDialogHandler;
    private Context context;
    public ProgressSubscriber(SubscriberOnNextListener mSubscriberOnNextListener, Context context) {
        this.mSubscriberOnNextListener = mSubscriberOnNextListener;
        this.context = context;
        mProgressDialogHandler = new ProgressDialogHandler(context, this, true);
    }
    private void showProgressDialog(){
        if (mProgressDialogHandler != null) {
            mProgressDialogHandler.obtainMessage(ProgressDialogHandler.SHOW_PROGRESS_DIALOG).sendToTarget();
        }
    }
    private void dismissProgressDialog(){
        if (mProgressDialogHandler != null) {
            mProgressDialogHandler.obtainMessage(ProgressDialogHandler.DISMISS_PROGRESS_DIALOG).sendToTarget();
            mProgressDialogHandler = null;
        }
    }
    /**
     * 订阅开始时调用
     * 显示ProgressDialog
     */
    @Override
    public void onStart() {
        showProgressDialog();
    }
    /**
     * 完成，隐藏ProgressDialog
     */
    @Override
    public void onCompleted() {
        dismissProgressDialog();
        Toast.makeText(context, "Get Top Movie Completed", Toast.LENGTH_SHORT).show();
    }
    /**
     * 对错误进行统一处理
     * 隐藏ProgressDialog
     * @param e
     */
    @Override
    public void onError(Throwable e) {
        if (e instanceof SocketTimeoutException) {
            Toast.makeText(context, "网络中断，请检查您的网络状态", Toast.LENGTH_SHORT).show();
        } else if (e instanceof ConnectException) {
            Toast.makeText(context, "网络中断，请检查您的网络状态", Toast.LENGTH_SHORT).show();
        }  else if (e instanceof Exception) {
            Toast.makeText(context, "网络中断，请检查您的网络状态", Toast.LENGTH_SHORT).show();
        } else {
            Toast.makeText(context, "error:" + e.getMessage(), Toast.LENGTH_SHORT).show();
        }
        dismissProgressDialog();
    }
    /**
     * 将onNext方法中的返回结果交给Activity或Fragment自己处理
     *
     * @param t 创建Subscriber时的泛型类型
     */    @Override    public void onNext(T t) {
        if (mSubscriberOnNextListener != null) {
            mSubscriberOnNextListener.onNext(t);
        }
    }
    /**
     * 取消ProgressDialog的时候，取消对observable的订阅，同时也取消了http请求
     */
    @Override
    public void onCancelProgress() {
        if (!this.isUnsubscribed()) {
            this.unsubscribe();
        }
    }
}

MainActivity使用是这样的：
先来定义一个SubscriberOnNextListener对象，可以在onCreate里面创建这个对象

private SubscriberOnNextListener getTopMovieOnNext;
@Override
protected void onCreate(Bundle savedInstanceState) {
 super.onCreate(savedInstanceState);
 setContentView(R.layout.activity_main);
 ButterKnife.bind(this);
 getTopMovieOnNext = new SubscriberOnNextListener<List<Subject>>() {
 @Override
 public void onNext(List<Subject> subjects) {
 resultTV.setText(subjects.toString());
 }};
}

getMovie方法这么写：

private void getMovie(){
 HttpMethods.getInstance().getTopMovie( new ProgressSubscriber(getTopMovieOnNext, MainActivity.this), 0, 10);
}

博客新手写的不咋滴，欢迎吐槽~~项目地址 https://github.com/pengwenliang/NowsTop-master

RxJava 与 Retrofit 结合

推荐阅读更多精彩内容