我们先从Retrofit的普通调用看起


BaseNetEngine.getInstance().getsApiService(ApiInterface.class) //1
                .mPutCommunityBind(thirdID) //2
                .compose(bindToLifecycle()) //3
                .compose(RxSchedulersHelper.io_main()) //4
                .subscribe(deviceBindBean -> {  //5   
                		网络调用返回......
                }, throwable -> {
                		......   
                });

上面是普通的调用，我们一行行的来分析.

开头注释的第一行代码分析

ApiInterface.class 这个是一个接口，里面定义了我们请求的方法，如下


public interface ApiInterface {
		
	@PUT(URLManager.PUT_COMMUNITY_BIND+"{id}")
   	Flowable<DeviceBindBean> mPutCommunityBind(@Path("id") String id);
}

上面定义了一个接口方法。

BaseNetEngine.getInstance() 是一个单例类，然后 getsApiService方法如下


public synchronized <T> T getsApiService(Class<T> clazz) {
   	T obj = (T) hashMap.get(clazz.getName());
   		if (obj == null) {
   			obj = getRetrofit().create(clazz);
   			hashMap.put(clazz.getName(), obj);
   		}	
   	return obj;
}

接着看一下getRetrofit()方法


public Retrofit getRetrofit() {
   if (sOkHttpClient == null) {
     initOkHttp();
   }
   if (sRetrofit == null) {
     ExtensionRegistry registry = ExtensionRegistry.newInstance();
     sRetrofit = new Retrofit.Builder()
         .baseUrl(BuildConfig.urlMAPI)
         .client(sOkHttpClient)
         .addConverterFactory(new NullOnEmptyConverterFactory())
         .addConverterFactory(ProtoConverterFactory.createWithRegistry(registry))//一定要在gsonconvert
         .addConverterFactory(GsonConverterFactory.create())
         .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
         .build();
   }
   return sRetrofit;
 }

该方法，先是初始化Okhttp, 然后就是addConverterFactory，进行相关的配置。

接着看一下create()方法


public <T> T create(final Class<T> service) {
  Utils.validateServiceInterface(service);
  if (validateEagerly) {
    eagerlyValidateMethods(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 {
          // If the method is a method from Object then defer to normal invocation.
          if (method.getDeclaringClass() == Object.class) {
            return method.invoke(this, args);
          }
          if (platform.isDefaultMethod(method)) {
            return platform.invokeDefaultMethod(method, service, proxy, args);
          }
          ServiceMethod<Object, Object> serviceMethod =
              (ServiceMethod<Object, Object>) loadServiceMethod(method);
          OkHttpCall<Object> okHttpCall = new OkHttpCall<>(serviceMethod, args);
          return serviceMethod.adapt(okHttpCall);
        }
      });
}

create()方法返回的是一个代理类，代理对象会调用ApiInterface.class 这个接口中的方法，当我们调用该接口里面的方法时，其实调用的是上述InvocationHandler()里面的invoke()方法。

开头注释的第二行代码分析

第二行通过代理对象调用了mPutCommunityBind(thirdID)方法，这个方法是ApiInterface.class里面的方法，所以会调用InvocationHandler里面的invoke方法，所以重点来了,会调用下面的方法，然后返回的是一个observable对象


ServiceMethod<Object, Object> serviceMethod =
                (ServiceMethod<Object, Object>) loadServiceMethod(method);
OkHttpCall<Object> okHttpCall = new OkHttpCall<>(serviceMethod, args);
return serviceMethod.adapt(okHttpCall);

首先看下 ServiceMethod，这里会创建ServiceMethod对象


ServiceMethod<?, ?> loadServiceMethod(Method method) {
   ServiceMethod<?, ?> result = serviceMethodCache.get(method);
   if (result != null) return result;
   synchronized (serviceMethodCache) {
     result = serviceMethodCache.get(method);
     if (result == null) {
       result = new ServiceMethod.Builder<>(this, method).build();
       serviceMethodCache.put(method, result);
     }
   }
   return result;
 }


 public ServiceMethod build() {
 	   //这里的callAdapter就是之前配置的RxJava2CallAdapterFactory.create()
      callAdapter = createCallAdapter();      
      
      responseType = callAdapter.responseType();
  	
  	   ......	
  	   //这里的responseConver也是之前配置的GsonConverterFactory.create()
      responseConverter = createResponseConverter();
      for (Annotation annotation : methodAnnotations) {
        parseMethodAnnotation(annotation);
      }
	......
	......	
	return new ServiceMethod<>(this);
}

ServiceMethod 主要是处理之前Retrofit的配置，其内部解析相关注解。

然后接着看OkHttpCall方法。这里面封装了okhttp的方法，比如request,response方法

然后接着看serviceMethod.adapt(okHttpCall); 这里的adapt方法是callAdapter.adapt(call); 这里的callAdapter是RxJava2CallAdapter。所以我们进入到RxJava2CallAdapter的adapt方法看一看，如下


@Override public Object adapt(Call<R> call) {
   
   //这里主要根据isAsync创建同步或者异步的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);
   }
  
   ......
   
   return observable;
 }

所以调用第二行代码返回CallEnqueueObservable，或者CallExecuteObservable 默认是返回CallExecuteObservable进行同步的请求

开头注释的第五行代码分析

第三行，第四行主要是Rxjava相关的方法，第三行是绑定Rx生命周期，第四行是Rx的线程切换，所以我们主要看第五行。第五行主要是方法请求，这里面会进行网络调用。

我们进入subscribe源码看一下


public final void subscribe(FlowableSubscriber<? super T> s) {
    ObjectHelper.requireNonNull(s, "s is null");
    try {
        ......	
        subscribeActual(z);
    } catch (NullPointerException e) { // NOPMD
        throw e;
    } catch (Throwable e) {
     ......
    }
}

1 2	这是Flowable里面的抽象方法 protected abstract void subscribeActual(Subscriber<? super T> s);

我们知道第二步返回的对象是CallExecuteObservable，那么看下该对象下的subscribeActual方法，，这里的网络请求是同步的方法


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();
    observer.onSubscribe(new CallDisposable(call));
    boolean terminated = false;
    try {
      //这里进行真正的网络请求
      Response<T> response = call.execute();
      if (!call.isCanceled()) {
        observer.onNext(response);
      }
      if (!call.isCanceled()) {
        terminated = true;
        observer.onComplete();
      }
    } catch (Throwable t) {
      Exceptions.throwIfFatal(t);
      if (terminated) {
        RxJavaPlugins.onError(t);
      } else if (!call.isCanceled()) {
        try {
          observer.onError(t);
        } catch (Throwable inner) {
          Exceptions.throwIfFatal(inner);
          RxJavaPlugins.onError(new CompositeException(t, inner));
        }
      }
    }
  }
}

所以当调用第五行的时候，会进行真正的网络请求

总结

流程图