Retrofit源码分析--Retrofit动态代理与CallAdapter的实现_retrofit call flowable区别-程序员宅基地

技术标签: android  

目前打算选用Retrofit2+RxJava2作为网络请求框架,下面从源码角度分析下这两个框架是怎么粘合在一起的。

看完本文,你将会看到:
- 动态代理模式在Retrofit中的运用。
- 抽象工厂模式在Retrofit中的运用。
- RxJava2如何和Retrofit结合实现网络请求模块。

使用Retrofit实现网络请求

定义一个Service接口

//https://api.douban.com/v2/book/1220562
public interface IDoubanService {
    

        @GET("/v2/{book}/{bookId}")
        Call<Book> getBookInfo(@Path("book") String book, @Path("bookId") String bookId);

        @GET("/v2/{book}/{bookId}")
        Observable<Book> getBookInfoObservable(@Path("book") String book, @Path("bookId") String bookId);
}

接口中的方法就是我们实际要执行的网络请求,方法的返回值就是CallAdapter的返回值,如果没有的话,默认返回Call对象。

如果我们自定义了一个返回类型,并且没有加上对应的CallAdapterFactory的话,那么抛出异常,如下:

  Caused by: java.lang.IllegalArgumentException: Unable to create call adapter for class java.lang.String

使用Retrofit创建接口的动态代理

final String bookId = "1220562";
Retrofit retrofit = new Retrofit.Builder()
        .baseUrl("https://api.douban.com/")
        // 将Response通过GSON转成POJO
        .addConverterFactory(GsonConverterFactory.create())
        // 将接口方法根据类型返回
        .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
        .client(new OkHttpClient())
        .build();
// 使用Retrofit创建接口的代理对象
IDoubanService service = retrofit.create(IDoubanService.class);
Call<Book> call = service.getBookInfo("book", bookId);
// 使用Call对象获取网络请求结果
Response<Book> response = call.execute();
Book book = response.body();

上面的示例中使用的Call进行获取的网络请求,实际上我们可以通过CallAdapterFactory这种抽象工厂来创建我们自己的返回类型,稍后详细分析。

Retrofit中的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);
            }
            // Platform这个类用来区分平台,如果是Java8的话,接口中会有default方法
            if (platform.isDefaultMethod(method)) {
              return platform.invokeDefaultMethod(method, service, proxy, args);
            }
            // 将这个反射的方法封装成ServiceMethod
            // ServiceMethod会将接口中的注解进行解析
            ServiceMethod<Object, Object> serviceMethod =
                (ServiceMethod<Object, Object>) loadServiceMethod(method);
            // 创建真实的Call对象    
            OkHttpCall<Object> okHttpCall = new OkHttpCall<>(serviceMethod, args);

            return serviceMethod.callAdapter.adapt(okHttpCall);
          }
        });
  }

ServiceMethod中的createCallAdapter()用于获取CallAdapter

private CallAdapter<T, R> createCallAdapter() {
      // 获取接口方法的返回值
      Type returnType = method.getGenericReturnType();

      // 检查返回类型
      if (Utils.hasUnresolvableType(returnType)) {
        throw methodError(
            "Method return type must not include a type variable or wildcard: %s", returnType);
      }
      // 返回值不能是void
      if (returnType == void.class) {
        throw methodError("Service methods cannot return void.");
      }
      Annotation[] annotations = method.getAnnotations();
      try {
        //noinspection unchecked
        // 最终会从Retrofit的CallAdapter中进行查找,还记得之前的addCallAdapterFactroy()抽象工厂么?
        return (CallAdapter<T, R>) retrofit.callAdapter(returnType, annotations);
      } catch (RuntimeException e) { // Wide exception range because factories are user code.
        throw methodError(e, "Unable to create call adapter for %s", returnType);
      }
    }

最终会调到Retrofit#nextCallAdapter(),在这个方法中,会进行遍历,将Annotion和returnType传入,直到有一个Factroy能够返回非空的CallAdapter

 /**
     * Returns a call adapter for interface methods that return {@code returnType}, or null if it
     * cannot be handled by this factory.
     */
     // 这个方法需要返回非空的CallAdapter
    public abstract @Nullable CallAdapter<?, ?> get(Type returnType, Annotation[] annotations,
        Retrofit retrofit);

RxJava2ConvertFactory的实现

核心的get()


  @Override
  public CallAdapter<?, ?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
    Class<?> rawType = getRawType(returnType);

    if (rawType == Completable.class) {
      // Completable is not parameterized (which is what the rest of this method deals with) so it
      // can only be created with a single configuration.
      return new RxJava2CallAdapter(Void.class, scheduler, isAsync, false, true, false, false,
          false, true);
    }

    boolean isFlowable = rawType == Flowable.class;
    boolean isSingle = rawType == Single.class;
    boolean isMaybe = rawType == Maybe.class;
    if (rawType != Observable.class && !isFlowable && !isSingle && !isMaybe) {
      return null;
    }

    boolean isResult = false;
    boolean isBody = false;
    Type responseType;
    if (!(returnType instanceof ParameterizedType)) {
      String name = isFlowable ? "Flowable"
          : isSingle ? "Single"
          : isMaybe ? "Maybe" : "Observable";
      throw new IllegalStateException(name + " return type must be parameterized"
          + " as " + name + "<Foo> or " + name + "<? extends Foo>");
    }

    Type observableType = getParameterUpperBound(0, (ParameterizedType) returnType);
    Class<?> rawObservableType = getRawType(observableType);
    if (rawObservableType == Response.class) {
      if (!(observableType instanceof ParameterizedType)) {
        throw new IllegalStateException("Response must be parameterized"
            + " as Response<Foo> or Response<? extends Foo>");
      }
      responseType = getParameterUpperBound(0, (ParameterizedType) observableType);
    } else if (rawObservableType == Result.class) {
     
      if (!(observableType instanceof ParameterizedType)) {
        throw new IllegalStateException("Result must be parameterized"
            + " as Result<Foo> or Result<? extends Foo>");
      }
      responseType = getParameterUpperBound(0, (ParameterizedType) observableType);
      isResult = true;
    } else {
      responseType = observableType;
      isBody = true;
    }
    // 一般情况返回一个 RxJava2CallAdapter
    return new RxJava2CallAdapter(responseType, scheduler, isAsync, isResult, isBody, isFlowable,
        isSingle, isMaybe, false);
  }

RxJava2CallAdapter中的adaper()

CallAdapter中有一个核心方法是adapt(),作用是将Call对象转化成任意的类型


  /**
   * Returns an instance of {
   @code T} which delegates to {
   @code call}.
   * <p>
   * For example, given an instance for a hypothetical utility, {
   @code Async}, this instance would
   * return a new {
   @code Async<R>} which invoked {
   @code call} when run.
   * <pre><code>
   * &#64;Override
   * public &lt;R&gt; Async&lt;R&gt; adapt(final Call&lt;R&gt; call) {
   *   return Async.create(new Callable&lt;Response&lt;R&gt;&gt;() {
   *     &#64;Override
   *     public Response&lt;R&gt; call() throws Exception {
   *       return call.execute();
   *     }
   *   });
   * }
   * </code></pre>
   */
  T adapt(Call<R> call);

RxJava2CallAdapter中的实现

 // 此处就将Call对象转换成Observable了
  @Override public Object adapt(Call<R> call) {
  // 先创建一个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 observable;
  }

RxJava进行网络请求

RxJava接受网络请求的过程如下

observable
        // 网络请求在IO线程执行
        .subscribeOn(Schedulers.io())
        // 回调到主线程
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe((book) -> {
            mTvText.setText(String.valueOf(book));
        });

Retrofit+Factory将网络请求抽象成了一个Observable对象,那么我们可以通过RxJava这样的方法操作网络请求了。
Retrofit返回Observable的实现是CallExecuteObservable或者CallEnqueueObservable,我们看下这个Observable的实现。


final class CallExecuteObservable<T> extends Observable<Response<T>> {
    
  private final Call<T> originalCall;

  CallExecuteObservable(Call<T> originalCall) {
    this.originalCall = originalCall;
  }

  // 如果在Observeable中进行订阅操作,那么实际会调用到这个方法
  @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 {
    // 会调用Call执行网络请求,获取Response对象
      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));
        }
      }
    }
  }

  private static final class CallDisposable implements Disposable {
    
    private final Call<?> call;

    CallDisposable(Call<?> call) {
      this.call = call;
    }

    @Override public void dispose() {
      call.cancel();
    }

    @Override public boolean isDisposed() {
      return call.isCanceled();
    }
  }
}

整体流程

版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/y874961524/article/details/80891146

智能推荐

LightOJ - 1067 Combinations(快速幂+逆元)_快速幂求逆元 在线oj-程序员宅基地

文章浏览阅读484次。DescriptionGiven n differentobjects, you want to take k of them. How many ways to can do it? For example, saythere are 4 items; you want to take 2 of them. So, you can do it 6 ways. Take 1, _快速幂求逆元 在线oj

Fairseq学习日记:注定麻烦的旅程_final_lr_scale-程序员宅基地

文章浏览阅读6.6k次,点赞10次,收藏23次。现在开头:Fairseq是一个正在快速迭代的产品,而且是开源的!这不是表扬,这意味着三件事情:1.他没有文档!所有框架代码都没有任何注释,包括函数docstring都没有2.他没有经过有效测试,估计是抢时间吧!即使是官网Readme里的例子也是无法跑起来的!3.他是一个框架,而且是一个非常不Pythonic的框架,充斥着inline/包装器/莫名其妙的语法。虽然这三点决定他真的对不住Facebook的金字招牌,但是作为一个学习者,总要把他运行起来,那么开始这场针对 FaceBOOK派“全_final_lr_scale

[Linux][Busybox]分享你不知道Top 命令参数_busybox top-程序员宅基地

文章浏览阅读5.1k次。目录摘要:基本操作与命令介绍:进入top后交互一点点新的操作Author: Keivn.Xu [email protected]摘要: 玩过Linux一定使用过busybox top命令,但下面的操作方法,你不一定有见过。基本操作与命令介绍:console:/ $ busybox top -help top: invalid optio..._busybox top

rssi参数获取_信号强度(RSSI)知识整理-程序员宅基地

文章浏览阅读1.6k次。为什么无线信号(RSSI)是负值答:其实归根到底为什么接收的无线信号是负值,这样子是不是容易理解多了。因为无线信号多为mW级别,所以对它进行了极化,转化为dBm而已,不表示信号是负的。1mW就是0dBm,小于1mW就是负数的dBm数。弄清信号强度的定义就行了:RSSI(接收信号强度)Received Signal Strength IndicatorRss=10logP,只需将接受到的信号功率P代..._c#获取低功耗设备的rssi信号强度

后端服务的雪崩效应及解决思路_接口超时时间过长导致雪崩效应-程序员宅基地

文章浏览阅读204次。1.RPC与本地调用的区别RPC远程调用,一般是跨平台、采用http协议,因为http协议底层使用socket技术,只要你的语言支持socket技术,就可以相互进行通讯。比如:java语言开发的接口,使用http协议,如此以来C#语言可以调用。本地调用:只支持java语言与java语言开发,使用虚拟机和虚拟机之间的通讯,RMI。2.雪崩效应产生的原因默认情况下只有一个线程池维护所有的服务接口,如果大量的请求访问同一个接口,达到tomcat线程池默认极限,可能会导致其他服务无法访问。3.雪_接口超时时间过长导致雪崩效应

linux操作redis_linux 连接redis-程序员宅基地

文章浏览阅读2.7w次,点赞4次,收藏35次。redis常用命令_linux 连接redis

随便推点

keil迁移到cubeIDE_cubeide生成代码 keil打开-程序员宅基地

文章浏览阅读2.7k次。以stm32f105rct6为例子,keil迁移到cubeIDE_cubeide生成代码 keil打开

语音处理:Python实现dBFS刻度和采样值相互转换_dbfs 频域-程序员宅基地

文章浏览阅读781次。以对数域常用的dBFS刻度为例,支持主流音频信号位深:整型16/24/32位和浮点32位,编写Python实现对数域和采样值单位互换功能_dbfs 频域

SOD(显著性目标检测)数据集_sod数据集-程序员宅基地

文章浏览阅读4k次,点赞5次,收藏29次。显著性目标检测常用十种数据集:SOD,提取码:f7uqDUT-OMRON,提取码:wqpnMSRA-B,提取码:rfrbSOC,提取码:d5b9SED2,提取码:q4iaHKU-IS,提取码:2f1iPASCAL-S,提取码:naaxDUTS,提取码:a5w7THUR-15K,提取码:ptk9ECSSD..._sod数据集

线程基础:多任务处理(18)——MESI协议以及带来的问题:伪共享-程序员宅基地

文章浏览阅读3.9k次,点赞10次,收藏17次。本文和后续文章将着眼CPU的工作原理阐述伪共享的解决方法和volatile关键字的应用。

Google Earth Engine (GEE) ——代码编辑器_gee怎么新建代码文件-程序员宅基地

文章浏览阅读1.7k次。问题 在线代码编辑器的主要功能是什么? 在学习 GEE 的过程中,我可以去哪里寻求帮助? 如何搜索和导入数据集? 如何创建、共享和保存脚本? 目标 了解代码编辑器中可用的工具 加载图像集合并将其过滤为相关图像 使用几何工具创建研究区域 代码编辑器概述GEE 有一个称为代码编辑器的集成开发环境 (IDE)。代码编辑器有许多功能可以帮助我们在本教程中更轻松地在这种环境中进行编程。有关详尽说明,请参阅GEE 用户指南..._gee怎么新建代码文件

关于 Eclipse 使用 Maven 打包,每次都需要下载 jar 包的问题_maven打包每次都要下载依赖-程序员宅基地

文章浏览阅读6.2k次。在是Eclipse开发的时候,使用Maven打包每次都需要联网下载jar包。第一次需要下载这个可以理解。但是每次都需要下载,就有点问题了...重点是,所用的网络不能访问Maven的私服,所以每次打包都断开网络。这样很麻烦,找资料发现勾选下图中的 Offline(离线) 就可以解决问题了。注意:当你需要下载其他的依赖时,就需要把这个勾去掉,不然连接不上仓库哦。..._maven打包每次都要下载依赖