Android Retrofit 框架适配器模块深入源码分析(五)

Android Retrofit 框架适配器模块深入源码分析

一、引言

在 Android 开发中，网络请求是一个常见且重要的功能。Retrofit 作为一个强大的网络请求框架，以其简洁的 API 和高度的可定制性受到了广泛的欢迎。适配器模块（CallAdapter）是 Retrofit 框架中非常重要的一部分，它负责将 Call 对象转换为其他类型，如 Observable、Completable、Single 等，从而实现不同的异步编程模式。通过对 Retrofit 适配器模块的源码分析，我们可以深入了解其工作原理，从而更好地使用和扩展这个框架。

二、适配器模块的核心概念

2.1 `CallAdapter` 接口

CallAdapter 接口是适配器模块的核心接口，它定义了将 Call 对象转换为其他类型的方法。以下是 CallAdapter 接口的源码：

java

// CallAdapter 接口定义了将 Call 对象转换为其他类型的方法
public interface CallAdapter<R, T> {
    // 获取响应类型
    Type responseType();

    // 将 Call 对象转换为其他类型
    T adapt(Call<R> call);

    // 工厂类，用于创建 CallAdapter 对象
    abstract class Factory {
        // 创建 CallAdapter 对象
        public abstract @Nullable CallAdapter<?, ?> get(Type returnType, Annotation[] annotations,
                                                        Retrofit retrofit);

        // 辅助方法，用于获取泛型类型参数
        protected static Type getParameterUpperBound(int index, ParameterizedType type) {
            return Utils.getParameterUpperBound(index, type);
        }

        // 辅助方法，用于获取原始类型
        protected static Class<?> getRawType(Type type) {
            return Utils.getRawType(type);
        }
    }
}

这个接口定义了两个方法：responseType 方法用于获取响应类型，adapt 方法用于将 Call 对象转换为其他类型。Factory 抽象类是用于创建 CallAdapter 对象的工厂类，其中 get 方法用于根据返回类型和注解创建 CallAdapter 对象。

2.2 `Retrofit` 类中的相关方法

Retrofit 类是 Retrofit 框架的核心类，它包含了与适配器模块相关的方法。以下是部分相关源码：

java

public final class Retrofit {
    // 适配器工厂列表
    private final List<CallAdapter.Factory> callAdapterFactories;

    Retrofit(okhttp3.Call.Factory callFactory, HttpUrl baseUrl,
             List<Converter.Factory> converterFactories, List<CallAdapter.Factory> callAdapterFactories,
             Executor callbackExecutor, boolean validateEagerly) {
        this.callFactory = callFactory;
        this.baseUrl = baseUrl;
        // 保存适配器工厂列表
        this.callAdapterFactories = unmodifiableList(callAdapterFactories);
        this.converterFactories = unmodifiableList(converterFactories);
        this.callbackExecutor = callbackExecutor;
        this.validateEagerly = validateEagerly;
    }

    // 创建 CallAdapter 对象
    private CallAdapter<Object, Object> createCallAdapter(Method method) {
        Type returnType = method.getGenericReturnType();
        Annotation[] annotations = method.getAnnotations();
        try {
            // 遍历适配器工厂列表，查找合适的适配器工厂
            for (CallAdapter.Factory factory : callAdapterFactories) {
                // 调用工厂的 get 方法创建 CallAdapter 对象
                CallAdapter<?, ?> adapter = factory.get(returnType, annotations, this);
                if (adapter != null) {
                    // 找到合适的适配器，返回
                    return (CallAdapter<Object, Object>) adapter;
                }
            }
        } catch (RuntimeException e) {
            throw methodError(method, e, "Unable to create call adapter for %s", returnType);
        }
        throw methodError(method, "Could not locate call adapter for %s.", returnType);
    }
}

Retrofit 类中保存了一个 callAdapterFactories 列表，用于存储所有的适配器工厂。createCallAdapter 方法用于创建 CallAdapter 对象，它会遍历 callAdapterFactories 列表，调用每个工厂的 get 方法，直到找到合适的适配器为止。

三、常见的适配器工厂实现

3.1 `DefaultCallAdapterFactory`

DefaultCallAdapterFactory 是 Retrofit 提供的默认适配器工厂，它处理 Call 类型的返回值。以下是其源码分析：

java

// 默认的 CallAdapter 工厂类
final class DefaultCallAdapterFactory extends CallAdapter.Factory {
    private final @Nullable Executor callbackExecutor;

    DefaultCallAdapterFactory(@Nullable Executor callbackExecutor) {
        this.callbackExecutor = callbackExecutor;
    }

    // 创建 CallAdapter 对象
    @Override
    public @Nullable CallAdapter<?, ?> get(Type returnType, Annotation[] annotations,
                                           Retrofit retrofit) {
        // 检查返回类型是否为 Call 类型
        if (getRawType(returnType) != Call.class) {
            return null;
        }
        // 检查返回类型是否为参数化类型
        if (!(returnType instanceof ParameterizedType)) {
            throw new IllegalStateException("Call return type must be parameterized as Call<Foo> or Call<? extends Foo>");
        }
        // 获取泛型类型参数
        final Type responseType = getParameterUpperBound(0, (ParameterizedType) returnType);

        final Executor executor = Utils.isAnnotationPresent(annotations, SkipCallbackExecutor.class)
              ? null
                : callbackExecutor;

        // 返回 CallAdapter 对象
        return new CallAdapter<Object, Call<?>>() {
            @Override
            public Type responseType() {
                return responseType;
            }

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

    // 带有回调执行器的 Call 包装类
    static final class ExecutorCallbackCall<T> implements Call<T> {
        final Executor callbackExecutor;
        final Call<T> delegate;

        ExecutorCallbackCall(Executor callbackExecutor, Call<T> delegate) {
            this.callbackExecutor = callbackExecutor;
            this.delegate = delegate;
        }

        @Override
        public void enqueue(final Callback<T> callback) {
            // 检查回调是否为空
            checkNotNull(callback, "callback == null");

            delegate.enqueue(new Callback<T>() {
                @Override
                public void onResponse(Call<T> call, final Response<T> response) {
                    // 在回调执行器中执行回调
                    callbackExecutor.execute(new Runnable() {
                        @Override
                        public void run() {
                            if (delegate.isCanceled()) {
                                // 如果请求已取消，调用回调的 onFailure 方法
                                callback.onFailure(ExecutorCallbackCall.this, new IOException("Canceled"));
                            } else {
                                // 调用回调的 onResponse 方法
                                callback.onResponse(ExecutorCallbackCall.this, response);
                            }
                        }
                    });
                }

                @Override
                public void onFailure(Call<T> call, final Throwable t) {
                    // 在回调执行器中执行回调
                    callbackExecutor.execute(new Runnable() {
                        @Override
                        public void run() {
                            // 调用回调的 onFailure 方法
                            callback.onFailure(ExecutorCallbackCall.this, t);
                        }
                    });
                }
            });
        }

        @Override
        public boolean isExecuted() {
            return delegate.isExecuted();
        }

        @Override
        public Response<T> execute() throws IOException {
            return delegate.execute();
        }

        @Override
        public void cancel() {
            delegate.cancel();
        }

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

        @SuppressWarnings("CloneDoesntCallSuperClone") // Performing deep clone.
        @Override
        public Call<T> clone() {
            return new ExecutorCallbackCall<>(callbackExecutor, delegate.clone());
        }

        @Override
        public Request request() {
            return delegate.request();
        }
    }
}

3.1.1 `DefaultCallAdapterFactory` 类

DefaultCallAdapterFactory 类实现了 CallAdapter.Factory 抽象类，其 get 方法首先检查返回类型是否为 Call 类型，然后获取泛型类型参数。如果需要回调执行器，则返回一个 ExecutorCallbackCall 对象，否则返回原始的 Call 对象。

3.1.2 `ExecutorCallbackCall` 类

ExecutorCallbackCall 类是一个 Call 的包装类，它在 enqueue 方法中，将回调的 onResponse 和 onFailure 方法放在回调执行器中执行，以确保回调在指定的线程中执行。

3.2 `RxJava2CallAdapterFactory`

RxJava2CallAdapterFactory 是 Retrofit 与 RxJava 2 集成的适配器工厂，它可以将 Call 对象转换为 Observable、Single、Completable 等 RxJava 类型。以下是其源码分析：

java

// RxJava2 的 CallAdapter 工厂类
public final class RxJava2CallAdapterFactory extends CallAdapter.Factory {
    private final @Nullable Scheduler scheduler;
    private final boolean isAsync;

    // 创建 RxJava2CallAdapterFactory 实例
    public static RxJava2CallAdapterFactory create() {
        return new RxJava2CallAdapterFactory(null, false);
    }

    // 创建 RxJava2CallAdapterFactory 实例，指定调度器
    public static RxJava2CallAdapterFactory createWithScheduler(Scheduler scheduler) {
        checkNotNull(scheduler, "scheduler == null");
        return new RxJava2CallAdapterFactory(scheduler, false);
    }

    RxJava2CallAdapterFactory(@Nullable Scheduler scheduler, boolean isAsync) {
        this.scheduler = scheduler;
        this.isAsync = isAsync;
    }

    // 创建 CallAdapter 对象
    @Override
    public @Nullable CallAdapter<?, ?> get(Type returnType, Annotation[] annotations,
                                           Retrofit retrofit) {
        // 获取原始类型
        Class<?> rawType = getRawType(returnType);

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

        // 检查返回类型是否为参数化类型
        if (!(returnType instanceof ParameterizedType)) {
            String name = isFlowable ? "Flowable" : isSingle ? "Single" : isMaybe ? "Maybe"
                    : isCompletable ? "Completable" : "Observable";
            throw new IllegalStateException(name + " return type must be parameterized"
                    + " as " + name + "<Foo> or " + name + "<? extends Foo>");
        }

        // 获取泛型类型参数
        Type responseType = getParameterUpperBound(0, (ParameterizedType) returnType);

        boolean isResult = false;
        boolean isBody = false;
        Class<?> rawWrapperType = getRawType(responseType);
        if (rawWrapperType == Response.class) {
            if (!(responseType instanceof ParameterizedType)) {
                throw new IllegalStateException("Response must be parameterized"
                        + " as Response<Foo> or Response<? extends Foo>");
            }
            responseType = getParameterUpperBound(0, (ParameterizedType) responseType);
        } else if (rawWrapperType == Result.class) {
            if (!(responseType instanceof ParameterizedType)) {
                throw new IllegalStateException("Result must be parameterized"
                        + " as Result<Foo> or Result<? extends Foo>");
            }
            responseType = getParameterUpperBound(0, (ParameterizedType) responseType);
            isResult = true;
        } else {
            isBody = true;
        }

        Scheduler scheduler = this.scheduler;
        if (scheduler == null) {
            boolean isAsync = Utils.isAnnotationPresent(annotations, Async.class);
            scheduler = isAsync ? Schedulers.io() : null;
        }

        // 返回 CallAdapter 对象
        if (isCompletable) {
            return new RxJava2CallAdapter(Void.class, scheduler, isAsync, isResult, isBody, false);
        } else if (isFlowable) {
            return new RxJava2CallAdapter(responseType, scheduler, isAsync, isResult, isBody, true);
        } else if (isSingle) {
            return new RxJava2CallAdapter(responseType, scheduler, isAsync, isResult, isBody, false);
        } else if (isMaybe) {
            return new RxJava2CallAdapter(responseType, scheduler, isAsync, isResult, isBody, false);
        } else {
            return new RxJava2CallAdapter(responseType, scheduler, isAsync, isResult, isBody, false);
        }
    }
}

// RxJava2 的 CallAdapter 实现类
final class RxJava2CallAdapter<R> implements CallAdapter<R, Object> {
    private final Type responseType;
    private final @Nullable Scheduler scheduler;
    private final boolean isAsync;
    private final boolean isResult;
    private final boolean isBody;
    private final boolean isFlowable;

    RxJava2CallAdapter(Type responseType, @Nullable Scheduler scheduler, boolean isAsync,
                       boolean isResult, boolean isBody, boolean isFlowable) {
        this.responseType = responseType;
        this.scheduler = scheduler;
        this.isAsync = isAsync;
        this.isResult = isResult;
        this.isBody = isBody;
        this.isFlowable = isFlowable;
    }

    @Override
    public Type responseType() {
        return responseType;
    }

    @Override
    public Object adapt(Call<R> call) {
        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);
        } else if (isSingle) {
            return observable.singleOrError();
        } else if (isMaybe) {
            return observable.singleElement();
        } else if (isCompletable) {
            return observable.ignoreElements();
        } else {
            return observable;
        }
    }
}

// 异步调用的 Observable 实现类
final class CallEnqueueObservable<T> extends Observable<Response<T>> {
    private final Call<T> originalCall;

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

    @Override
    protected void subscribeActual(Observer<? super Response<T>> observer) {
        // 克隆原始的 Call 对象
        Call<T> call = originalCall.clone();
        observer.onSubscribe(new CallDisposable(call));
        call.enqueue(new Callback<T>() {
            @Override
            public void onResponse(Call<T> call, Response<T> response) {
                if (call.isCanceled()) return;
                observer.onNext(response);
                observer.onComplete();
            }

            @Override
            public void onFailure(Call<T> call, Throwable t) {
                if (call.isCanceled()) return;
                observer.onError(t);
            }
        });
    }

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

// 同步调用的 Observable 实现类
final class CallExecuteObservable<T> extends Observable<Response<T>> {
    private final Call<T> originalCall;

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

    @Override
    protected void subscribeActual(Observer<? super Response<T>> observer) {
        // 克隆原始的 Call 对象
        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));
                }
            }
        }
    }

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

// 处理结果的 Observable 实现类
final class ResultObservable<T> extends Observable<Result<T>> {
    private final Observable<Response<T>> upstream;

    ResultObservable(Observable<Response<T>> upstream) {
        this.upstream = upstream;
    }

    @Override
    protected void subscribeActual(Observer<? super Result<T>> observer) {
        upstream.subscribe(new ResultObserver<>(observer));
    }

    private static final class ResultObserver<R> implements Observer<Response<R>> {
        private final Observer<? super Result<R>> observer;

        ResultObserver(Observer<? super Result<R>> observer) {
            this.observer = observer;
        }

        @Override
        public void onSubscribe(Disposable d) {
            observer.onSubscribe(d);
        }

        @Override
        public void onNext(Response<R> response) {
            observer.onNext(Result.response(response));
        }

        @Override
        public void onError(Throwable t) {
            try {
                observer.onNext(Result.error(t));
            } catch (Throwable inner) {
                Exceptions.throwIfFatal(inner);
                RxJavaPlugins.onError(new CompositeException(t, inner));
            }
            observer.onComplete();
        }

        @Override
        public void onComplete() {
            observer.onComplete();
        }
    }
}

// 处理响应体的 Observable 实现类
final class BodyObservable<T> extends Observable<T> {
    private final Observable<Response<T>> upstream;

    BodyObservable(Observable<Response<T>> upstream) {
        this.upstream = upstream;
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        upstream.subscribe(new BodyObserver<>(observer));
    }

    private static final class BodyObserver<R> implements Observer<Response<R>> {
        private final Observer<? super R> observer;
        private boolean terminated;

        BodyObserver(Observer<? super R> observer) {
            this.observer = observer;
        }

        @Override
        public void onSubscribe(Disposable d) {
            observer.onSubscribe(d);
        }

        @Override
        public void onNext(Response<R> response) {
            if (response.isSuccessful()) {
                observer.onNext(response.body());
            } else {
                terminated = true;
                Throwable t = new HttpException(response);
                try {
                    observer.onError(t);
                } catch (Throwable inner) {
                    Exceptions.throwIfFatal(inner);
                    RxJavaPlugins.onError(new CompositeException(t, inner));
                }
            }
        }

        @Override
        public void onError(Throwable t) {
            if (!terminated) {
                observer.onError(t);
            } else {
                // This should never happen! onNext handles and forwards errors automatically.
                Throwable broken = new AssertionError(
                        "This should never happen! Report as a bug with the full stacktrace.");
                //noinspection UnnecessaryInitCause Two-arg AssertionError constructor is 1.7+ only.
                broken.initCause(t);
                RxJavaPlugins.onError(broken);
            }
        }

        @Override
        public void onComplete() {
            if (!terminated) {
                observer.onComplete();
            }
        }
    }
}

3.2.1 `RxJava2CallAdapterFactory` 类

RxJava2CallAdapterFactory 类实现了 CallAdapter.Factory 抽象类，其 get 方法首先检查返回类型是否为 RxJava 类型（如 Observable、Flowable、Single、Maybe、Completable），然后获取泛型类型参数。根据不同的情况，返回不同的 RxJava2CallAdapter 对象。

3.2.2 `RxJava2CallAdapter` 类

RxJava2CallAdapter 类实现了 CallAdapter 接口，其 adapt 方法根据不同的配置，将 Call 对象转换为不同的 RxJava 类型。例如，如果是异步调用，使用 CallEnqueueObservable；如果是同步调用，使用 CallExecuteObservable。

3.2.3 `CallEnqueueObservable` 类

CallEnqueueObservable 类是一个 Observable 的实现类，它在 subscribeActual 方法中，使用 Call 的 enqueue 方法进行异步调用，并将结果传递给观察者。

3.2.4 `CallExecuteObservable` 类

CallExecuteObservable 类是一个 Observable 的实现类，它在 subscribeActual 方法中，使用 Call 的 execute 方法进行同步调用，并将结果传递给观察者。

3.2.5 `ResultObservable` 类

ResultObservable 类是一个 Observable 的实现类，它将 Response 对象包装成 Result 对象，并传递给观察者。

3.2.6 `BodyObservable` 类

BodyObservable 类是一个 Observable 的实现类，它将 Response 对象的响应体传递给观察者，如果响应不成功，则抛出 HttpException 异常。

四、适配器模块的工作流程

4.1 创建 Retrofit 实例时添加适配器工厂

java

// 创建 Retrofit 实例时添加 RxJava2CallAdapterFactory
Retrofit retrofit = new Retrofit.Builder()
       .baseUrl("https://api.example.com/")
       .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
       .build();

在创建 Retrofit 实例时，通过 addCallAdapterFactory 方法添加适配器工厂，这些工厂会被添加到 Retrofit 类的 callAdapterFactories 列表中。

4.2 接口方法调用时创建适配器

java

// 定义 API 接口
public interface ApiService {
    @GET("users/{id}")
    Observable<User> getUser(@Path("id") int id);
}

// 创建 API 服务实例
ApiService apiService = retrofit.create(ApiService.class);
Observable<User> observable = apiService.getUser(1);

当调用 API 接口方法时，Retrofit 会根据方法的返回类型，调用 createCallAdapter 方法，遍历 callAdapterFactories 列表，查找合适的适配器工厂并创建适配器。

4.3 适配器将 `Call` 对象转换为其他类型

java

// 在 RxJava2CallAdapter 类中进行转换
final class RxJava2CallAdapter<R> implements CallAdapter<R, Object> {
    @Override
    public Object adapt(Call<R> call) {
        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);
        } else if (isSingle) {
            return observable.singleOrError();
        } else if (isMaybe) {
            return observable.singleElement();
        } else if (isCompletable) {
            return observable.ignoreElements();
        } else {
            return observable;
        }
    }
}

在 adapt 方法中，根据不同的配置，将 Call 对象转换为不同的 RxJava 类型，如 Observable、Flowable、Single、Maybe、Completable 等。

五、自定义适配器工厂

5.1 自定义适配器接口

java

// 自定义 CallAdapter 接口
public interface CustomCallAdapter<R, T> {
    // 获取响应类型
    Type responseType();

    // 将 Call 对象转换为其他类型
    T adapt(Call<R> call);
}

// 自定义 CallAdapter 工厂接口
public abstract class CustomCallAdapterFactory extends CallAdapter.Factory {
    public abstract @Nullable CustomCallAdapter<?, ?> get(Type returnType, Annotation[] annotations,
                                                          Retrofit retrofit);
}

首先定义自定义的适配器接口 CustomCallAdapter 和适配器工厂接口 CustomCallAdapterFactory。

5. 自定义适配器工厂

5.2 实现自定义适配器工厂

以下是一个简单的自定义适配器工厂示例，该适配器工厂将 Call 对象转换为自定义的 CustomWrapper 类型。

java

import java.lang.reflect.Type;
import java.util.concurrent.Executor;

import retrofit2.Call;
import retrofit2.CallAdapter;
import retrofit2.Retrofit;

// 自定义包装类
class CustomWrapper<T> {
    private final Call<T> call;

    CustomWrapper(Call<T> call) {
        this.call = call;
    }

    public Call<T> getCall() {
        return call;
    }
}

// 自定义 CallAdapter 实现
class CustomCallAdapter<R> implements CallAdapter<R, CustomWrapper<R>> {
    private final Type responseType;
    private final Executor callbackExecutor;

    CustomCallAdapter(Type responseType, Executor callbackExecutor) {
        this.responseType = responseType;
        this.callbackExecutor = callbackExecutor;
    }

    @Override
    public Type responseType() {
        return responseType;
    }

    @Override
    public CustomWrapper<R> adapt(Call<R> call) {
        // 这里可以添加一些自定义逻辑，比如线程调度等
        return new CustomWrapper<>(call);
    }
}

// 自定义 CallAdapter 工厂实现
class CustomCallAdapterFactory extends CallAdapter.Factory {
    private final Executor callbackExecutor;

    CustomCallAdapterFactory(Executor callbackExecutor) {
        this.callbackExecutor = callbackExecutor;
    }

    @Override
    public CallAdapter<?, ?> get(Type returnType, Annotation[] annotations, Retrofit retrofit) {
        // 检查返回类型是否为自定义的 CustomWrapper 类型
        if (getRawType(returnType) != CustomWrapper.class) {
            return null;
        }
        // 检查返回类型是否为参数化类型
        if (!(returnType instanceof ParameterizedType)) {
            throw new IllegalStateException("CustomWrapper return type must be parameterized as CustomWrapper<Foo>");
        }
        // 获取泛型类型参数
        Type responseType = getParameterUpperBound(0, (ParameterizedType) returnType);
        return new CustomCallAdapter<>(responseType, callbackExecutor);
    }
}

代码解释：

CustomWrapper 类：这是一个简单的包装类，用于包装 Call 对象。
CustomCallAdapter 类：实现了 CallAdapter 接口，负责将 Call 对象转换为 CustomWrapper 对象。adapt 方法中可以添加一些自定义的逻辑，例如线程调度等。
CustomCallAdapterFactory 类：实现了 CallAdapter.Factory 抽象类，get 方法用于检查返回类型是否为 CustomWrapper 类型，如果是，则创建并返回 CustomCallAdapter 实例。

5.3 使用自定义适配器工厂

java

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import retrofit2.Retrofit;
import retrofit2.Call;
import retrofit2.http.GET;

// 定义 API 接口
interface CustomApiService {
    @GET("data")
    CustomWrapper<String> getData();
}

public class Main {
    public static void main(String[] args) {
        // 创建回调执行器
        Executor callbackExecutor = Executors.newSingleThreadExecutor();

        // 创建 Retrofit 实例并添加自定义适配器工厂
        Retrofit retrofit = new Retrofit.Builder()
               .baseUrl("https://example.com/")
               .addCallAdapterFactory(new CustomCallAdapterFactory(callbackExecutor))
               .build();

        // 创建 API 服务实例
        CustomApiService apiService = retrofit.create(CustomApiService.class);
        // 调用接口方法
        CustomWrapper<String> wrapper = apiService.getData();
        Call<String> call = wrapper.getCall();
        // 执行请求
        call.enqueue(new retrofit2.Callback<String>() {
            @Override
            public void onResponse(Call<String> call, retrofit2.Response<String> response) {
                if (response.isSuccessful()) {
                    System.out.println("Response: " + response.body());
                } else {
                    System.out.println("Error: " + response.code());
                }
            }

            @Override
            public void onFailure(Call<String> call, Throwable t) {
                System.out.println("Failure: " + t.getMessage());
            }
        });
    }
}

代码解释：

CustomApiService 接口：定义了一个返回 CustomWrapper<String> 类型的接口方法。
Main 类：创建了一个 Retrofit 实例，并添加了自定义的适配器工厂。然后创建了 API 服务实例，调用接口方法，最后执行请求并处理响应。

六、适配器模块的异常处理

6.1 创建适配器时的异常处理

在 Retrofit 类的 createCallAdapter 方法中，会遍历适配器工厂列表来创建适配器。如果在创建过程中出现异常，会捕获并抛出相应的错误信息。

java

public final class Retrofit {
    // ... 其他代码 ...

    private CallAdapter<Object, Object> createCallAdapter(Method method) {
        Type returnType = method.getGenericReturnType();
        Annotation[] annotations = method.getAnnotations();
        try {
            // 遍历适配器工厂列表，查找合适的适配器工厂
            for (CallAdapter.Factory factory : callAdapterFactories) {
                // 调用工厂的 get 方法创建 CallAdapter 对象
                CallAdapter<?, ?> adapter = factory.get(returnType, annotations, this);
                if (adapter != null) {
                    // 找到合适的适配器，返回
                    return (CallAdapter<Object, Object>) adapter;
                }
            }
        } catch (RuntimeException e) {
            // 捕获创建适配器过程中可能出现的运行时异常
            throw methodError(method, e, "Unable to create call adapter for %s", returnType);
        }
        throw methodError(method, "Could not locate call adapter for %s.", returnType);
    }

    // ... 其他代码 ...
}

代码解释：

如果在调用 factory.get 方法时抛出 RuntimeException，会将其封装成 methodError 异常抛出，提示无法为指定的返回类型创建适配器。
如果遍历完所有适配器工厂都没有找到合适的适配器，也会抛出 methodError 异常。

6.2 适配器转换过程中的异常处理

在适配器的 adapt 方法中，也可能会出现异常。例如，在 RxJava2CallAdapter 中，如果在创建 Observable 或进行类型转换时出现异常，会通过 RxJava 的错误处理机制进行处理。

java

final class RxJava2CallAdapter<R> implements CallAdapter<R, Object> {
    // ... 其他代码 ...

    @Override
    public Object adapt(Call<R> call) {
        try {
            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);
            } else if (isSingle) {
                return observable.singleOrError();
            } else if (isMaybe) {
                return observable.singleElement();
            } else if (isCompletable) {
                return observable.ignoreElements();
            } else {
                return observable;
            }
        } catch (Exception e) {
            // 处理异常，这里可以根据具体需求进行日志记录或其他操作
            throw new RuntimeException("Error adapting Call to RxJava type", e);
        }
    }
}

代码解释：

在 adapt 方法中，使用 try-catch 块捕获可能出现的异常。如果出现异常，会抛出一个新的 RuntimeException，并将原始异常作为 cause 传递。

6.3 异步调用中的异常处理

在异步调用中，如 CallEnqueueObservable 中，会在 onFailure 方法中处理请求失败的异常，并将其传递给观察者。

java

final class CallEnqueueObservable<T> extends Observable<Response<T>> {
    // ... 其他代码 ...

    @Override
    protected void subscribeActual(Observer<? super Response<T>> observer) {
        // 克隆原始的 Call 对象
        Call<T> call = originalCall.clone();
        observer.onSubscribe(new CallDisposable(call));
        call.enqueue(new Callback<T>() {
            @Override
            public void onResponse(Call<T> call, Response<T> response) {
                if (call.isCanceled()) return;
                observer.onNext(response);
                observer.onComplete();
            }

            @Override
            public void onFailure(Call<T> call, Throwable t) {
                if (call.isCanceled()) return;
                observer.onError(t);
            }
        });
    }

    // ... 其他代码 ...
}

代码解释：

当 Call 的 enqueue 方法调用失败时，会调用 onFailure 方法，将异常 t 传递给观察者的 onError 方法。

七、适配器模块的性能优化

7.1 缓存适配器

为了避免每次调用接口方法时都创建新的适配器，可以使用缓存机制。以下是一个简单的缓存实现示例：

java

import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.Map;

class CallAdapterCache {
    private static final Map<Type, CallAdapter<?, ?>> cache = new HashMap<>();

    public static <R, T> CallAdapter<R, T> getAdapter(Type returnType, CallAdapter.Factory factory, Retrofit retrofit) {
        @SuppressWarnings("unchecked")
        CallAdapter<R, T> adapter = (CallAdapter<R, T>) cache.get(returnType);
        if (adapter == null) {
            adapter = (CallAdapter<R, T>) factory.get(returnType, new Annotation[0], retrofit);
            if (adapter != null) {
                cache.put(returnType, adapter);
            }
        }
        return adapter;
    }
}

// 修改 Retrofit 类的 createCallAdapter 方法
public final class Retrofit {
    // ... 其他代码 ...

    private CallAdapter<Object, Object> createCallAdapter(Method method) {
        Type returnType = method.getGenericReturnType();
        Annotation[] annotations = method.getAnnotations();
        try {
            for (CallAdapter.Factory factory : callAdapterFactories) {
                CallAdapter<Object, Object> adapter = CallAdapterCache.getAdapter(returnType, factory, this);
                if (adapter != null) {
                    return adapter;
                }
            }
        } catch (RuntimeException e) {
            throw methodError(method, e, "Unable to create call adapter for %s", returnType);
        }
        throw methodError(method, "Could not locate call adapter for %s.", returnType);
    }

    // ... 其他代码 ...
}

代码解释：

CallAdapterCache 类使用一个 HashMap 来缓存适配器。getAdapter 方法首先检查缓存中是否存在对应的适配器，如果存在则直接返回，否则调用工厂的 get 方法创建适配器并缓存起来。
修改 Retrofit 类的 createCallAdapter 方法，使用缓存来获取适配器。

7.2 减少不必要的类型转换

在适配器的 adapt 方法中，尽量减少不必要的类型转换和中间对象的创建。例如，在 RxJava2CallAdapter 中，避免在不必要的情况下进行 Observable 的多次转换。

java

final class RxJava2CallAdapter<R> implements CallAdapter<R, Object> {
    // ... 其他代码 ...

    @Override
    public Object adapt(Call<R> call) {
        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);
        } else if (isSingle) {
            return observable.singleOrError();
        } else if (isMaybe) {
            return observable.singleElement();
        } else if (isCompletable) {
            return observable.ignoreElements();
        }
        return observable;
    }
}

代码解释：

在 adapt 方法中，根据不同的配置直接返回相应的 RxJava 类型，避免不必要的中间转换。

7.3 合理选择调度器

在使用 RxJava 适配器时，合理选择调度器可以提高性能。例如，如果是网络请求，通常使用 Schedulers.io() 来进行异步操作，避免阻塞主线程。

java

// 在 RxJava2CallAdapterFactory 中设置调度器
public final class RxJava2CallAdapterFactory extends CallAdapter.Factory {
    private final @Nullable Scheduler scheduler;
    private final boolean isAsync;

    // ... 其他代码 ...

    RxJava2CallAdapterFactory(@Nullable Scheduler scheduler, boolean isAsync) {
        this.scheduler = scheduler;
        this.isAsync = isAsync;
    }

    @Override
    public @Nullable CallAdapter<?, ?> get(Type returnType, Annotation[] annotations,
                                           Retrofit retrofit) {
        // ... 其他代码 ...

        Scheduler scheduler = this.scheduler;
        if (scheduler == null) {
            boolean isAsync = Utils.isAnnotationPresent(annotations, Async.class);
            scheduler = isAsync ? Schedulers.io() : null;
        }

        // ... 其他代码 ...
    }
}

代码解释：

在 RxJava2CallAdapterFactory 中，如果没有指定调度器，且是异步调用，则使用 Schedulers.io() 作为调度器。

八、适配器模块与其他模块的交互

8.1 与数据转换模块的交互

适配器模块和数据转换模块在请求和响应处理过程中相互协作。数据转换模块负责将请求参数序列化为 HTTP 请求体，以及将 HTTP 响应体反序列化为 Java 对象；适配器模块负责将 Call 对象转换为其他类型。

java

// 定义 API 接口
public interface ApiService {
    @GET("users/{id}")
    Observable<User> getUser(@Path("id") int id);
}

Retrofit retrofit = new Retrofit.Builder()
       .baseUrl("https://api.example.com/")
       .addConverterFactory(GsonConverterFactory.create()) // 添加数据转换工厂
       .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 添加适配器工厂
       .build();

ApiService apiService = retrofit.create(ApiService.class);
Observable<User> observable = apiService.getUser(1);

代码解释：

GsonConverterFactory 负责将响应体转换为 User 对象，RxJava2CallAdapterFactory 负责将 Call 对象转换为 Observable 对象。

8.2 与 OkHttp 模块的交互

Retrofit 底层使用 OkHttp 进行网络请求，适配器模块和 OkHttp 模块在请求和响应处理上进行交互。OkHttp 负责发送请求和接收响应，适配器模块负责将 Call 对象转换为其他类型。

java

final class OkHttpCall<T> implements Call<T> {
    // ... 其他代码 ...

    @Override
    public void enqueue(final Callback<T> callback) {
        // 检查回调是否为空
        checkNotNull(callback, "callback == null");

        delegate.enqueue(new Callback<T>() {
            @Override
            public void onResponse(Call<T> call, final Response<T> response) {
                if (call.isCanceled()) return;
                // 调用适配器的逻辑，可能会将响应传递给观察者
                callback.onResponse(OkHttpCall.this, response);
            }

            @Override
            public void onFailure(Call<T> call, final Throwable t) {
                if (call.isCanceled()) return;
                // 调用适配器的逻辑，可能会将异常传递给观察者
                callback.onFailure(OkHttpCall.this, t);
            }
        });
    }

    // ... 其他代码 ...
}

代码解释：

在 OkHttpCall 的 enqueue 方法中，当请求成功或失败时，会调用回调的 onResponse 或 onFailure 方法，适配器模块可以在这些回调中进行相应的处理，例如将响应或异常传递给观察者。

九、适配器模块的未来发展趋势

9.1 支持更多的异步编程模型

随着 Android 开发中异步编程模型的不断发展，Retrofit 的适配器模块可能会支持更多的异步编程模型，如 Kotlin 的协程、Java 的 CompletableFuture 等。

9.2 更好的性能优化

未来的适配器模块可能会进行更多的性能优化，例如减少内存开销、提高并发处理能力等。可以通过使用更高效的数据结构和算法来实现这些优化。

9.3 与更多的框架集成

Retrofit 可能会与更多的框架进行集成，如 RxJava 3、Flow 等，以提供更好的开发体验。适配器模块可以提供相应的适配器工厂，方便开发者在不同的框架之间进行切换。

9.4 增强的错误处理和调试功能

适配器模块可能会提供更详细的错误信息和调试工具，帮助开发者更快地定位和解决问题。例如，在异常信息中提供更多的上下文信息，或者提供可视化的调试工具。

十、总结

Retrofit 的适配器模块是一个非常重要的模块，它负责将 Call 对象转换为其他类型，从而实现不同的异步编程模式。通过对适配器模块的源码分析，我们了解了其核心概念、常见的适配器工厂实现、工作流程、异常处理、性能优化、与其他模块的交互以及未来发展趋势。掌握这些知识可以帮助我们更好地使用和扩展 Retrofit 框架，提高开发效率和应用性能。在实际开发中，我们可以根据具体的需求选择合适的适配器工厂，进行自定义适配器的开发，以及对适配器模块进行性能优化，以满足不同的业务需求。同时，我们也需要关注适配器模块的未来发展趋势，以便及时跟上技术的发展步伐。

以上的分析涵盖了 Retrofit 适配器模块的各个方面，从核心概念到源码实现，从异常处理到性能优化，以及与其他模块的交互和未来发展趋势。通过深入理解这些内容，开发者可以更好地利用 Retrofit 的适配器模块来构建高效、灵活的网络请求系统。

请注意，Retrofit 是一个不断发展的框架，以上分析基于其常见的实现和版本，实际情况可能会因版本更新而有所不同。在使用时，建议参考官方文档和最新的源码。