retrofit+okhttp源码流程

Posted 2021-07-04 BigSweetee

初始化如下

Retrofit.Builder rBuilder = new Retrofit.Builder().client(okHttpClientBuilder.build())
                .addConverterFactory(GsonConverterFactory.create()) 
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .baseUrl(EnvConfig.getPadBaseUrl())
                .client(okHttpClientBuilder.build());
                
 rBuilder .create(serviceClass);

在retrofit创建的时候，会生成一个动态代理，每次的接口请求都会在这里触发

 public <T> T create(final Class<T> service) {
    validateServiceInterface(service);
    return (T) Proxy.newProxyInstance(service.getClassLoader(), new Class<?>[] { service },
        new InvocationHandler() {
           return loadServiceMethod(method).invoke(args != null ? args : emptyArgs);
  }

主要流程分析
loadServiceMethod

ServiceMethod<?> loadServiceMethod(Method method) {
   //省略缓存逻辑。。
    result = ServiceMethod.parseAnnotations(this, method);
    return result;
  }
  
static <T> ServiceMethod<T> parseAnnotations(Retrofit retrofit, Method method) {
    RequestFactory requestFactory = RequestFactory.parseAnnotations(retrofit, method);
    return HttpServiceMethod.parseAnnotations(retrofit, method, requestFactory);
  }

parseAnnotations这个方法主要是解析了注解

static <ResponseT, ReturnT> HttpServiceMethod<ResponseT, ReturnT> parseAnnotations(
//第一步
Annotation[] annotations = method.getAnnotations();
//第二步这里的calladapter是前面设置的RxJava2CallAdapterFactory
 CallAdapter<ResponseT, ReturnT> callAdapter =
        createCallAdapter(retrofit, method, adapterType, annotations);
//第三步
 Converter<ResponseBody, ResponseT> responseConverter =
        createResponseConverter(retrofit, method, responseType);
//最后创建了一个calladapter把上面的三个参数传了进去
new CallAdapted<>(requestFactory, callFactory, responseConverter, callAdapter);

}

loadServiceMethod主要功能:
1,解析请求的注解
2,生成calladapter，responseConverter
3,创建CallAdapted

接下来看loadServiceMethod(method).invoke(args != null ? args : emptyArgs);
loadserviceMethod返回的是一个CallAdapted

CallAdapted的代码如下

static final class CallAdapted<ResponseT, ReturnT> extends HttpServiceMethod<ResponseT, ReturnT> {
    private final CallAdapter<ResponseT, ReturnT> callAdapter;

    CallAdapted(RequestFactory requestFactory, okhttp3.Call.Factory callFactory,
        Converter<ResponseBody, ResponseT> responseConverter,
        CallAdapter<ResponseT, ReturnT> callAdapter) {
      super(requestFactory, callFactory, responseConverter);
      this.callAdapter = callAdapter;
    }

    @Override protected ReturnT adapt(Call<ResponseT> call, Object[] args) {
      return callAdapter.adapt(call);
    }
  }

然后触发的是loadServiceMethod(method).invoke(
invoke方法,invoke方法在父类里面

calladapter的父类为HttpServiceMethod

HttpServiceMethod(RequestFactory requestFactory, okhttp3.Call.Factory callFactory,
      Converter<ResponseBody, ResponseT> responseConverter) {
    this.requestFactory = requestFactory;
    this.callFactory = callFactory;
    this.responseConverter = responseConverter;
  }

  @Override final @Nullable ReturnT invoke(Object[] args) {
    Call<ResponseT> call = new OkHttpCall<>(requestFactory, args, callFactory, responseConverter);
    return adapt(call, args);
  }

创建了一个okhttpcall，调用了子类的adapt方法
子类的adapt方法触发了callAdapter.adapt(call);
而根据前面的分析得出callAdapter是rxjavaadapter，所以直接去看rxjavaadapter的adapt方法

  @Override protected ReturnT adapt(Call<ResponseT> call, Object[] args) {
      return callAdapter.adapt(call);
    }

 @Override public Object adapt(Call<R> call) {
 //注意这里有同步和异步俩个Observable
 创建好Observable之后
    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 RxJavaPlugins.onAssembly(observable);
  }

这里封装成了Observable
最后在进行网络请求subscribe的时候

viewModel.getCycleHistory(map).subscribe(object : ApiObserver<GeneralResponse<List<UserCourseBean>>>() {
            override fun onSuccess(resp: GeneralResponse<List<UserCourseBean>>) {

            }
        })

就会触发CallEnqueueObservable的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);
    if (!callback.isDisposed()) {
      call.enqueue(callback);
    }
  }

最后触发了call的enqueue方法。
前面new出来的是一个new OkHttpCall
看下OkHttpCall的enqueue方法

call.enqueue(new okhttp3.Callback() {
      @Override public void onResponse(okhttp3.Call call, okhttp3.Response rawResponse) {
        Response<T> response;
        try {
          response = parseResponse(rawResponse);
        } catch (Throwable e) {
          throwIfFatal(e);
          callFailure(e);
          return;
        }

        try {
          callback.onResponse(OkHttpCall.this, response);
        } catch (Throwable t) {
          throwIfFatal(t);
          t.printStackTrace(); // TODO this is not great
        }
      }

      @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) {
          throwIfFatal(t);
          t.printStackTrace(); // TODO this is not great
        }
      }
    });

通过okhttp发送了请求，并且在请求有响应的时候
response = parseResponse(rawResponse);

Response<T> parseResponse(okhttp3.Response rawResponse) throws IOException {
    ResponseBody rawBody = rawResponse.body();
      T body = responseConverter.convert(catchingBody);
      return Response.success(body, rawResponse);

初始化的时候设置的responseConverter是GsonConverterFactory
所以调用了GsonConverterFactory的convert

@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();
    }
  }

在数据gson转换完成之后调用
callback.onResponse(OkHttpCall.this, response);
callback在subscribeActual中进行了初始化
CallCallback callback = new CallCallback<>(call, observer);
最后看下callback的回调onResponse

@Override public void onResponse(Call<T> call, Response<T> response) {
   if (disposed) return;

   try {
     observer.onNext(response);

     if (!disposed) {
       terminated = true;
       observer.onComplete();
     }
   } catch (Throwable t) {
     Exceptions.throwIfFatal(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.onNext(response);
流程结束