generalsの学习笔记

Appearance

OkHttp 与 Retrofit 原理深入详解 ​

更新: 5/15/2026 字数: 0 字 时长: 0 分钟

1. OkHttp 整体架构 ​

1.1 核心类关系 ​

OkHttpClient（配置中心）
├── Dispatcher（调度器：管理异步请求的线程池）
├── ConnectionPool（连接池：复用 TCP 连接）
├── Interceptors（应用拦截器列表）
├── NetworkInterceptors（网络拦截器列表）
└── Cache（HTTP 缓存）

请求流程：
Call → RealCall → getResponseWithInterceptorChain() → 拦截器链

1.2 Dispatcher 调度器 ​

java
public final class Dispatcher {
    private int maxRequests = 64;           // 最大并发请求数
    private int maxRequestsPerHost = 5;     // 每个主机最大并发数
    private ExecutorService executorService; // 线程池

    // 三个队列
    private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();  // 等待执行
    private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>(); // 正在执行（异步）
    private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();   // 正在执行（同步）

    // 异步请求入队
    void enqueue(AsyncCall call) {
        synchronized (this) {
            readyAsyncCalls.add(call);
        }
        promoteAndExecute(); // 尝试将 ready 移到 running
    }

    private boolean promoteAndExecute() {
        for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
            AsyncCall asyncCall = i.next();

            if (runningAsyncCalls.size() >= maxRequests) break;        // 达到最大并发
            if (asyncCall.callsPerHost().get() >= maxRequestsPerHost) continue; // 达到单主机上限

            i.remove();
            asyncCall.callsPerHost().incrementAndGet();
            runningAsyncCalls.add(asyncCall);
            executorService().execute(asyncCall); // 提交到线程池
        }
    }
}

线程池配置：

java
// 核心线程数 0，最大线程数 Integer.MAX_VALUE，空闲 60 秒回收
// 使用 SynchronousQueue：不缓存任务，来一个执行一个
new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
    new SynchronousQueue<>(), threadFactory);

2. 拦截器链源码分析 ​

2.1 责任链模式 ​

java
// RealCall.getResponseWithInterceptorChain()
Response getResponseWithInterceptorChain() throws IOException {
    List<Interceptor> interceptors = new ArrayList<>();

    interceptors.addAll(client.interceptors());         // 1. 应用拦截器（用户添加的）
    interceptors.add(new RetryAndFollowUpInterceptor()); // 2. 重试和重定向
    interceptors.add(new BridgeInterceptor());           // 3. 桥接（补全请求头）
    interceptors.add(new CacheInterceptor());            // 4. 缓存
    interceptors.add(new ConnectInterceptor());          // 5. 建立连接
    interceptors.addAll(client.networkInterceptors());   // 6. 网络拦截器（用户添加的）
    interceptors.add(new CallServerInterceptor());       // 7. 发送请求/读取响应

    Interceptor.Chain chain = new RealInterceptorChain(interceptors, 0, request, ...);
    return chain.proceed(request);
}
java
// RealInterceptorChain.proceed()
public Response proceed(Request request) throws IOException {
    // 创建下一个 Chain，index + 1
    RealInterceptorChain next = new RealInterceptorChain(
        interceptors, index + 1, request, ...);

    // 取出当前拦截器
    Interceptor interceptor = interceptors.get(index);

    // 调用拦截器，传入 next chain
    Response response = interceptor.intercept(next);

    return response;
}

2.2 各拦截器详解 ​

RetryAndFollowUpInterceptor：

java
@Override
public Response intercept(Chain chain) throws IOException {
    while (true) {
        try {
            response = chain.proceed(request);
        } catch (RouteException e) {
            // 路由异常，尝试恢复
            if (!recover(e.getLastConnectException(), false)) throw e;
            continue; // 重试
        }

        // 处理重定向（301/302/307/308）
        Request followUp = followUpRequest(response);
        if (followUp == null) return response; // 不需要重定向

        if (++followUpCount > MAX_FOLLOW_UPS) throw new ProtocolException("Too many redirects");
        request = followUp; // 用重定向的 URL 重新请求
    }
}

BridgeInterceptor：

java
// 补全请求头
@Override
public Response intercept(Chain chain) throws IOException {
    Request.Builder requestBuilder = userRequest.newBuilder();

    // 自动添加 Content-Type
    if (body != null) {
        requestBuilder.header("Content-Type", body.contentType().toString());
        requestBuilder.header("Content-Length", Long.toString(body.contentLength()));
    }

    // 自动添加 Host
    requestBuilder.header("Host", userRequest.url().host());

    // 自动添加 Accept-Encoding: gzip
    if (userRequest.header("Accept-Encoding") == null) {
        requestBuilder.header("Accept-Encoding", "gzip");
    }

    // 自动添加 Cookie
    List<Cookie> cookies = cookieJar.loadForRequest(userRequest.url());
    if (!cookies.isEmpty()) {
        requestBuilder.header("Cookie", cookieHeader(cookies));
    }

    Response response = chain.proceed(requestBuilder.build());

    // 自动解压 gzip
    if ("gzip".equalsIgnoreCase(response.header("Content-Encoding"))) {
        response = response.newBuilder()
            .body(new RealResponseBody(contentType, -1, Okio.buffer(new GzipSource(source))))
            .build();
    }
    return response;
}

CacheInterceptor：

java
@Override
public Response intercept(Chain chain) throws IOException {
    Response cacheCandidate = cache != null ? cache.get(chain.request()) : null;

    CacheStrategy strategy = new CacheStrategy.Factory(now, request, cacheCandidate).compute();
    Request networkRequest = strategy.networkRequest;   // null 表示不需要网络
    Response cacheResponse = strategy.cacheResponse;    // null 表示没有可用缓存

    // 情况1：不需要网络，有缓存 → 直接返回缓存（强缓存命中）
    if (networkRequest == null && cacheResponse != null) {
        return cacheResponse;
    }

    // 情况2：需要网络
    Response networkResponse = chain.proceed(networkRequest);

    // 情况3：304 Not Modified → 用缓存 body + 网络 header
    if (networkResponse.code() == 304 && cacheResponse != null) {
        Response response = cacheResponse.newBuilder()
            .headers(combine(cacheResponse.headers(), networkResponse.headers()))
            .build();
        cache.update(cacheResponse, response);
        return response;
    }

    // 情况4：200 → 更新缓存，返回网络响应
    cache.put(networkResponse);
    return networkResponse;
}

ConnectInterceptor：

java
@Override
public Response intercept(Chain chain) throws IOException {
    RealInterceptorChain realChain = (RealInterceptorChain) chain;

    // 从连接池获取或创建新连接
    Exchange exchange = realChain.call().initExchange(chain);

    return realChain.proceed(request, exchange);
}

2.3 应用拦截器 vs 网络拦截器 ​

请求流程：
  应用拦截器 → Retry → Bridge → Cache → Connect → 网络拦截器 → CallServer

                                    ↑ 缓存命中时到这里就返回了
维度应用拦截器网络拦截器
位置最外层Connect 之后
调用次数只调用一次重定向时多次调用
能否短路可以（不调用 proceed）可以
看到的 Request原始请求Bridge 补全后的请求
看到的 Response最终响应原始网络响应（未解压 gzip）
缓存命中时仍然执行不执行
适用场景日志、公共参数、Token网络监控、流量统计

3. 连接池详解 ​

3.1 连接复用 ​

java
public final class ConnectionPool {
    // 默认：最多 5 个空闲连接，空闲 5 分钟回收
    private final int maxIdleConnections;
    private final long keepAliveDurationNs;
    private final Deque<RealConnection> connections = new ArrayDeque<>();

    // 清理线程
    private final Runnable cleanupRunnable = () -> {
        while (true) {
            long waitNanos = cleanup(System.nanoTime());
            if (waitNanos == -1) return; // 没有连接了
            wait(waitNanos); // 等待下次清理
        }
    };
}

连接复用的条件：

java
// RealConnection.isEligible()
boolean isEligible(Address address, List<Route> routes) {
    // 1. 连接没有达到最大并发流数（HTTP/2 默认 256）
    if (allocationLimit <= allocations.size()) return false;

    // 2. 地址匹配（host, port, proxy, ssl 等）
    if (!address.equalsNonHost(this.route.address())) return false;

    // 3. 主机名匹配
    if (address.url().host().equals(this.route().address().url().host())) {
        return true; // 完全匹配
    }

    // 4. HTTP/2 连接合并（不同主机但同一 IP + 证书覆盖）
    if (this.http2Connection != null) {
        // 检查证书是否覆盖目标主机
        return route.address().hostnameVerifier().verify(address.url().host(), handshake);
    }

    return false;
}

3.2 连接清理 ​

java
long cleanup(long now) {
    int idleConnectionCount = 0;
    RealConnection longestIdleConnection = null;
    long longestIdleDurationNs = Long.MIN_VALUE;

    for (RealConnection connection : connections) {
        // 统计空闲连接
        if (connection.allocations.isEmpty()) {
            idleConnectionCount++;
            long idleDurationNs = now - connection.idleAtNanos;
            if (idleDurationNs > longestIdleDurationNs) {
                longestIdleDurationNs = idleDurationNs;
                longestIdleConnection = connection;
            }
        }
    }

    if (longestIdleDurationNs >= keepAliveDurationNs
        || idleConnectionCount > maxIdleConnections) {
        // 超时或超量，移除最久空闲的连接
        connections.remove(longestIdleConnection);
        longestIdleConnection.socket().close();
        return 0; // 立即再次清理
    }

    if (idleConnectionCount > 0) {
        // 有空闲连接，等到最久的那个超时
        return keepAliveDurationNs - longestIdleDurationNs;
    }

    return -1; // 没有空闲连接
}

4. Retrofit 动态代理详解 ​

4.1 create() 源码 ​

java
public <T> T create(final Class<T> service) {
    // 验证接口
    validateServiceInterface(service);

    return (T) Proxy.newProxyInstance(
        service.getClassLoader(),
        new Class<?>[] { service },
        new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // Object 的方法直接调用
                if (method.getDeclaringClass() == Object.class) {
                    return method.invoke(this, args);
                }

                // 加载 ServiceMethod（解析注解，缓存结果）
                return loadServiceMethod(method).invoke(args);
            }
        });
}

4.2 ServiceMethod 解析注解 ​

java
// ServiceMethod.parseAnnotations()
static <T> ServiceMethod<T> parseAnnotations(Retrofit retrofit, Method method) {
    // 1. 解析方法注解（@GET, @POST, @Headers 等）
    RequestFactory requestFactory = RequestFactory.parseAnnotations(retrofit, method);

    // 2. 获取返回类型
    Type returnType = method.getGenericReturnType();

    // 3. 创建 CallAdapter（将 Call 适配为目标类型）
    CallAdapter<ResponseT, ReturnT> callAdapter = createCallAdapter(retrofit, returnType);

    // 4. 创建 Converter（将 ResponseBody 转为数据类）
    Converter<ResponseBody, ResponseT> responseConverter = createResponseConverter(retrofit, responseType);

    return new HttpServiceMethod<>(requestFactory, callAdapter, responseConverter);
}

4.3 RequestFactory 解析注解 ​

java
// 解析 @GET("/users/{id}")
// 解析 @Query("page")
// 解析 @Body
// 解析 @Path("id")
// 解析 @Header("Authorization")

// 最终构建 OkHttp Request
Request toRequest(Object[] args) {
    RequestBuilder requestBuilder = new RequestBuilder(
        httpMethod, baseUrl, relativeUrl, headers, contentType, ...);

    // 将方法参数填入对应位置
    for (int i = 0; i < args.length; i++) {
        parameterHandlers[i].apply(requestBuilder, args[i]);
        // PathParameterHandler: 替换 URL 中的 {id}
        // QueryParameterHandler: 添加 ?page=1
        // BodyParameterHandler: 序列化为 RequestBody
        // HeaderParameterHandler: 添加请求头
    }

    return requestBuilder.build();
}

4.4 suspend 函数支持 ​

java
// Retrofit 检测 suspend 函数
// suspend fun getUser(): User
// 编译后变成: fun getUser(continuation: Continuation<User>): Any

if (isKotlinSuspendFunction) {
    // 最后一个参数是 Continuation
    return new SuspendForBody<>(requestFactory, callFactory, responseConverter, ...);
}

// SuspendForBody.invoke()
@Override
Object invoke(Object[] args) {
    // 取出 Continuation
    Continuation<ResponseT> continuation = (Continuation<ResponseT>) args[args.length - 1];

    Call<ResponseT> call = new OkHttpCall<>(requestFactory, args, responseConverter);

    // 将 OkHttp 的异步回调转为协程挂起
    return KotlinExtensions.awaitResponse(call, continuation);
}

// KotlinExtensions.kt
suspend fun <T> Call<T>.await(): T {
    return suspendCancellableCoroutine { continuation ->
        continuation.invokeOnCancellation { cancel() }

        enqueue(object : Callback<T> {
            override fun onResponse(call: Call<T>, response: Response<T>) {
                if (response.isSuccessful) {
                    continuation.resume(response.body()!!)
                } else {
                    continuation.resumeWithException(HttpException(response))
                }
            }

            override fun onFailure(call: Call<T>, t: Throwable) {
                continuation.resumeWithException(t)
            }
        })
    }
}

4.5 Converter 和 CallAdapter ​

kotlin
// 自定义 Converter：处理特殊的响应格式
class ApiResponseConverter<T>(
    private val delegate: Converter<ResponseBody, T>
) : Converter<ResponseBody, T> {
    override fun convert(value: ResponseBody): T? {
        // 先解析外层包装
        val wrapper = JSON.parse(value.string())
        if (wrapper.code != 0) {
            throw ApiException(wrapper.code, wrapper.message)
        }
        // 再解析 data 字段
        return delegate.convert(ResponseBody.create(null, wrapper.data.toString()))
    }
}

// 自定义 CallAdapter：将 Call<T> 适配为 Flow<T>
class FlowCallAdapter<T>(private val responseType: Type) : CallAdapter<T, Flow<T>> {
    override fun responseType() = responseType

    override fun adapt(call: Call<T>): Flow<T> = flow {
        val response = call.awaitResponse()
        if (response.isSuccessful) {
            emit(response.body()!!)
        } else {
            throw HttpException(response)
        }
    }
}