研究了两个小时，终于让我明白了Retrofit和RxJava是如何结合在一起的，分享给大家。本文将会从源码角度来看这两个家伙是如何配合的。阅读前请确保理解Retrofit源码和RxJava的基本使用。
支持原创，转载请注明出处。
类图
Retrofit&RxJava.png
使用场景
//创建接口
public interface IShot {
//返回一个Observable，数据类型是List<Shot>
Observable<Response<List<Shot>>> getShot(@Query("page") int page);
}
//创建Retrofit
Retrofit retrofit = new Retrofit.Builder()
.baseUrl(PeanutInfo.URL_BASE)
.addConverterFactory(GsonConverterFactory.create())//添加GsonConverter
.addCallAdapterFactory(RxJavaCallAdapterFactory.create())//添加RxJavaCallAdapter
.build();
//创建代理对象
IShot server = retrofit.create(Shot.class);
//发送请求，返回Observable
Observable<Response<List<Shot>>> observable = server.getShot(1);
//传入Subscribe，进行订阅
observable.subscribe(new Subscribe....);
创建Observable
在创建Retrofit时我们添加了GsonConverter和RxJavaCallAdapter。我们回想下Retrofit源码，在调用了代理对象的getShot方法后会调用MethodHandler的invoke方法：
Object invoke(Object... args) {
return callAdapter.adapt(
new OkHttpCall<>(callFactory, requestFactory, args, responseConverter));
}
这里的callAdpter是由RxJavaCallAdapterFactory返回的，因为我们指定了CallAdapterFactory为RxJavaCallAdapterFactory，我们看看RxJavaCallAdapterFactory的源码，这个类在retrofit2/adapter包下：
public final class RxJavaCallAdapterFactory extends CallAdapter.Factory
---------------------------------get方法---------------------------------------------
@Override
public CallAdapter<?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
Class<?> rawType = getRawType(returnType); //获取范型类型，这里是Response<List<Shot>>
boolean isSingle = "rx.Single".equals(rawType.getCanonicalName());
if (rawType != Observable.class && !isSingle) {
return null;
}
if (!(returnType instanceof ParameterizedType)) {
String name = isSingle ? "Single" : "Observable";
throw new IllegalStateException(name + " return type must be parameterized"
+ " as " + name + "<Foo> or " + name + "<? extends Foo>");
}
CallAdapter<Observable<?>> callAdapter = getCallAdapter(returnType); //根据返回类型获取CallAdapter
if (isSingle) {
// Add Single-converter wrapper from a separate class. This defers classloading such that
// regular Observable operation can be leveraged without relying on this unstable RxJava API.
return SingleHelper.makeSingle(callAdapter);
}
return callAdapter;
}
这个方法里根据返回类型returnType获取范型类型rawType，我们这里是Response<List<Shot>>，然后根据rawType获取CallAdapter，我们接着看下getCallAdapter方法：
private CallAdapter<Observable<?>> getCallAdapter(Type returnType) {
Type observableType = getParameterUpperBound(0, (ParameterizedType) returnType);
Class<?> rawObservableType = getRawType(observableType);
if (rawObservableType == Response.class) { //Response类型
if (!(observableType instanceof ParameterizedType)) {
throw new IllegalStateException("Response must be parameterized"
+ " as Response<Foo> or Response<? extends Foo>");
}
Type responseType = getParameterUpperBound(0, (ParameterizedType) observableType);
return new ResponseCallAdapter(responseType); //返回ResponseCallAdapter
}
if (rawObservableType == Result.class) {
if (!(observableType instanceof ParameterizedType)) {
throw new IllegalStateException("Result must be parameterized"
+ " as Result<Foo> or Result<? extends Foo>");
}
Type responseType = getParameterUpperBound(0, (ParameterizedType) observableType);
return new ResultCallAdapter(responseType); //返回ResultCallAdapter
}
return new SimpleCallAdapter(observableType); //返回SimpleCallAdapter
}
该方法根据返回值的具体类型返回对应的CallAdapter，我们的类型是Response<List<Shot>>，对应Response类，所以返回ResponseCallAdapter，终于找到了CallAdapter类了。再来回顾下MethodHandler的invoke方法：
Object invoke(Object... args) {
return callAdapter.adapt(
new OkHttpCall<>(callFactory, requestFactory, args, responseConverter));
}
我们看下ResponseCallAdapter的adapt()方法：
static final class ResponseCallAdapter implements CallAdapter<Observable<?>> {
private final Type responseType;
ResponseCallAdapter(Type responseType) {
this.responseType = responseType;
}
@Override public Type responseType() {
return responseType;
}
@Override public <R> Observable<Response<R>> adapt(Call<R> call) {
return Observable.create(new CallOnSubscribe<>(call)); //创建一个CallOnSubscribe，用这个CallOnSubscribe创建Observable，并返回。
}
}
到这里终于创建好了Observable。我们看下CallOnSubscribe这个类：
static final class CallOnSubscribe<T> implements Observable.OnSubscribe<Response<T>> {
private final Call<T> originalCall; //引用OkHttpCall
CallOnSubscribe(Call<T> originalCall) {
this.originalCall = originalCall; //传入的是OkHttpCall
}
@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();
subscriber.add(Subscriptions.create(new Action0() {
@Override public void call() { //订阅时调用这个方法
call.cancel();
}
}));
try {
Response<T> response = call.execute(); //执行网络请求
if (!subscriber.isUnsubscribed()) {
subscriber.onNext(response); //将结果传入subscriber.onNext
}
} catch (Throwable t) {
Exceptions.throwIfFatal(t);
if (!subscriber.isUnsubscribed()) {
subscriber.onError(t); //出现错误是调用 subscriber.onError
}
return;
}
if (!subscriber.isUnsubscribed()) {
subscriber.onCompleted();
}
}
}
CallOnSubscribe实现了OnSubscribe接口，内部引用了OkHttpCall，执行真正的网络请求。进行订阅时会调用call方法，该方法调用OkHttpCall.execute获取响应，传递给subscriber.onNext这样就发生了一次订阅。到这里Retrofit和RxJava就完美结合了。
最后看下类图，就更清楚了。
Retrofit&RxJava.png
总结
Retrofit添加了RxJavaCallAdapterFactory后，对应的callAdapter就是ResponseAdapter或SimpleCallAdapter类，他们的adapt方法返回一个Observable对象，这个Observable内部引用了真正执行网络请求的OkHttpCall。
后面会将更多笔记整理成博客，欢迎关注。
支持原创，转载请注明出处。
