Retrofit+RxJava请求一次接口全过程分析

我们定义接口请求的接口类

interface Api {
    // 登录
    @POST("v1/crmpassport/login")
    fun login(@Body body: Map<String, String>): Observable<ResponseJson<User>>
}
// 得到接口的实例对象
T = retrofit.create(clazz)
// 调用方法，返回一个Call或者Observable对象， 请求网络
T.login(..): Call or Observable

Retrofit的create方法，是一个动态代理

  public <T> T create(final Class<T> service) {
    Utils.validateServiceInterface(service);
   ...
    return (T) Proxy.newProxyInstance(service.getClassLoader(), new Class<?>[] { service },
        new InvocationHandler() {
          private final Platform platform = Platform.get();

          @Override public Object invoke(Object proxy, Method method, @Nullable Object[] args)
              throws Throwable {
      ...
            ServiceMethod<Object, Object> serviceMethod =
                (ServiceMethod<Object, Object>) loadServiceMethod(method);
            OkHttpCall<Object> okHttpCall = new OkHttpCall<>(serviceMethod, args);
            return serviceMethod.adapt(okHttpCall);
          }
        });
  }

生成的动态代理类中，接口Api中的方法，都会在代理类中生成一个方法，不过方法里中执行的是InvocationHandler的invoke方法，再通过反射原理执行调用的方法。
//TODO 生成的代理类，后续补上。

所以我们调用Api.login(...)时会最终会执行到

new InvocationHandler() {
          @Override public Object invoke(Object proxy, Method method, @Nullable Object[] args)
              throws Throwable {
            ...
            ServiceMethod<Object, Object> serviceMethod =
                (ServiceMethod<Object, Object>) loadServiceMethod(method);
            OkHttpCall<Object> okHttpCall = new OkHttpCall<>(serviceMethod, args);
            return serviceMethod.adapt(okHttpCall);
          }
        }

到这，找到了我们执行网络请求的开始。

针对每一个我们调用的方法，生成一个ServiceMethod对象，再根据serviewMethod生成一个OkHttpCall对象，OkHttpCall对象再去引用okhttp3.Call去发起请求

看下这个过程。

生成ServiceMethod对象

// this:retrofit对象
new ServiceMethod.Builder<>(this, method).build()

重点是build()方法
1、callAdapter: 由方法返回值和注解从我们添加的addCallAdapterFactory(...)的工厂类中获取对应的CallAdapter
2、responseType：由callAdapter获取到接口返回的类型
// 注释1，后续在讲到得到http的Response之后如何转换成我们需要的数据类型时，会详细讲到
3、responseConverter: 我们添加addConverterFactory的工厂类中得出负责接口数据转换
4、处理注解
...

接着执行

serviceMethod.adapt(OkHttpCall)

 T adapt(Call<R> call) {
    return callAdapter.adapt(call);
  }

从上面可知callAdapter是我们add的适配器，我们按照RxJava2CallAdapterFactory对应的RxJava2CallAdapter来分析adapt方法

// Call<R>是OkHttpCall针对我们请求方法的实例化对象
@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;
    }
    ...
    // 注释4， 为什么返回的Observable不是responseObservable呢？
    return observable
 }

我们来分析CallEnqueueObservable异步请求是如何实现的？

CallEnqueueObservable 和 CallExecuteObservable都继承了Observable，所以我们在订阅了之后会执行subscribeActual方法。

@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();
    CallCallback<T> callback = new CallCallback<>(call, observer);
    observer.onSubscribe(callback);
    call.enqueue(callback);
  }

call对象是OkHttpCall的实例

@Override public void enqueue(final Callback<T> callback) {
     // 最终使用的Call对象是okhttp的
     okhttp3.Call call;
     ... //创建call对象
     // 执行okhttp的方法
     call.enqueue(new okhttp3.Callback() {
      @Override public void onResponse(okhttp3.Call call, okhttp3.Response rawResponse)       {
        Response<T> response;
        try {
          // 注释2，后续在讲到得到http的Response之后如何转换成我们需要的数据类型时，会详细讲到
          response = parseResponse(rawResponse);
        } catch (Throwable e) {
          callFailure(e);
          return;
        }

        try {
           // 回调到OkHttpCall中enqueue传入的callback中，即我们分析的CallEnqueueObservable的 CallCallback中
          callback.onResponse(OkHttpCall.this, response);
        } catch (Throwable t) {
          t.printStackTrace();
        }
      }

      @Override public void onFailure(okhttp3.Call call, IOException e) {
        callFailure(e);
      }

      private void callFailure(Throwable e) {
        try {
          callback.onFailure(OkHttpCall.this, e);
        } catch (Throwable t) {
          t.printStackTrace();
        }
      }
    });
}

CallEnqueueObservable内部类CallCallback的onResponse方法

// 注释3， 为什么得到的数据类型是Response类型，而不是我们自定义的返回类型（泛型）
@Override public void onResponse(Call<T> call, Response<T> response) {
      if (disposed) return;

      try {
        // 触发观察者的onNext方法
        observer.onNext(response);
        if (!disposed) {
          terminated = true;
           // 触发观察者的onComplete方法
          observer.onComplete();
        }
      } catch (Throwable t) {
        if (terminated) {
          RxJavaPlugins.onError(t);
        } else if (!disposed) {
          try {
            observer.onError(t);
          } catch (Throwable inner) {
            Exceptions.throwIfFatal(inner);
            RxJavaPlugins.onError(new CompositeException(t, inner));
          }
        }
      }
    }

触发观察者的相关操作，之后的处理，我们就可以自己封装我们自己业务需要的Observer了

到这，我们走完了addCallAdapterFactory方法。

那我们怎样得到我们所需的数据类型呢？

回到我们在生成ServiceMethod时候提到的注释1
具体如何生成responseConverter并做了什么操作呢

private Converter<ResponseBody, T> createResponseConverter() {
      Annotation[] annotations = method.getAnnotations();
      try {
        // responseType是前面我们得到返回值的类型
        return retrofit.responseBodyConverter(responseType, annotations);
      } catch (RuntimeException e) { // Wide exception range because factories are user code.
        throw methodError(e, "Unable to create converter for %s", responseType);
      }
    }

 public <T> Converter<ResponseBody, T> responseBodyConverter(Type type, Annotation[] annotations) {
    return nextResponseBodyConverter(null, type, annotations);
  }

最终调用

// type 为我们的返回值类型
public <T> Converter<ResponseBody, T> nextResponseBodyConverter(
      @Nullable Converter.Factory skipPast, Type type, Annotation[] annotations) {
    checkNotNull(type, "type == null");
    checkNotNull(annotations, "annotations == null");
    
    int start = converterFactories.indexOf(skipPast) + 1;
    // 根据我们addConvertFactory()方法添加的converterFactories
    for (int i = start, count = converterFactories.size(); i < count; i++) {
      Converter<ResponseBody, ?> converter =
          converterFactories.get(i).responseBodyConverter(type, annotations, this);
      if (converter != null) {
        //noinspection unchecked
        return (Converter<ResponseBody, T>) converter;
      }
    }
}

以我们经常用到的GsonConverterFactory为例

@Override
  public Converter<ResponseBody, ?> responseBodyConverter(Type type, Annotation[] annotations,
      Retrofit retrofit) {
      // 根据我们接口请求返回值类型，生成指定的TypeAdapter类型
    TypeAdapter<?> adapter = gson.getAdapter(TypeToken.get(type));
    return new GsonResponseBodyConverter<>(gson, adapter);
  }

至此，我们得到的responseConverter是包含了我们请求返回值类型的转换器

接着我们的注释2 parseResponse方法，如何实现格式的转换

Response<T> parseResponse(okhttp3.Response rawResponse) throws IOException {
    ResponseBody rawBody = rawResponse.body();
    ...
    // 调用serviceMethod实现转换
    T body = serviceMethod.toResponse(catchingBody);
    // 封装成Response对象传递给下一级
    return Response.success(body, rawResponse);   
}

 R toResponse(ResponseBody body) throws IOException {
    // responseConverter实际上是GsonResponseBodyConverter实例对象
    return responseConverter.convert(body);
  }

//简单的gson格式转换
@Override public T convert(ResponseBody value) throws IOException {
    JsonReader jsonReader = gson.newJsonReader(value.charStream());
    try {
      T result = adapter.read(jsonReader);
      if (jsonReader.peek() != JsonToken.END_DOCUMENT) {
        throw new JsonIOException("JSON document was not fully consumed.");
      }
      return result;
    } finally {
      value.close();
    }
  }

得到数据类型之后，再封装成Response<T>类型，回调到OkhttpCall中的CallCallback中的onResponse方法中。

当然到这还没有结束。还有注释3和注释4的疑问
这两个疑问是相关联的。

@Override public Object adapt(Call<R> call) {
    ...
     Observable<?> observable;
    if (isResult) {
      observable = new ResultObservable<>(responseObservable);
    } else if (isBody) {
      observable = new BodyObservable<>(responseObservable);
    } else {
      observable = responseObservable;
    }
    ...
}

我们以ResultObservable为例来分析
将真正请求网络的responseObservable赋值给ResultObservable，然后让用户去观察ResultObservable，触发ResultObservable的subscribeActual方法

ResultObservable(Observable<Response<T>> upstream) {
    this.upstream = upstream;
  }

@Override protected void subscribeActual(Observer<? super Result<T>> observer) {
    // 触发真正网络请求的Observable, 请求结果回掉给ResultObserver
    // 但是参数observer才是我们用户真正的observer
    upstream.subscribe(new ResultObserver<T>(observer));
  }

private static class ResultObserver<R> implements Observer<Response<R>> {
    // this.observer 是我们用户真正的observer
    ResultObserver(Observer<? super Result<R>> observer) {
      this.observer = observer;
    }
    // 真正网络请求的Observable触发onNext, 回掉到这
    @Override public void onNext(Response<R> response) {
       // 返回到用户subscribe中的onNext方法中
      observer.onNext(Result.response(response));
    }
}

至此，整个过程分析完毕。

我还有一个疑问的是，Result.response(response)返回的结果是Result<T>类型，但是我们在onNext里面拿到的确是T

Retrofit+RxJava请求一次接口全过程分析

推荐阅读更多精彩内容