我们在前面花了大量的笔墨讲了下面两篇文章：
主要是为了理清楚retrofit2和rxjava2的工作原理为这一篇两者结合使用做铺垫
一、这里我们先来看看两者是怎么结合使用的(这里依然以登录接口为例)，与第一篇类似依然是三个步骤：
1.定义一个登录接口
public interface APIFunction {
/**
* 登录请求
* @param map 请求参数
* @return
*/
@POST(HttpConfig.REQUEST_LOGIN)
@Headers("Content-Type:application/vnd.api+json")
Observable<ResponseBean<LoginBean>> login(@Body Map<String, Object> map);
}
2.初始化retrofit：
//共通参数拦截器
HeaderParamInterceptor commonParamInterceptor = new HeaderParamInterceptor();
// 初始化okhttp
OkHttpClient client = new OkHttpClient.Builder()
.addInterceptor(commonParamInterceptor)
.build();
// 初始化Retrofit
mRetrofit = new Retrofit.Builder()
.client(client)
.baseUrl(HttpConfig.IP)
.addCallAdapterFactory(RxJava2CallAdapterFactory.create())//添加rxjava适配器
.addConverterFactory(GsonConverterFactory.create())//添加gson转换器
.build();
// 初始化Service
mApiFunction = mRetrofit.create(APIFunction.class);
3.进行网络请求操作：
RetrofitFactory.getInstance().API().login(map)
.subscribeOn(Schedulers.io())//此处必须将网络请求切到子线程上面去
.observeOn(AndroidSchedulers.mainThread())//UI操作切回到主线程
.subscribe(new Observer<ResponseBean<LoginBean>>() {
@Override
public void onSubscribe(Disposable d) {
//做网络请求前的准备工作
}
@Override
public void onNext(ResponseBean<LoginBean> loginBeanResponseBean) {
//网络请求成功以后的处理
}
@Override
public void onError(Throwable e) {
//网络请求失败以后的处理
}
@Override
public void onComplete() {
//整个流程结束以后的处理
}
});
二、看了上面几个步骤，可能会有几个疑问：
- rxjava2为什么可以和retrofit2完美结合？
- retrofit2的enqueue方法可以将网络请求放到线程里面，然后回调的处理切回到主线程，为什么这里还要使用rxjava2切换线程的方法？
- 第三步订阅以后是如何进行网络请求的？
先别急，看完下面的解析就能一目了然了。
三、那么我们一探究竟吧：
（关于retrofit2和rxjava2的原理这里就不多做介绍了，主要讲解这两者结合使用的地方）
1.我们第一步里面将将接口返回对象改成了Observable，这样就可以直接使用rxjava2的操作符以及订阅方法了
2.第二个问题的话我们需要看第二步里面的一个操作：
.addCallAdapterFactory(RxJava2CallAdapterFactory.create())//添加rxjava适配器
通过第一篇文章我们知道这里添加的是一个适配器，主要用于ServiceMethod.adapt使用：
T adapt(Call<R> call) {
return callAdapter.adapt(call);
}
这里的callAdapter就是上面配置的RxJava2CallAdapter。所以接下来看看RxJava2CallAdapter的代码吧
final class RxJava2CallAdapter<R> implements CallAdapter<R, Object> {
...代码省略...
@Override public Object adapt(Call<R> call) {
Observable<Response<R>> responseObservable = isAsync
? new CallEnqueueObservable<>(call)
: new CallExecuteObservable<>(call);
Observable<?> observable;
if (isResult) {
observable = new ResultObservable<>(responseObservable);
} else if (isBody) {
observable = new BodyObservable<>(responseObservable);
} else {
observable = responseObservable;
}
if (scheduler != null) {
observable = observable.subscribeOn(scheduler);
}
if (isFlowable) {
return observable.toFlowable(BackpressureStrategy.LATEST);
}
if (isSingle) {
return observable.singleOrError();
}
if (isMaybe) {
return observable.singleElement();
}
if (isCompletable) {
return observable.ignoreElements();
}
return observable;
}
}
public final class RxJava2CallAdapterFactory extends CallAdapter.Factory {
/**
* Returns an instance which creates synchronous observables that do not operate on any scheduler
* by default.
*/
public static RxJava2CallAdapterFactory create() {
return new RxJava2CallAdapterFactory(null, false);
}
...代码省略...
private final @Nullable Scheduler scheduler;
private final boolean isAsync;
private RxJava2CallAdapterFactory(@Nullable Scheduler scheduler, boolean isAsync) {
this.scheduler = scheduler;
this.isAsync = isAsync;
}
...代码省略...
}
因为在添加RxJava2CallAdapter的时候，调用的是RxJava2CallAdapterFactory.create()方法，所以上面代码中isAsync是false，因此我们使用的CallExecuteObservable，看这个名字大致就能猜出这个是一个同步的操作方法。当然这种不能靠猜，还是得看看到底是不是这样实现的
final class CallExecuteObservable<T> extends Observable<Response<T>> {
...代码省略...
@Override protected void subscribeActual(Observer<? super Response<T>> observer) {
// Since Call is a one-shot type, clone it for each new observer.
Call<T> call = originalCall.clone();
...代码省略...
boolean terminated = false;
try {
Response<T> response = call.execute();//执行retrofit的execute方法
...代码省略...
} catch (Throwable t) {
...代码省略...
}
}
...代码省略...
}
从代码中可以发现这里是执行了call.exexute，这就跟我们第一篇文章讲第三步讲的网络请求一样了，只不过这里用了同步的操作方法。
因此我们这一篇文章的第三步里面需要用rxjava2将网络请求操作放到子线程中。
3.最后一个问题的话依然回到我们的CallExecuteObservable中来看看：
final class CallExecuteObservable<T> extends Observable<Response<T>> {
private final Call<T> originalCall;
CallExecuteObservable(Call<T> originalCall) {
this.originalCall = originalCall;
}
@Override protected void subscribeActual(Observer<? super Response<T>> observer) {
// Since Call is a one-shot type, clone it for each new observer.
Call<T> call = originalCall.clone();
CallDisposable disposable = new CallDisposable(call);
observer.onSubscribe(disposable);
boolean terminated = false;
try {
Response<T> response = call.execute();
if (!disposable.isDisposed()) {
observer.onNext(response);
}
if (!disposable.isDisposed()) {
terminated = true;
observer.onComplete();
}
} catch (Throwable t) {
Exceptions.throwIfFatal(t);
if (terminated) {
RxJavaPlugins.onError(t);
} else if (!disposable.isDisposed()) {
try {
observer.onError(t);
} catch (Throwable inner) {
Exceptions.throwIfFatal(inner);
RxJavaPlugins.onError(new CompositeException(t, inner));
}
}
}
}
private static final class CallDisposable implements Disposable {
private final Call<?> call;
private volatile boolean disposed;
CallDisposable(Call<?> call) {
this.call = call;
}
@Override public void dispose() {
disposed = true;
call.cancel();
}
@Override public boolean isDisposed() {
return disposed;
}
}
}
仿照这位博主的流程图，做了个简易的关于retrofit2+rxjava2的流程图：
retrofit2+rxjava2流程图 .png
我们知道subscribeActual这个方法是在各自的Observable的subscribe方法里面调用的，所以在第三步的subscribe方法执行以后，最终会执行CallExecuteObservable的subscribeActual方法。
总结
retrofit2+rxjava2结合使用的代码量不多，最主要的两个类就是CallExecuteObservable和RxJava2CallAdapter。只要熟悉了前面两篇文章的原理，那么这个基本上就是小意思。
