Android中的网络库详解(四. Retrofit)

1.前言

经过前面几章的学习,对于andorid中的网络库已经有了充分的了解,但是技术一直是向前发展的,所以我们每时每刻也都得超越自己

本章网络库系列的最后一章,讲解的是又一android端最为流行的网络库框架Retrofit,他同样也是由square出品,他和Volley一样是一个网络封装库,底层是由OkHttp完成的,通过注解实现RESTful式的Http网络请求

2.目录

目录

3.Retrofit

3.1.Retrofit的优点

• 1.耦合性极低
• 2.可以自由配置底层网络请求框架
• 3.支持同步异步,并且可结合RxJava使用
• 4.可以配置不同的交互数据解析工具来解析数据
• 5.使用灵活方便

3.2.Retrofit的简单使用

• 1.申明API接口

private interface RetrofitService {
    //测试调用(wanandroid主页)
    @GET("article/list/{page}/json")
    Call<ResponseBody> getArticle(@Path("page") int page);
}

• 2.构建Retrofit对象

Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("https://www.wanandroid.com/")
                .build();

• 3.获取API接口,直接调用回去实体回调的Call

RetrofitService service = retrofit.create(RetrofitService.class);
Call<ResponseBody> article = service.getArticle(1);

• 4.执行Call获取响应

// 同步调用 什么线程执行 回调就在什么线程
Response<ResponseBody> response = article.execute();
String json = response.body().string();
// 异步调用 不同于OkHttp会回调到主线程
Call<ResponseBody> article = service.getArticle(1);
article.enqueue(new Callback<ResponseBody>() {
    @Override
    public void onResponse(Call<ResponseBody> call, Response<ResponseBody> response) {
        // 有IOException
        String json = response.body().string();
    }
    @Override
    public void onFailure(Call<ResponseBody> call, Throwable t) {
    }
});

3.3.Retrofit源码分析

首先查看Retrofit的构建过程

public final class Retrofit {
  // 以Method-ServiceMethod映射缓存网络请求相关配置,防止并发问题使用的ConcurrentHashMap 
  private final Map<Method, ServiceMethod<?, ?>> serviceMethodCache = new ConcurrentHashMap<>();
  //网络请求工厂
  final okhttp3.Call.Factory callFactory;
  //基准Url
  final HttpUrl baseUrl;
  //数据转换器工厂集合
  final List<Converter.Factory> converterFactories;
  //网络请求适配器工厂集合
  final List<CallAdapter.Factory> callAdapterFactories;
  //回调方法执行器
  final @Nullable Executor callbackExecutor;
  //是否缓存创建的ServiceMethod
  final boolean validateEagerly;
    
   public Retrofit build() {
        //基准Url必须设定 且怒能为null
        if (baseUrl == null) {
            throw new IllegalStateException("Base URL required.");
        }
        //设置配置的网络请求工程
        okhttp3.Call.Factory callFactory = this.callFactory;
        if (callFactory == null) {
            //默认为OkHttpClient
            callFactory = new OkHttpClient();
        }
        //请求回调方法执行器
        Executor callbackExecutor = this.callbackExecutor;
        if (callbackExecutor == null) {
            //
            callbackExecutor = platform.defaultCallbackExecutor();
        }
        //添加配置的CallAdapter.Factory 再追加platform.defaultCallAdapterFactories
        // Make a defensive copy of the adapters and add the default Call adapter.
        List<CallAdapter.Factory> callAdapterFactories = new ArrayList<>(this.callAdapterFactories);
        callAdapterFactories.addAll(platform.defaultCallAdapterFactories(callbackExecutor));
        //添加BuiltInConverters
        //和配置的Conver.Factory
        //以及platform.defaultConverterFactories
        // Make a defensive copy of the converters.
        List<Converter.Factory> converterFactories = new ArrayList<>(
                1 + this.converterFactories.size() + platform.defaultConverterFactoriesSize());
        // Add the built-in converter factory first. This prevents overriding its behavior but also
        // ensures correct behavior when using converters that consume all types.
        converterFactories.add(new BuiltInConverters());
        converterFactories.addAll(this.converterFactories);
        converterFactories.addAll(platform.defaultConverterFactories());
        //构建Retrofit对象
        return new Retrofit(callFactory, baseUrl, unmodifiableList(converterFactories),
                unmodifiableList(callAdapterFactories), callbackExecutor, validateEagerly);
    }
}

然后再查看网络接口的构建过程
RetrofitService service = retrofit.create(RetrofitService.class)

public <T> T create(final Class<T> service) {
    //对service进行校验 必须是一个接口 且没有继承其它接口
    Utils.validateServiceInterface(service);
    //是否需要提前验证
    if (validateEagerly) {
        eagerlyValidateMethods(service);
    }
    //利用动态代理技术,自动生成Service接口的实现类,将Service接口中的参数交给InvocationHandler处理
    return (T) Proxy.newProxyInstance(service.getClassLoader(), new Class<?>[] { service },
            new InvocationHandler() {
                private final retrofit2.Platform platform = retrofit2.Platform.get();
                private final Object[] emptyArgs = new Object[0];

                @Override public @Nullable Object invoke(Object proxy, Method method,
                                                         @Nullable Object[] args) throws Throwable {
                    //Object类的方法 直接调用
                    // 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);
                    }
                    //否则就通过loadServiceMethod(method)方法获取对应的serviceMethod 并invoke
                    return loadServiceMethod(method).invoke(args != null ? args : emptyArgs);
                }
            });
}

这里最后就是用了动态代理生成Service接口的实现类,将参数交由InvocationHandler执行invoke()方法,再下来再看看loadServiceMethod(method)方法

ServiceMethod<?> loadServiceMethod(Method method) {
    //先从缓存中获取
    ServiceMethod<?> result = serviceMethodCache.get(method);
    //有则直接返回
    if (result != null) return result;
    synchronized (serviceMethodCache) {
        //再次获取 多线阻塞的情况下 可能已由其它线程创建完毕(双重校验)
        result = serviceMethodCache.get(method);
        if (result == null) {
            //解析注解 创建ServiceMethod
            result = ServiceMethod.parseAnnotations(this, method);
            //存入缓存
            serviceMethodCache.put(method, result);
        }
    }
    return result;
}

从缓存获取解析注解获取ServiceMethod,前面提前验证判断也会调用这个方法,然后再看整个解析过程ServiceMethod.parseAnnotations()

static <T> ServiceMethod<T> parseAnnotations(Retrofit retrofit, Method method) {
    //将解析转化为RequestFactory
    RequestFactory requestFactory = RequestFactory.parseAnnotations(retrofit, method);
    //获取返回类型
    Type returnType = method.getGenericReturnType();
    //是不是无法解析的类型
    if (Utils.hasUnresolvableType(returnType)) {
        throw methodError(method,
                "Method return type must not include a type variable or wildcard: %s", returnType);
    }
    //返回值类型不能void
    if (returnType == void.class) {
        throw methodError(method, "Service methods cannot return void.");
    }
    //交由HttpServiceMethod解析
    return HttpServiceMethod.parseAnnotations(retrofit, method, requestFactory);
}

接着查看RequestFactory.parseAnnotations()

static RequestFactory parseAnnotations(Retrofit retrofit, Method method) {
    //建造者模式配置参数
    return new RequestFactory.Builder(retrofit, method).build();
}
Builder(Retrofit retrofit, Method method) {
    this.retrofit = retrofit;
    this.method = method;
    //获取网络请求方法里的注解
    this.methodAnnotations = method.getAnnotations();
    //获取参数类型
    this.parameterTypes = method.getGenericParameterTypes();
    //获取参数注解
    this.parameterAnnotationsArray = method.getParameterAnnotations();
}

RequestFactory build() {
    for (Annotation annotation : methodAnnotations) {
        //解析注解参数 异常判断
        parseMethodAnnotation(annotation);
    }
    //错误检测,最后会抛IllegalArgumentException
    ...
    int parameterCount = parameterAnnotationsArray.length;
    parameterHandlers = new ParameterHandler<?>[parameterCount];
    for (int p = 0, lastParameter = parameterCount - 1; p < parameterCount; p++) {
        parameterHandlers[p] =
                //解析注解参数 异常判断
                parseParameter(p, parameterTypes[p], parameterAnnotationsArray[p], p == lastParameter);
    }
    //错误检测,最后会抛IllegalArgumentException
    ...
    return new RequestFactory(this);
}

然后再查看HttpServiceMethod.parseAnnotations()

static <ResponseT, ReturnT> HttpServiceMethod<ResponseT, ReturnT> parseAnnotations(
        Retrofit retrofit, Method method, RequestFactory requestFactory) {
    boolean isKotlinSuspendFunction = requestFactory.isKotlinSuspendFunction;
    boolean continuationWantsResponse = false;
    boolean continuationBodyNullable = false;
    Annotation[] annotations = method.getAnnotations();
    Type adapterType;
    //如果方法是kotlin中的suspend方法
    if (isKotlinSuspendFunction) {
        //获取泛型参数
        Type[] parameterTypes = method.getGenericParameterTypes();
        Type responseType = Utils.getParameterLowerBound(0,
                (ParameterizedType) parameterTypes[parameterTypes.length - 1]);
        //如果参数是Response类型
        if (getRawType(responseType) == Response.class && responseType instanceof ParameterizedType) {
            // Unwrap the actual body type from Response<T>.
            responseType = Utils.getParameterUpperBound(0, (ParameterizedType) responseType);
            //设置为true 说明需要的是Response
            continuationWantsResponse = true;
        } else {
            // TODO figure out if type is nullable or not
            // Metadata metadata = method.getDeclaringClass().getAnnotation(Metadata.class)
            // Find the entry for method
            // Determine if return type is nullable or not
        }

        adapterType = new Utils.ParameterizedTypeImpl(null, Call.class, responseType);
        annotations = SkipCallbackExecutorImpl.ensurePresent(annotations);
    } else {
        //否则湖区方法返回值的泛型参数 即为请求需要的返回值类型
        adapterType = method.getGenericReturnType();
    }
    //根据网络请求接口方法的返回值和注解类型
    //从Retrofit对象中获取网络请求适配器
    CallAdapter<ResponseT, ReturnT> callAdapter =
            createCallAdapter(retrofit, method, adapterType, annotations);
    //得到响应类型
    Type responseType = callAdapter.responseType();
    if (responseType == okhttp3.Response.class) {
        throw methodError(method, "'"
                + getRawType(responseType).getName()
                + "' is not a valid response body type. Did you mean ResponseBody?");
    }
    if (responseType == Response.class) {
        throw methodError(method, "Response must include generic type (e.g., Response<String>)");
    }
    // TODO support Unit for Kotlin?
    if (requestFactory.httpMethod.equals("HEAD") && !Void.class.equals(responseType)) {
        throw methodError(method, "HEAD method must use Void as response type.");
    }
    //根据网络请求接口方法的返回值和注解类型从Retrofit对象中获取对应的数据转换器
    Converter<ResponseBody, ResponseT> responseConverter =
            createResponseConverter(retrofit, method, responseType);

    okhttp3.Call.Factory callFactory = retrofit.callFactory;
    if (!isKotlinSuspendFunction) {
        //不是suspend方法的话则直接创建并返回一个CallAdapted对象
        return new HttpServiceMethod.CallAdapted<>(requestFactory, callFactory, responseConverter, callAdapter);
    } else if (continuationWantsResponse) {
        //noinspection unchecked Kotlin compiler guarantees ReturnT to be Object.
        //返回Response类型的HttpServiceMethod
        return (HttpServiceMethod<ResponseT, ReturnT>) new HttpServiceMethod.SuspendForResponse<>(requestFactory,
                callFactory, responseConverter, (CallAdapter<ResponseT, Call<ResponseT>>) callAdapter);
    } else {
        //noinspection unchecked Kotlin compiler guarantees ReturnT to be Object.
        //返回具体类型的HttpServiceMethod
        return (HttpServiceMethod<ResponseT, ReturnT>) new HttpServiceMethod.SuspendForBody<>(requestFactory,
                callFactory, responseConverter, (CallAdapter<ResponseT, Call<ResponseT>>) callAdapter,
                continuationBodyNullable);
    }
}

上述方法主要作用是获取CallAdapter以及Converter对象,并构建对应的HttpServiceMethod,根据是否是kotlin suspend方法,如果是构建并返回一个CallAdapter对象,否则根据是否需要具体类型返回SuspendForBody和SuspendForResponse

接着来查看CallAdapter的获取方法 createCallAdapter()

private static <ResponseT, ReturnT> CallAdapter<ResponseT, ReturnT> createCallAdapter(
        Retrofit retrofit, Method method, Type returnType, Annotation[] annotations) {
    try {
        //noinspection unchecked
        return (CallAdapter<ResponseT, ReturnT>) retrofit.callAdapter(returnType, annotations);
    } catch (RuntimeException e) { // Wide exception range because factories are user code.
        throw methodError(method, e, "Unable to create call adapter for %s", returnType);
    }
}
public CallAdapter<?, ?> callAdapter(Type returnType, Annotation[] annotations) {
    return nextCallAdapter(null, returnType, annotations);
}
public CallAdapter<?, ?> nextCallAdapter(@Nullable CallAdapter.Factory skipPast, Type returnType,
                                         Annotation[] annotations) {
    checkNotNull(returnType, "returnType == null");
    checkNotNull(annotations, "annotations == null");
    int start = callAdapterFactories.indexOf(skipPast) + 1;
    //遍历CallAdapter.Factory集合寻找合适的CallAdapter
    for (int i = start, count = callAdapterFactories.size(); i < count; i++) {
        CallAdapter<?, ?> adapter = callAdapterFactories.get(i).get(returnType, annotations, this);
        if (adapter != null) {
            return adapter;
        }
    }
    //否则抛出异常
    StringBuilder builder = new StringBuilder("Could not locate call adapter for ")
            .append(returnType)
            .append(".\n");
    if (skipPast != null) {
        builder.append("  Skipped:");
        for (int i = 0; i < start; i++) {
            builder.append("\n   * ").append(callAdapterFactories.get(i).getClass().getName());
        }
        builder.append('\n');
    }
    builder.append("  Tried:");
    for (int i = start, count = callAdapterFactories.size(); i < count; i++) {
        builder.append("\n   * ").append(callAdapterFactories.get(i).getClass().getName());
    }
    throw new IllegalArgumentException(builder.toString());
}

接着来查看Converter的获取方法 createResponseConverter()
,和CallAdapter的获取方法相似

private static <ResponseT> Converter<ResponseBody, ResponseT> createResponseConverter(
        Retrofit retrofit, Method method, Type responseType) {
    Annotation[] annotations = method.getAnnotations();
    try {
        return retrofit.responseBodyConverter(responseType, annotations);
    } catch (RuntimeException e) { // Wide exception range because factories are user code.
        throw methodError(method, e, "Unable to create converter for %s", responseType);
    }
}
public <T> Converter<ResponseBody, T> responseBodyConverter(Type type, Annotation[] annotations) {
    return nextResponseBodyConverter(null, type, annotations);
}
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;
    //遍历converterFactories集合,找到合适的Converter
    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;
        }
    }
    //没找到则抛出IllegalArgumentException
    StringBuilder builder = new StringBuilder("Could not locate ResponseBody converter for ")
            .append(type)
            .append(".\n");
    if (skipPast != null) {
        builder.append("  Skipped:");
        for (int i = 0; i < start; i++) {
            builder.append("\n   * ").append(converterFactories.get(i).getClass().getName());
        }
        builder.append('\n');
    }
    builder.append("  Tried:");
    for (int i = start, count = converterFactories.size(); i < count; i++) {
        builder.append("\n   * ").append(converterFactories.get(i).getClass().getName());
    }
    throw new IllegalArgumentException(builder.toString());
}

然后回到刚开始动态代理创建网络接口Service的地方

//获取响应类型
abstract @Nullable T invoke(Object[] args);

然后看HttpServiceMethod对invoke()方法的实现

@Override final @Nullable ReturnT invoke(Object[] args) {
    Call<ResponseT> call = new OkHttpCall<>(requestFactory, args, callFactory, responseConverter);
    return adapt(call, args);
}
//抽象方法 有HttpServiceMethod的子类实现
protected abstract @Nullable ReturnT adapt(Call<ResponseT> call, Object[] args);

非协程情况下是CallAdapter,最终由ExecutorCallbackCall(也是一个Call对象)执行网络请求

return new CallAdapter<Object, Call<?>>() {
    @Override public Type responseType() {
        return responseType;
    }

    @Override public Call<Object> adapt(Call<Object> call) {
        return executor == null
                ? call
                : new DefaultCallAdapterFactory.ExecutorCallbackCall<>(executor, call);
    }
};  

接着回到call的创建Call<ResponseBody> article = service.getArticle(1),上面也已讲解到最终会调用InvocationHandler.invoke()方法,默认情况下得到的是ExecutorCallbackCall对象

接着查看请求的执行,同步请求Response<ResponseBody> response = article.execute()

@Override public Response<T> execute() throws IOException {
    okhttp3.Call call;
    //防止并发安全
    synchronized (this) {
        //同一个请求只能执行一次
        if (executed) throw new IllegalStateException("Already executed.");
        executed = true;
        if (creationFailure != null) {
            if (creationFailure instanceof IOException) {
                throw (IOException) creationFailure;
            } else if (creationFailure instanceof RuntimeException) {
                throw (RuntimeException) creationFailure;
            } else {
                throw (Error) creationFailure;
            }
        }
        call = rawCall;
        if (call == null) {
            try {
                //创建OkHttp的Call对象
                call = rawCall = createRawCall();
            } catch (IOException | RuntimeException | Error e) {
                throwIfFatal(e); //  Do not assign a fatal error to creationFailure.
                creationFailure = e;
                throw e;
            }
        }
    }
    if (canceled) {
        //取消请求
        call.cancel();
    }
    //执行请求并解析返回结果
    return parseResponse(call.execute());
}

接着查看Call对象的创建createRawCall()

private okhttp3.Call createRawCall() throws IOException {
    //callFactory默认为OkHttpClient 创建RealCall的过程
    okhttp3.Call call = callFactory.newCall(requestFactory.create(args));
    if (call == null) {
        throw new NullPointerException("Call.Factory returned null.");
    }
    return call;
}

下面接着查看请求执行后,响应结果的解析parseResponse (),会抛出
IOException

Response<T> parseResponse(okhttp3.Response rawResponse) throws IOException {
    ResponseBody rawBody = rawResponse.body();
    // Remove the body's source (the only stateful object) so we can pass the response along.
    rawResponse = rawResponse.newBuilder()
            .body(new OkHttpCall.NoContentResponseBody(rawBody.contentType(), rawBody.contentLength()))
            .build();
    int code = rawResponse.code();
    if (code < 200 || code >= 300) {
        try {
            // Buffer the entire body to avoid future I/O.
            ResponseBody bufferedBody = Utils.buffer(rawBody);
            return Response.error(bufferedBody, rawResponse);
        } finally {
            rawBody.close();
        }
    }
    if (code == 204 || code == 205) {
        rawBody.close();
        return Response.success(null, rawResponse);
    }
    OkHttpCall.ExceptionCatchingResponseBody catchingBody = new OkHttpCall.ExceptionCatchingResponseBody(rawBody);
    try {
        //调用Converter解析body
        T body = responseConverter.convert(catchingBody);
        return Response.success(body, rawResponse);
    } catch (RuntimeException e) {
        // If the underlying source threw an exception, propagate that rather than indicating it was
        // a runtime exception.
        catchingBody.throwIfCaught();
        throw e;
    }
}

同步调用这样执行完成了,总的来说就是通过动态代理来生成网络接口Service,并解析注解获取请求方式和参数,然后调用invoke()方法获取Call,最后同步执行请求

下面接着查看异步请求的过程article.enqueue(new Callback())

@Override public void enqueue(final Callback<T> callback) {
    checkNotNull(callback, "callback == null");
    okhttp3.Call call;
    Throwable failure;
    synchronized (this) {
        if (executed) throw new IllegalStateException("Already executed.");
        executed = true;
        call = rawCall;
        failure = creationFailure;
        if (call == null && failure == null) {
            try {
                //创建OkHttp的Call对象
                call = rawCall = createRawCall();
            } catch (Throwable t) {
                throwIfFatal(t);
                failure = creationFailure = t;
            }
        }
    }
    if (failure != null) {
        //回调失败
        callback.onFailure(this, failure);
        return;
    }
    if (canceled) {
        //取消请求
        call.cancel();
    }
    //调用call的异步执行方法 根据OkHttp的Callback设置回调结果
    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
            }
        }
    });
}

异步请求的实现是创建一个RealCall,然后eqeue()到请求队列中,通过监听OkHttp的Callback来设置自己的回调

4.自定义Factory

4.1. GsonConverterFactory数据解析工厂

首先看一下GsonConverterFactory,他是在配置Retrofit时添加的

public Retrofit.Builder addConverterFactory(Converter.Factory factory) {
    converterFactories.add(checkNotNull(factory, "factory == null"));
    return this;
}

然后再查看GsonConverterFactory

public final class GsonConverterFactory extends Converter.Factory {
    //创建Gson转换器
    public static retrofit2.converter.gson.GsonConverterFactory create() {
        return create(new Gson());
    }
    //创建GsonConverterFactory
    @SuppressWarnings("ConstantConditions")
    public static retrofit2.converter.gson.GsonConverterFactory create(Gson gson) {
        if (gson == null) throw new NullPointerException("gson == null");
        return new retrofit2.converter.gson.GsonConverterFactory(gson);
    }
    private final Gson gson;
    //传入Gson实例
    private GsonConverterFactory(Gson gson) {
        this.gson = gson;
    }
    //转化响应体(在获取响应数据时调用)
    @Override
    public Converter<ResponseBody, ?> responseBodyConverter(Type type, Annotation[] annotations,
                                                            Retrofit retrofit) {
        //type为响应实体
        TypeAdapter<?> adapter = gson.getAdapter(TypeToken.get(type));
        return new GsonResponseBodyConverter<>(gson, adapter);
    }
    //转换请求体(在解析注解请求方法时调用)
    @Override
    public Converter<?, RequestBody> requestBodyConverter(Type type,
                                                          Annotation[] parameterAnnotations, Annotation[] methodAnnotations, Retrofit retrofit) {
        TypeAdapter<?> adapter = gson.getAdapter(TypeToken.get(type));
        return new GsonRequestBodyConverter<>(gson, adapter);
    }
}

然后再查看GsonRequestBodyConverter类

final class GsonResponseBodyConverter<T> implements Converter<ResponseBody, T> {
    private final Gson gson;
    private final TypeAdapter<T> adapter;
    //传入Gson对象和类型适配器
    GsonResponseBodyConverter(Gson gson, TypeAdapter<T> adapter) {
        this.gson = gson;
        this.adapter = adapter;
    }
    //将ResponseBody转化为实体类
    @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();
        }
    }
}

4.2. RxJava2CallAdapterFactory数据回调适配工厂

同上RxJava2CallAdapterFactory也是在Retrofit初始化时配置的

public Retrofit.Builder addCallAdapterFactory(CallAdapter.Factory factory) {
    callAdapterFactories.add(checkNotNull(factory, "factory == null"));
    return this;
}

接着查看RxJava2CallAdapterFactory类

public final class RxJava2CallAdapterFactory extends CallAdapter.Factory {
    //RxJava2CallAdapterFactory初始化
    public static RxJava2CallAdapterFactory create() {
        return new RxJava2CallAdapterFactory(null, false);
    }
    public static RxJava2CallAdapterFactory createAsync() {
        return new RxJava2CallAdapterFactory(null, true);
    }
    @SuppressWarnings("ConstantConditions") // Guarding public API nullability.
    public static RxJava2CallAdapterFactory createWithScheduler(Scheduler scheduler) {
        if (scheduler == null) throw new NullPointerException("scheduler == null");
        return new RxJava2CallAdapterFactory(scheduler, false);
    }
    //线程调度器
    private final @Nullable Scheduler scheduler;
    //同步还是异步请求
    private final boolean isAsync;
    private RxJava2CallAdapterFactory(@Nullable Scheduler scheduler, boolean isAsync) {
        this.scheduler = scheduler;
        this.isAsync = isAsync;
    }
    @Override public @Nullable CallAdapter<?, ?> get(
            Type returnType, Annotation[] annotations, Retrofit retrofit) {
        //returnType 就是请求方法的返回值类型 Observable<TestBean>
        //返回的class Observable.class
        Class<?> rawType = getRawType(returnType);
        //Completable 只关心onComplete事件 实体类为void
        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>");
        }
        //获取响应数据类型 实体类TestBean
        Type observableType = getParameterUpperBound(0, (ParameterizedType) returnType);
        //再次获取数据类型的class
        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 {
            //设置实体类 TestBean
            responseType = observableType;
            isBody = true;
        }
        //生成具体的RxJava2CallAdapter
        return new RxJava2CallAdapter(responseType, scheduler, isAsync, isResult, isBody, isFlowable,
                isSingle, isMaybe, false);
    }
}

然后再看RxJava2CallAdapter的创建

final class RxJava2CallAdapter<R> implements CallAdapter<R, Object> {
    //返回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) {
            //响应实体类 isBody为true
            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();
        }
        //返回BodyObservable
        return RxJavaPlugins.onAssembly(observable);
    }
}

大概就介绍到这里,后面再了解RxJava源码时,再回头复习下

5.总结

这一章节学习了android中的又一流行框架,使用起来是非常灵活的,耦合性非常低,可以自定义CallAdapterFactory和ConverterFactory,设计的十分巧妙