Retrofit-2.4.0结合Rxjava2.3.0源码分析

2019-03-27 本文已影响0人森林苍穹

我们先从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) {
         ......
        }
    }

这是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));
        }
      }
    }
  }

}

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

插曲：我们在进行网络请求的时候，可能会出现android retrofit End of input at line 1 column 1 path这个问题。
这是因为通过GsonConverterFactory将服务端相应内容解析成对应的实体类，在接口正常响应时（有数据返回），并没有什么异常发生，但当接口请求的数据为空，我们的服务端人员并不是返回理论意义上的空，null或者[]（数据集合空），而是返回没有响应体body，只有响应头header，content-length为0的Response
这时候GsonConverterFactory就解析异常了。
解决的办法是自定义一个ConverterFactory，如下：

public class NullOnEmptyConverterFactory extends Converter.Factory {
 
    @Override
    public Converter<ResponseBody, ?> responseBodyConverter(Type type, Annotation[] annotations, Retrofit retrofit) {
        final Converter<ResponseBody, ?> delegate = retrofit.nextResponseBodyConverter(this, type, annotations);
        return new Converter<ResponseBody,Object>() {
            @Override
            public Object convert(ResponseBody body) throws IOException {
                if (body.contentLength() == 0) return null;
                return delegate.convert(body);
            }
        };
    }
}

总结

流程图

Retrofit-2.4.0结合Rxjava2.3.0源码分析

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

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

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

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

总结

猜你喜欢

热点阅读