我們在前面花了大量的筆墨講了下面兩篇文章：
主要是為了理清楚retrofit2和rxjava2的工作原理為這一篇兩者結(jié)合使用做鋪墊
一逻悠、這里我們先來看看兩者是怎么結(jié)合使用的(這里依然以登錄接口為例)，與第一篇類似依然是三個步驟：
1.定義一個登錄接口
public interface APIFunction {
/**
* 登錄請求
* @param map 請求參數(shù)
* @return
*/
@POST(HttpConfig.REQUEST_LOGIN)
@Headers("Content-Type:application/vnd.api+json")
Observable<ResponseBean<LoginBean>> login(@Body Map<String, Object> map);
}
2.初始化retrofit：
//共通參數(shù)攔截器
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轉(zhuǎn)換器
.build();
// 初始化Service
mApiFunction = mRetrofit.create(APIFunction.class);
3.進(jìn)行網(wǎng)絡(luò)請求操作：
RetrofitFactory.getInstance().API().login(map)
.subscribeOn(Schedulers.io())//此處必須將網(wǎng)絡(luò)請求切到子線程上面去
.observeOn(AndroidSchedulers.mainThread())//UI操作切回到主線程
.subscribe(new Observer<ResponseBean<LoginBean>>() {
@Override
public void onSubscribe(Disposable d) {
//做網(wǎng)絡(luò)請求前的準(zhǔn)備工作
}
@Override
public void onNext(ResponseBean<LoginBean> loginBeanResponseBean) {
//網(wǎng)絡(luò)請求成功以后的處理
}
@Override
public void onError(Throwable e) {
//網(wǎng)絡(luò)請求失敗以后的處理
}
@Override
public void onComplete() {
//整個流程結(jié)束以后的處理
}
});
二弯囊、看了上面幾個步驟肾档，可能會有幾個疑問：
- rxjava2為什么可以和retrofit2完美結(jié)合臊旭？
- retrofit2的enqueue方法可以將網(wǎng)絡(luò)請求放到線程里面，然后回調(diào)的處理切回到主線程，為什么這里還要使用rxjava2切換線程的方法庐完？
- 第三步訂閱以后是如何進(jìn)行網(wǎng)絡(luò)請求的舶治？
先別急分井，看完下面的解析就能一目了然了。
三霉猛、那么我們一探究竟吧：
（關(guān)于retrofit2和rxjava2的原理這里就不多做介紹了尺锚，主要講解這兩者結(jié)合使用的地方）
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的時候，調(diào)用的是RxJava2CallAdapterFactory.create()方法坛悉，所以上面代碼中isAsync是false伐厌，因此我們使用的CallExecuteObservable，看這個名字大致就能猜出這個是一個同步的操作方法裸影。當(dāng)然這種不能靠猜挣轨，還是得看看到底是不是這樣實現(xiàn)的
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();//執(zhí)行retrofit的execute方法
...代碼省略...
} catch (Throwable t) {
...代碼省略...
}
}
...代碼省略...
}
從代碼中可以發(fā)現(xiàn)這里是執(zhí)行了call.exexute，這就跟我們第一篇文章講第三步講的網(wǎng)絡(luò)請求一樣了轩猩，只不過這里用了同步的操作方法卷扮。
因此我們這一篇文章的第三步里面需要用rxjava2將網(wǎng)絡(luò)請求操作放到子線程中。
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;
}
}
}
仿照這位博主的流程圖均践，做了個簡易的關(guān)于retrofit2+rxjava2的流程圖：
retrofit2+rxjava2流程圖 .png
我們知道subscribeActual這個方法是在各自的Observable的subscribe方法里面調(diào)用的晤锹，所以在第三步的subscribe方法執(zhí)行以后，最終會執(zhí)行CallExecuteObservable的subscribeActual方法彤委。
總結(jié)
retrofit2+rxjava2結(jié)合使用的代碼量不多鞭铆，最主要的兩個類就是CallExecuteObservable和RxJava2CallAdapter。只要熟悉了前面兩篇文章的原理葫慎，那么這個基本上就是小意思衔彻。
