研究了两个小时，终于让我明白了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。