大纲

• 前言
• 概述
• 基本用法
• 基础篇小结
• 源码篇
• 源码篇小结
• 思考
• 参考资料

前言

文章是跟这 OkHttp 的源码去读的，结合了网上其它的总结资料，写完后篇幅确实比 Retrofit 长… 先说一下本人的源码阅读流程，还是根据调用开始阅读，从构建到调用，开始从同步调用的方式进入，到责任链，再返回到结束。然后再从异步调用的方式进入，发现同步和异步的区别是在进入责任链之前就区分好了，方法入口不一样，到责任链环节的时候流程大致是一致的，但异步调用的话，责任链的返回时通过回调的方式而已。所以本篇在源码篇上的排版顺序调整了一下（原先是按照我阅读的方式同步调用 -> 结束，异步调用 -> 结束），将两种调用方式先放到前面，然后将责任链的流程按每个拦截器分小点，方便阅读。

概述

HTTP 是现代应用网络请求用于交换数据和媒体资源的方式，有效执行 HTTP 可以让内容家在速度更快且能节省带宽。

OkHttp 则是一个高效的 HTTP 客户端，它具备以下特点

• HTTP/2 支持允许对同一主机的所有请求共享一个套接字
• 连接池减少请求延迟（如果 HTTP/2 不可用）
• 透明 GZIP 可缩小下载大小
• 响应缓存完全避免网络重复请求

OkHttp 在网络出现问题时，会从常见的连接问题中默默恢复。如果服务存在多个 IP 地址，那么 OkHttp 会在第一次连接失败时尝试备用地址。这对于在冗余数据中心托管的 IPv4+IPv6 和服务是必需的。OkHttp 支持现在 TLS 功能（TLS 1.3、ALPN、证书锁定）。可以配置为回退以实现广泛的连接。

OkHttp 的请求/响应 API 设计有流畅的构建器和不变性。支持同步阻塞调用和带有回调的异步调用。（依赖于 Okio 和 Kotlin 标准库）

硬生生地翻译了一下官方文档… 大概了解到 OkHttp 的基本作用，接下来看看它怎么用吧

基本用法

依赖引入: implementation("com.squareup.okhttp3:okhttp:4.9.0")

获取一个 URL 的内容

public class GetExample {
  // 创建 OkHttpClient
  OkHttpClient client = new OkHttpClient();

  String run(String url) throws IOException {
    // 创建 Request
    Request request = new Request.Builder()
        .url(url)
        .build();

    // client.newCall 发起请求（很熟悉吧，在 Retrofit 中也通过 Call 发起请求）
    try (Response response = client.newCall(request).execute()) {
      // 返回结果，string() 调用一次后，会将流关闭
      return response.body().string();
    }
  }

  public static void main(String[] args) throws IOException {
    GetExample example = new GetExample();
    String response = example.run("https://raw.github.com/square/okhttp/master/README.md");
    System.out.println(response);
  }
}

往服务器推送数据

// 指定请求的 MediaType
public static final MediaType JSON
    = MediaType.get("application/json; charset=utf-8");

// 构造 client
OkHttpClient client = new OkHttpClient();

String post(String url, String json) throws IOException {
  // 构造 RequestBody（参数）
  RequestBody body = RequestBody.create(JSON, json);
  // 构造 Request
  Request request = new Request.Builder()
      .url(url)
      .post(body)
      .build();
  // 发起请求
  try (Response response = client.newCall(request).execute()) {
    // 返回结果
    return response.body().string();
  }
}

上面两个例子，属于同步请求，会进入阻塞状态并等待直接结果返回，下面看一个异步请求

//1 请求的 Client
val okHttpClient = OkHttpClient()
//2 构造出一个 Request 对象
val request = Request.Builder()
    //API 接口由 wanandroid.com 提供
    .url("https://wanandroid.com/wxarticle/chapters/json")
    .get()
    .build()
//3 创建出一个执行的 Call 对象
val call = okHttpClient.newCall(request)
//4 异步执行请求
call.enqueue(object : Callback {
    override fun onFailure(call: Call, e: IOException) {
        // 失败回调
        e.printStackTrace()
    }

    override fun onResponse(call: Call, response: Response) {
        // 成功回调
        "response.code = ${response.code}".log()
    }
})

基础篇小结

看过上面的 demo，OkHttp 的使用就是这样简简单单，先构造一个 OkHttpClient，还得有一个请求体 Request，如果有参数就构造一个 RequestBody（要指定 MediaType，网络请求的基础），再通过 OkHttpClient 创建一个 Call 实例（client.newCall），同步的话就调用 execute()，异步的话就调用 enqueue()，拿到 Response，那这一次的请求就完成了。

单从基本用法上，看不出网络请求的详细信息，不是有请求方法的区分吗？不是有超时机制吗？该怎么配置？同步异步的切换是怎么做到的…

那就看看他怎么是实现的吧

源码篇

OkHttpClient

按照习惯，从调用的地方开始

OkHttpClient client = new OkHttpClient();

OkHttpClient.kt

// OkHttpClient 无参构造方法直接通过 Builder 创建出来
constructor() : this(Builder())

OkHttpClient 的实例也是通过构建者模式创建，Builder 中包含了一系列配置

Builder

// 从 Builder 中可以了解到如果想要配置参数，在创建的时候指定行
class Builder constructor() {
    // 分发器，用于何时执行请求的策略
    internal var dispatcher: Dispatcher = Dispatcher()
    // 连接池，管理 HTTP 和 HTTP/2 连接的重用以减少网络延迟
    // 相同的 HTTP 地址肯能会共享同一个连接
    internal var connectionPool: ConnectionPool = ConnectionPool()
    // 拦截器列表，用于观察、修改和短链请求和相应的返回响应
    // 通常拦截器会添加、删除或转换请求或响应的头部信息
    internal val interceptors: MutableList<Interceptor> = mutableListOf()
    internal val networkInterceptors: MutableList<Interceptor> = mutableListOf()
    // 指标事件监听器，用于监控应用的 HTTP 调用数量、大小和持续时间（工厂模式）
    internal var eventListenerFactory: EventListener.Factory = EventListener.NONE.asFactory()
    // 连接失败后是否尝试重连
    internal var retryOnConnectionFailure = true
    // 服务器身份验证
    internal var authenticator: Authenticator = Authenticator.NONE
    // 是否允许重定向
    internal var followRedirects = true
    // 是否允许 ssl 重定向
    internal var followSslRedirects = true
    // HTTP Cookie
    internal var cookieJar: CookieJar = CookieJar.NO_COOKIES
    // HTTP 缓存（存到文件中以便重用，从而节省时间和带宽）
    internal var cache: Cache? = null
    // 域名管理系统
    internal var dns: Dns = Dns.SYSTEM
    // 代理设置（通常是 http 或者 socks 和一个套接字地址）
    internal var proxy: Proxy? = null
    // 代理选择器
    internal var proxySelector: ProxySelector? = null
    // 代理服务器身份验证
    internal var proxyAuthenticator: Authenticator = Authenticator.NONE
    // socket 工厂
    internal var socketFactory: SocketFactory = SocketFactory.getDefault()
    // ssl socket 工厂
    internal var sslSocketFactoryOrNull: SSLSocketFactory? = null
    // ssl 握手异常管理
    internal var x509TrustManagerOrNull: X509TrustManager? = null
    // 传输层版本和连接协议（默认支持 HTTP 和 HTTPS
    internal var connectionSpecs: List<ConnectionSpec> = DEFAULT_CONNECTION_SPECS
    // HTTP 协议（默认 HTTP_2, HTTP_1_1）
    internal var protocols: List<Protocol> = DEFAULT_PROTOCOLS
    // 主机名字确认
    internal var hostnameVerifier: HostnameVerifier = OkHostnameVerifier
    // 证书链
    internal var certificatePinner: CertificatePinner = CertificatePinner.DEFAULT
    // 证书清理链
    internal var certificateChainCleaner: CertificateChainCleaner? = null
    // 默认呼叫超时时间内（毫秒），默认情况下完整调用没有超时
    // 主要用于调用中的连接、写入和读取操作
    internal var callTimeout = 0
    // 连接超时（默认 10 秒）超时时间在创建的时候就指定了
    internal var connectTimeout = 10_000
    // 读取超时（默认 10 秒）
    internal var readTimeout = 10_000
    // 写入超时（默认 10 秒）
    internal var writeTimeout = 10_000
    // web sockt 和 HTTP/2 ping 操作的间隔（单位: 毫秒）
    // 默认情况下不发送 ping
    internal var pingInterval = 0
    // web sockt 最小应该被压缩的值
    internal var minWebSocketMessageToCompress = RealWebSocket.DEFAULT_MINIMUM_DEFLATE_SIZE
    // 路由黑名单，如果尝试连接到一个特定的 IP 或代理服务器出现错误
    // 那么该路由信息将被记录作为备用路由
    internal var routeDatabase: RouteDatabase? = null

    // ...
}

说实话，有点劝退，Builder 里的属性很多，光每个点击去看看注释都花了小半小时，但是了解框架还是先看主流程，把开始对这个库的疑云扫清，其它细节可以慢慢看… 既然 OkHttpClient 的创建看完，那就看看 Request 吧（毕竟是第二步）

Request

/**
 * 相比之下这个类就简单一些
 * url 请求的 url
 * method 请求的方法默认是 GET（方法在这里指定）
 * headers 头部信息（键值对），它的键值对用 List 存取，有点东西...
 * body 请求体
 * 在使用的过程中，按需传入
 */
class Request internal constructor(
  @get:JvmName("url") val url: HttpUrl,
  @get:JvmName("method") val method: String,
  @get:JvmName("headers") val headers: Headers,
  @get:JvmName("body") val body: RequestBody?,
  internal val tags: Map<Class<*>, Any>
) {
  // ...
}

Request 的构建比较常规，都是一些基本的请求参数，值得注意的是 Header 的存储虽说在平时传入的时候是以键值对的形式传入，但他内部有将键值对转换成字符串 List 的逻辑，所以能在请求是 Header 写入是一串字符串。接下来就是在 Retrofit 中看到过的 newCall 操作，OkHttpClient 实现了 Call.Factory 接口，重写返回了 RealCall 的实例，代码如下

OkHttpClient.kt

// 也就是说 execute 和 enqueue 的逻辑都在 RealCall 中，那么就看看这里的代码吧
/** Prepares the [request] to be executed at some point in the future. */
override fun newCall(request: Request): Call = RealCall(this, request, forWebSocket = false)

剩下的，就是同步发起请求和异步发起请求了，来看看 RealCall 这个类中做了什么。按顺序

execute()，此方法前置知识点 Kotlin 契约 和 AtomicBoolean（采用 CAS 方法，高并发下只有一个线程能访问该属性值）

enqueue()，需要了解一下 ThreadPoolExecutor

同步请求

RealCall.kt

/**
 * client 请求 client
 * originalRequest 请求参数
 * forWebScoket 默认不是 web socket 请求
 * 通过 newCall 后拿到 Call 的实例，分为同步请求 execute(), 异步请求 enqueue()
 */
class RealCall(
  val client: OkHttpClient,
  /** The application's original request unadulterated by redirects or auth headers. */
  val originalRequest: Request,
  val forWebSocket: Boolean
) : Call {
  // ...
  
  /**
   * 同步请求方法
   * 这方法没有入参，返回值是 Response
   * 这个方法在做了一些前置逻辑后，实际的执行在 callStart() 后
   * 通过 Dispatcher 来执行请求，那么去 Dispatcher 中看看相关的代码
   */
  override fun execute(): Response {
    // 检查 executed 这个变量是否可以从 false 改为 true，不能就抛出异常
    check(executed.compareAndSet(false, true)) { "Already Executed" }

    // timeout 是 AsyncTimeout，在后台线程中准确执行超时操作
    timeout.enter()
    // 事件回调，请求开始
    callStart()
    try {
      // 调用 dispatcher#executed 将 RealCall 传禁区
      client.dispatcher.executed(this)
      // 从拦截链获取 Response 并返回（就是服务器返回的响应）
      return getResponseWithInterceptorChain()
    } finally {
      // 结束请求
      client.dispatcher.finished(this)
    }
  }
}

Dispatcher.kt

/**
 * 是个普通的类
 */
class Dispatcher constructor() {
  
  // ...
  
  /**
   * 双向队列，线程不安全，容量不足会自动扩容
   * 每添加一个元素都会将其加到数组的尾部，head 指针不变，tail 指针加一
   * 因为指针是循环家的，所以当 tail 追上 head 时会进行扩容（扩大一倍）
   * (this.tail = this.tail + 1 & this.elements.length - 1) == this.head
   * 这个队列使用记录正在运行的同步请求（包括已经取消和没有完成的）
   */
  /** Running synchronous calls. Includes canceled calls that haven't finished yet. */
  private val runningSyncCalls = ArrayDeque<RealCall>()
  
  // ...
  
  /**
   * executed 方法将 RealCall 实例放到了一个双向队列中
   * @Synchronized 跟 Java 中 synchronized 关键字一样的作用
   * 没错 Kotlin 把 Java 中并发相关的关键字都去掉了，使用注解代替
   * 可是 executed() 方法仅是将 call 放到队列中
   * 所以还得看看怎么拿到真正的 Response
   * 还得回到 RealCall 中看看 getResponseWithInterceptorChain
   */
  /** Used by [Call.execute] to signal it is in-flight. */
  @Synchronized internal fun executed(call: RealCall) {
    runningSyncCalls.add(call)
  }
}

异步请求

接下来再看看异步调用 enqueue()

RealCall.kt

override fun enqueue(responseCallback: Callback) {
  check(executed.compareAndSet(false, true)) { "Already Executed" }

  // 回调监听
  callStart()
  // 调用 dispatcher#enqueue，入为 AsyncCall 并带上了一个 responseCallback
  client.dispatcher.enqueue(AsyncCall(responseCallback))
}

Dispatcher.kt

internal fun enqueue(call: AsyncCall) {
  synchronized(this) {
    // 将 AsyncCall 实例添加到 readyAsyncCalls list 中
    readyAsyncCalls.add(call)

    // Mutate the AsyncCall so that it shares the AtomicInteger of an existing running call to
    // the same host.
    if (!call.call.forWebSocket) {
      val existingCall = findExistingCallWithHost(call.host)
      if (existingCall != null) call.reuseCallsPerHostFrom(existingCall)
    }
  }
  // 执行
  promoteAndExecute()
}

private fun promoteAndExecute(): Boolean {
  this.assertThreadDoesntHoldLock()

  // 声明 AsyncCall 集合（表示可以被执行的请求）
  val executableCalls = mutableListOf<AsyncCall>()
  val isRunning: Boolean
  synchronized(this) {
    // 开始遍历 readyAsyncCalls 集合
    val i = readyAsyncCalls.iterator()
    while (i.hasNext()) {
      val asyncCall = i.next()

      if (runningAsyncCalls.size >= this.maxRequests) break // Max capacity.
      if (asyncCall.callsPerHost.get() >= this.maxRequestsPerHost) continue // Host max capacity.

      i.remove()
      // 将拿到 AsyncCall 存到 executableCalls 集合中
      asyncCall.callsPerHost.incrementAndGet()
      executableCalls.add(asyncCall)
      runningAsyncCalls.add(asyncCall)
    }
    isRunning = runningCallsCount() > 0
  }

  // 遍历 executableCalls 集合并调用每个 asyncCall#executeOn()
  // executorService 是 ExecutorService 类型的线程池，用于执行后台任务
  // 没有核心线程，非核心线程不限，闲置的时候，一分钟后会被回收（ThreadPoolExecutor）
  for (i in 0 until executableCalls.size) {
    val asyncCall = executableCalls[i]
    asyncCall.executeOn(executorService)
  }

  return isRunning
}

RealCall.kt（AsyncCall 是 RealCall 的 inner class）

/**
 * Attempt to enqueue this async call on [executorService]. This will attempt to clean up
 * if the executor has been shut down by reporting the call as failed.
 */
fun executeOn(executorService: ExecutorService) {
  client.dispatcher.assertThreadDoesntHoldLock()

  var success = false
  try {
    // 执行 AsyncCall（是个 Runnable）
    executorService.execute(this)
    success = true
  } catch (e: RejectedExecutionException) {
    val ioException = InterruptedIOException("executor rejected")
    ioException.initCause(e)
    noMoreExchanges(ioException)
    // 线程池任务满了，拒绝执行，抛异常直接回调失败
    responseCallback.onFailure(this@RealCall, ioException)
  } finally {
    if (!success) {
      client.dispatcher.finished(this) // This call is no longer running!
    }
  }
}

override fun run() {
  threadName("OkHttp ${redactedUrl()}") {
    var signalledCallback = false
    timeout.enter()
    try {
      // 也是通过责任链 getResponseWithInterceptorChain
      val response = getResponseWithInterceptorChain()
      signalledCallback = true
      // 通过 responseCallback 将 response 返回
      responseCallback.onResponse(this@RealCall, response)
    } catch (e: IOException) {
      if (signalledCallback) {
        // Do not signal the callback twice!
        Platform.get().log("Callback failure for ${toLoggableString()}", Platform.INFO, e)
      } else {
        // 异常回调
        responseCallback.onFailure(this@RealCall, e)
      }
    } catch (t: Throwable) {
      cancel()
      if (!signalledCallback) {
        val canceledException = IOException("canceled due to $t")
        canceledException.addSuppressed(t)
        // 异常回调
        responseCallback.onFailure(this@RealCall, canceledException)
      }
      throw t
    } finally {
      client.dispatcher.finished(this)
    }
  }
}

可以看到，异步调用是通过线程池执行 AsyncCall 这个 Runnable 开启责任链的，然后责任链拿到返回结果后再通过 responseCallback 以回调的形式将结果返回。同步调用没有开线程去请求，如果在 Android 使用的话，需要手动开个线程去执行。

责任链环节

同步请求和异步请求最终都会调用 getResponseWithInterceptorChain 进入到责任链环节建立连接并发起请求

RealCall.kt

// ...

/**
 * 这个方法中使用了传入的拦截器和一些默认拦截器
 * 然后构造出责任链实例进行处理（使用到责任链模式）
 * 每个拦截器负责相应的功能，整个请求过程就是通过一个个拦截器来使得请求完整
 * 并在服务器返回时，经过一个个拦截器处理后返回 Response
 */
@Throws(IOException::class)
internal fun getResponseWithInterceptorChain(): Response {
  // Build a full stack of interceptors.
  // 创建一个拦截器集合
  val interceptors = mutableListOf<Interceptor>()
  // 添加自定义的拦截器（在创建 OkHttpClient 时可以传入）
  // 可以配置公共参数，在请求之前
  interceptors += client.interceptors
  // 添加重试与重定向拦截器
  // 网络请求出错或服务器返回 301、302，会自动进行重定向
  interceptors += RetryAndFollowUpInterceptor(client)
  // 添加桥拦截器
  // 拼接成一个标准的 Http 协议请求，请求行，Header，Body 等
  interceptors += BridgeInterceptor(client.cookieJar)
  // 添加缓存拦截器
  // 根据 Header 进行网络缓存
  interceptors += CacheInterceptor(client.cache)
  // 添加连接拦截器
  // 开启一个目标服务器的连接
  interceptors += ConnectInterceptor
  if (!forWebSocket) {
    // 添加自定义的网络拦截器
    // 在请求结果返回时，对接口数据进行处理，如日志
    interceptors += client.networkInterceptors
  }
  // 添加服务器请求拦截器，请求服务器
  interceptors += CallServerInterceptor(forWebSocket)

  // 构建责任链
  // call 当前 RealCall 实例
  // interceptors 当前拦截器集合
  // index 当前拦截器集合索引
  // request 原始请求参数（未经过拦截器处理的）
  // 剩下的三个是超时时间（都有默认值）
  val chain = RealInterceptorChain(
      call = this,
      interceptors = interceptors,
      index = 0,
      exchange = null,
      request = originalRequest,
      connectTimeoutMillis = client.connectTimeoutMillis,
      readTimeoutMillis = client.readTimeoutMillis,
      writeTimeoutMillis = client.writeTimeoutMillis
  )

  var calledNoMoreExchanges = false
  try {
    // 处理责任链中的拦截器
    val response = chain.proceed(originalRequest)
    // 异常判断
    if (isCanceled()) {
      response.closeQuietly()
      throw IOException("Canceled")
    }
    return response
  } catch (e: IOException) {
    calledNoMoreExchanges = true
    throw noMoreExchanges(e) as Throwable
  } finally {
    if (!calledNoMoreExchanges) {
      noMoreExchanges(null)
    }
  }
}

//...

RealInterceptorChain

与远程 web 服务器的连接，能够承载 1 个或多个并发流。

RealInterceptorChain.kt

/**
 * Interceptor.Chain 的实现类
 */
class RealInterceptorChain(
  internal val call: RealCall,
  private val interceptors: List<Interceptor>,
  private val index: Int,
  internal val exchange: Exchange?,
  internal val request: Request,
  internal val connectTimeoutMillis: Int,
  internal val readTimeoutMillis: Int,
  internal val writeTimeoutMillis: Int
) : Interceptor.Chain {
  
  // ...
  
  @Throws(IOException::class)
  override fun proceed(request: Request): Response {
    check(index < interceptors.size)

    calls++

    if (exchange != null) {
      check(exchange.finder.sameHostAndPort(request.url)) {
        "network interceptor ${interceptors[index - 1]} must retain the same host and port"
      }
      check(calls == 1) {
        "network interceptor ${interceptors[index - 1]} must call proceed() exactly once"
      }
    }

    // 1、通过 copy() 拿到下一个 RealInterceptorChain 实例
    // copy 函数代码贴在下面
    // Call the next interceptor in the chain.
    val next = copy(index = index + 1, request = request)
    // 2、获取当前拦截器
    val interceptor = interceptors[index]

    // 3、调用当前拦截器的 intercept()，入参为下一个拦截器
    @Suppress("USELESS_ELVIS")
    val response = interceptor.intercept(next) ?: throw NullPointerException(
        "interceptor $interceptor returned null")

    if (exchange != null) {
      check(index + 1 >= interceptors.size || next.calls == 1) {
        "network interceptor $interceptor must call proceed() exactly once"
      }
    }

    check(response.body != null) { "interceptor $interceptor returned a response with no body" }

    return response
  }
  
  // ...
  
  internal fun copy(
    index: Int = this.index,
    exchange: Exchange? = this.exchange,
    request: Request = this.request,
    connectTimeoutMillis: Int = this.connectTimeoutMillis,
    readTimeoutMillis: Int = this.readTimeoutMillis,
    writeTimeoutMillis: Int = this.writeTimeoutMillis
  ) = RealInterceptorChain(call, interceptors, index, exchange, request, connectTimeoutMillis, readTimeoutMillis, writeTimeoutMillis)
}

这段代码里的 1、2、3 步逻辑，先通过 copy() 根据下一个拦截器创建一个新的责任链，从其中 copy 的入参为 index + 1 可以得出这个结论，此时当前 index 并没有被改变，所以第二步取的是当前的拦截器，然后调用当前拦截器的拦截方法，入参为新的责任链，如果拦截方法可以顺利完成的话，那么会返回 Response，否则会在拦截方法中继续处理责任链（也是 Chain#proceed()）。那排除掉自定义拦截器，自然默认第一个拦截器是 RetryAndFollowUpInterceptor，下面来看看这里面的 intercept() 做了什么。

RetryAndFollowUpInterceptor

重试与重定向拦截器。

RetryAndFollowUpInterceptor.kt

@Throws(IOException::class)
override fun intercept(chain: Interceptor.Chain): Response {
  val realChain = chain as RealInterceptorChain
  var request = chain.request
  val call = realChain.call
  var followUpCount = 0
  var priorResponse: Response? = null
  var newExchangeFinder = true
  var recoveredFailures = listOf<IOException>()
  while (true) {
    // 进入网络拦截器处理，这个方法在 RealCall 中，目的是创建一个 ExchangeFinder
    call.enterNetworkInterceptorExchange(request, newExchangeFinder)

    var response: Response
    var closeActiveExchange = true
    try {
      if (call.isCanceled()) {
        throw IOException("Canceled")
      }

      try {
        // 执行下一个拦截器，拿到结果后再往下执行
        response = realChain.proceed(request)
        newExchangeFinder = true
      } catch (e: RouteException) {
        // 发生 Route 异常，尝试进行恢复
        // The attempt to connect via a route failed. The request will not have been sent.
        if (!recover(e.lastConnectException, call, request, requestSendStarted = false)) {
          throw e.firstConnectException.withSuppressed(recoveredFailures)
        } else {
          recoveredFailures += e.firstConnectException
        }
        newExchangeFinder = false
        continue
      } catch (e: IOException) {
        // 发生 IO 异常，尝试进行恢复
        // An attempt to communicate with a server failed. The request may have been sent.
        if (!recover(e, call, request, requestSendStarted = e !is ConnectionShutdownException)) {
          throw e.withSuppressed(recoveredFailures)
        } else {
          recoveredFailures += e
        }
        newExchangeFinder = false
        continue
      }

      // 构建 budy 为空的响应体
      // Attach the prior response if it exists. Such responses never have a body.
      if (priorResponse != null) {
        response = response.newBuilder()
            .priorResponse(priorResponse.newBuilder()
                .body(null)
                .build())
              .build()
      }

      // call.interceptorScopedExchange 在数据流结束的时候会返回 null
      val exchange = call.interceptorScopedExchange
      // 检查是否需要重定向，不需要则 followup 为 null
      val followUp = followUpRequest(response, exchange)

      if (followUp == null) {
        if (exchange != null && exchange.isDuplex) {
          call.timeoutEarlyExit()
        }
        closeActiveExchange = false
        // 不需要重定向，并且数据流正常结束了，返回之前的 response
        return response
      }

      val followUpBody = followUp.body
      if (followUpBody != null && followUpBody.isOneShot()) {
        closeActiveExchange = false
        // 最多需要一次 writeTo 且只传输一次，则返回响应体
        return response
      }

      // 关闭资源
      response.body?.closeQuietly()

      // 重定向次数大于最大值，抛出异常
      if (++followUpCount > MAX_FOLLOW_UPS) {
        throw ProtocolException("Too many follow-up requests: $followUpCount")
      }

      // 循环判断
      request = followUp
      priorResponse = response
    } finally {
      // 退出网络拦截器处理
      call.exitNetworkInterceptorExchange(closeActiveExchange)
    }
  }
}

RealCall.kt

/**
 * 创建 ExchangeFinder 实例
 */
fun enterNetworkInterceptorExchange(request: Request, newExchangeFinder: Boolean) {
  check(interceptorScopedExchange == null)

  synchronized(this) {
    check(!responseBodyOpen) {
      "cannot make a new request because the previous response is still open: " +
          "please call response.close()"
    }
    check(!requestBodyOpen)
  }

  if (newExchangeFinder) {
    this.exchangeFinder = ExchangeFinder(
        connectionPool,
        createAddress(request.url),
        this,
        eventListener
    )
  }
}

上面的代码中，RetryAndFollowUpInterceptor 这个拦截器首先会去创建 ExchangeFinder 实例，然后往下执行，其实就到了下一个拦截器了，后面的逻辑是在下一个拦截器处理后再进行的，那按照默认拦截器的添加顺序，到 BridgeInterceptor 了，因为在 realChain.proceed(request) 这行代码中，调用的依然是 RealInterceptorChain#proceed 的逻辑，拿到下一个拦截器，作为当前拦截器拦截方法的入参，因此直接定位到 BridgeInterceptor#intercept

BridgeInterceptor

桥拦截器，应用程序代码到网络代码的桥梁，首先根据用户请求构建网络请求，然后继续调用网络，最后根据网络响应构建用户响应。

BridgeInterceptor.kt

@Throws(IOException::class)
override fun intercept(chain: Interceptor.Chain): Response {
  // 这个 request 还是原始的 request
  val userRequest = chain.request()
  val requestBuilder = userRequest.newBuilder()

  // 拼接请求体
  val body = userRequest.body
  if (body != null) {
    val contentType = body.contentType()
    if (contentType != null) {
      requestBuilder.header("Content-Type", contentType.toString())
    }

    val contentLength = body.contentLength()
    if (contentLength != -1L) {
      requestBuilder.header("Content-Length", contentLength.toString())
      requestBuilder.removeHeader("Transfer-Encoding")
    } else {
      requestBuilder.header("Transfer-Encoding", "chunked")
      requestBuilder.removeHeader("Content-Length")
    }
  }

  // 拼接请求头
  if (userRequest.header("Host") == null) {
    requestBuilder.header("Host", userRequest.url.toHostHeader())
  }

  if (userRequest.header("Connection") == null) {
    requestBuilder.header("Connection", "Keep-Alive")
  }

  // If we add an "Accept-Encoding: gzip" header field we're responsible for also decompressing
  // the transfer stream.
  var transparentGzip = false
  if (userRequest.header("Accept-Encoding") == null && userRequest.header("Range") == null) {
    transparentGzip = true
    requestBuilder.header("Accept-Encoding", "gzip")
  }

  val cookies = cookieJar.loadForRequest(userRequest.url)
  if (cookies.isNotEmpty()) {
    requestBuilder.header("Cookie", cookieHeader(cookies))
  }

  if (userRequest.header("User-Agent") == null) {
    requestBuilder.header("User-Agent", userAgent)
  }

  // 这个会进入下一个拦截器，将组装好的请求体作为入参，拿到网络响应后
  val networkResponse = chain.proceed(requestBuilder.build())

  // 处理 cookie 信息（这个扩展里会判断 NO_COOKIES 的话，直接 return，代码就不拷了）
  cookieJar.receiveHeaders(userRequest.url, networkResponse.headers)

  // 处理服务器返回的响应，将其转换为用户可用的响应
  val responseBuilder = networkResponse.newBuilder()
      .request(userRequest)

  if (transparentGzip &&
      "gzip".equals(networkResponse.header("Content-Encoding"), ignoreCase = true) &&
      networkResponse.promisesBody()) {
    val responseBody = networkResponse.body
    if (responseBody != null) {
      val gzipSource = GzipSource(responseBody.source())
      val strippedHeaders = networkResponse.headers.newBuilder()
          .removeAll("Content-Encoding")
          .removeAll("Content-Length")
          .build()
      responseBuilder.headers(strippedHeaders)
      val contentType = networkResponse.header("Content-Type")
      responseBuilder.body(RealResponseBody(contentType, -1L, gzipSource.buffer()))
    }
  }

  return responseBuilder.build()
}

BridgeInterceptor 主要的工作就是将开发者传入的网络请求信息进行转换为实际的 HTTP 请求，以及将 HTTP 响应转换为开发者能够使用的信息。老套路，下一个就该 CacheInterceptor 了，直接看看他的 intercept

CacheInterceptor

缓存拦截器，处理来自缓存的请求并将响应写入缓存（用到了策略模式）。

CacheInterceptor.kt

@Throws(IOException::class)
override fun intercept(chain: Interceptor.Chain): Response {
  // 拿到当前的 RealCall 实例
  val call = chain.call()
  // 这个 cache 是 DiskLruCache（最近最少使用），key 是请求的 url，返回 Response 实例
  val cacheCandidate = cache?.get(chain.request())

  val now = System.currentTimeMillis()

  // 缓存策略，用于判断使用缓存还是使用网络请求，或者都用
  val strategy = CacheStrategy.Factory(now, chain.request(), cacheCandidate).compute()
  val networkRequest = strategy.networkRequest
  val cacheResponse = strategy.cacheResponse

  cache?.trackResponse(strategy)
  val listener = (call as? RealCall)?.eventListener ?: EventListener.NONE

  // 有缓存，但策略中不使用，则释放缓存资源
  if (cacheCandidate != null && cacheResponse == null) {
    // The cache candidate wasn't applicable. Close it.
    cacheCandidate.body?.closeQuietly()
  }

  // 不使用网络请求，也不使用缓存，直接返回失败的响应
  // If we're forbidden from using the network and the cache is insufficient, fail.
  if (networkRequest == null && cacheResponse == null) {
    return Response.Builder()
        .request(chain.request())
        .protocol(Protocol.HTTP_1_1)
        .code(HTTP_GATEWAY_TIMEOUT)
        .message("Unsatisfiable Request (only-if-cached)")
        .body(EMPTY_RESPONSE)
        .sentRequestAtMillis(-1L)
        .receivedResponseAtMillis(System.currentTimeMillis())
        .build().also {
          listener.satisfactionFailure(call, it)
        }
  }

  // 策略中不使用网络请求，则说明用的是缓存，直接返回缓存
  // If we don't need the network, we're done.
  if (networkRequest == null) {
    return cacheResponse!!.newBuilder()
        .cacheResponse(stripBody(cacheResponse))
        .build().also {
          listener.cacheHit(call, it)
        }
  }

  // 回调缓存的监听
  if (cacheResponse != null) {
    listener.cacheConditionalHit(call, cacheResponse)
  } else if (cache != null) {
    listener.cacheMiss(call)
  }

  var networkResponse: Response? = null
  try {
    // 执行下一个拦截器（ConnectInterceptor）进行网络请求，返回网络的响应
    networkResponse = chain.proceed(networkRequest)
  } finally {
    // 如果发生了 IO 或者其它异常，为了不泄漏缓存体，需要释放资源
    // If we're crashing on I/O or otherwise, don't leak the cache body.
    if (networkResponse == null && cacheCandidate != null) {
      cacheCandidate.body?.closeQuietly()
    }
  }

  // 如果策略中使用缓存，并且响应码为 304（无改变，无需传输内容），则返回缓存
  // If we have a cache response too, then we're doing a conditional get.
  if (cacheResponse != null) {
    if (networkResponse?.code == HTTP_NOT_MODIFIED) {
      val response = cacheResponse.newBuilder()
          .headers(combine(cacheResponse.headers, networkResponse.headers))
          .sentRequestAtMillis(networkResponse.sentRequestAtMillis)
          .receivedResponseAtMillis(networkResponse.receivedResponseAtMillis)
          .cacheResponse(stripBody(cacheResponse))
          .networkResponse(stripBody(networkResponse))
          .build()

      networkResponse.body!!.close()

      // Update the cache after combining headers but before stripping the
      // Content-Encoding header (as performed by initContentStream()).
      cache!!.trackConditionalCacheHit()
      // 更新缓存
      cache.update(cacheResponse, response)
      return response.also {
        listener.cacheHit(call, it)
      }
    } else {
      cacheResponse.body?.closeQuietly()
    }
  }

  // 根据网络响应构建响应体
  val response = networkResponse!!.newBuilder()
      .cacheResponse(stripBody(cacheResponse))
      .networkResponse(stripBody(networkResponse))
      .build()

  if (cache != null) {
    if (response.promisesBody() && CacheStrategy.isCacheable(response, networkRequest)) {
      // 将请求返回的结果存进缓存
      // Offer this request to the cache.
      val cacheRequest = cache.put(response)
      return cacheWritingResponse(cacheRequest, response).also {
        if (cacheResponse != null) {
          // This will log a conditional cache miss only.
          listener.cacheMiss(call)
        }
      }
    }

    if (HttpMethod.invalidatesCache(networkRequest.method)) {
      try {
        cache.remove(networkRequest)
      } catch (_: IOException) {
        // The cache cannot be written.
      }
    }
  }

  return response
}

CacheInterceptor 实现了缓存的读取和存储，当网络请求的时候执行缓存拦截器的时候，会根据缓存策略去判断是否需要使用缓存、是否使用网络请求数据，如果使用缓存并且有缓存的话会直接返回缓存，没有则会执行后面的拦截器（ConnectInterceptor）继续请求网络，请求成功回将请求到的数据缓存起来。

ConnectInterceptor

连接拦截器，打开与目标服务器的连接并继续下一个拦截器，网络可能可用于返回的响应或使用条件 GET 验证缓存的响应。

ConnectInterceptor.kt

@Throws(IOException::class)
override fun intercept(chain: Interceptor.Chain): Response {
  val realChain = chain as RealInterceptorChain
  // 调用 RealCall#initExchange，就看看这个方法的逻辑，怎么建立的连接
  val exchange = realChain.call.initExchange(chain)
  val connectedChain = realChain.copy(exchange = exchange)
  // 执行下一个拦截器（不考虑 forWebSocket，是 CallServerInterceptor）
  return connectedChain.proceed(realChain.request)
}

RealCall.kt

internal fun initExchange(chain: RealInterceptorChain): Exchange {
  synchronized(this) {
    check(expectMoreExchanges) { "released" }
    check(!responseBodyOpen)
    check(!requestBodyOpen)
  }

  // 这个取的就是在 RetryAndFollowUpInterceptor 中创建的 ExchangeFinder
  val exchangeFinder = this.exchangeFinder!!
  // 调用 find 方法
  val codec = exchangeFinder.find(client, chain)
  val result = Exchange(this, eventListener, exchangeFinder, codec)
  this.interceptorScopedExchange = result
  this.exchange = result
  synchronized(this) {
    this.requestBodyOpen = true
    this.responseBodyOpen = true
  }

  if (canceled) throw IOException("Canceled")
  return result
}

ExchangeFinder.kt

fun find(
  client: OkHttpClient,
  chain: RealInterceptorChain
): ExchangeCodec {
  try {
    // 调用本身的 findHealthyConnection
    val resultConnection = findHealthyConnection(
        connectTimeout = chain.connectTimeoutMillis,
        readTimeout = chain.readTimeoutMillis,
        writeTimeout = chain.writeTimeoutMillis,
        pingIntervalMillis = client.pingIntervalMillis,
        connectionRetryEnabled = client.retryOnConnectionFailure,
        doExtensiveHealthChecks = chain.request.method != "GET"
    )
    return resultConnection.newCodec(client, chain)
  } catch (e: RouteException) {
    trackFailure(e.lastConnectException)
    throw e
  } catch (e: IOException) {
    trackFailure(e)
    throw RouteException(e)
  }
}

@Throws(IOException::class)
private fun findHealthyConnection(
  connectTimeout: Int,
  readTimeout: Int,
  writeTimeout: Int,
  pingIntervalMillis: Int,
  connectionRetryEnabled: Boolean,
  doExtensiveHealthChecks: Boolean
): RealConnection {
  while (true) {
    // 调用本身的 findConnection
    val candidate = findConnection(
        connectTimeout = connectTimeout,
        readTimeout = readTimeout,
        writeTimeout = writeTimeout,
        pingIntervalMillis = pingIntervalMillis,
        connectionRetryEnabled = connectionRetryEnabled
    )

    // Confirm that the connection is good.
    if (candidate.isHealthy(doExtensiveHealthChecks)) {
      return candidate
    }

    // If it isn't, take it out of the pool.
    candidate.noNewExchanges()

    // Make sure we have some routes left to try. One example where we may exhaust all the routes
    // would happen if we made a new connection and it immediately is detected as unhealthy.
    if (nextRouteToTry != null) continue

    val routesLeft = routeSelection?.hasNext() ?: true
    if (routesLeft) continue

    val routesSelectionLeft = routeSelector?.hasNext() ?: true
    if (routesSelectionLeft) continue

    throw IOException("exhausted all routes")
  }
}

@Throws(IOException::class)
private fun findConnection(
  connectTimeout: Int,
  readTimeout: Int,
  writeTimeout: Int,
  pingIntervalMillis: Int,
  connectionRetryEnabled: Boolean
): RealConnection {
  if (call.isCanceled()) throw IOException("Canceled")

  // 获取 RealCall 里的连接并尝试重用
  // Attempt to reuse the connection from the call.
  val callConnection = call.connection // This may be mutated by releaseConnectionNoEvents()!
  if (callConnection != null) {
    var toClose: Socket? = null
    synchronized(callConnection) {
      if (callConnection.noNewExchanges || !sameHostAndPort(callConnection.route().address.url)) {
        toClose = call.releaseConnectionNoEvents()
      }
    }

    // 连接未被释放，则重用
    // If the call's connection wasn't released, reuse it. We don't call connectionAcquired() here
    // because we already acquired it.
    if (call.connection != null) {
      check(toClose == null)
      return callConnection
    }

    // 连接已被释放则关闭 socket 并回调事件
    // The call's connection was released.
    toClose?.closeQuietly()
    eventListener.connectionReleased(call, callConnection)
  }

  // 创建新的连接需要刷新计数字段
  // We need a new connection. Give it fresh stats.
  refusedStreamCount = 0
  connectionShutdownCount = 0
  otherFailureCount = 0

  // 尝试从连接池（RealConnectionPool）中获取连接
  // 这里会判断连接是否可以被分配传送到指定地址
  // 判断后最终会执行 RealCall#acquireConnectionNoEvents
  // 这个方法判断拿到的 connection 会判断是否持有锁和判空
  // 通过判断最后会赋值到 RealCall#connection 中
  // RealCall 会以弱引用的形式被添加到 RealConnection#calls 中
  // RealConnection 会记录当前连接的请求
  // 此时没有路由
  // Attempt to get a connection from the pool.
  if (connectionPool.callAcquirePooledConnection(address, call, null, false)) {
    val result = call.connection!!
    // 成功后会回调，获取连接成功
    eventListener.connectionAcquired(call, result)
    return result
  }

  // 找合适的路由地址，先判断有没有已标记的，没有就尝试拿到一个新的路由
  // Nothing in the pool. Figure out what route we'll try next.
  val routes: List<Route>?
  val route: Route
  if (nextRouteToTry != null) {
    // Use a route from a preceding coalesced connection.
    routes = null
    route = nextRouteToTry!!
    nextRouteToTry = null
  } else if (routeSelection != null && routeSelection!!.hasNext()) {
    // Use a route from an existing route selection.
    routes = null
    route = routeSelection!!.next()
  } else {
    // Compute a new route selection. This is a blocking operation!
    var localRouteSelector = routeSelector
    if (localRouteSelector == null) {
      localRouteSelector = RouteSelector(address, call.client.routeDatabase, call, eventListener)
      this.routeSelector = localRouteSelector
    }
    val localRouteSelection = localRouteSelector.next()
    routeSelection = localRouteSelection
    routes = localRouteSelection.routes

    if (call.isCanceled()) throw IOException("Canceled")

    // 拿到路由地址列表后，再尝试找连接，如果找到直接返回
    // Now that we have a set of IP addresses, make another attempt at getting a connection from
    // the pool. We have a better chance of matching thanks to connection coalescing.
    if (connectionPool.callAcquirePooledConnection(address, call, routes, false)) {
      val result = call.connection!!
      eventListener.connectionAcquired(call, result)
      return result
    }

    route = localRouteSelection.next()
  }

  // 找不到连接，则会根据路由新建一个 RealConnection
  // Connect. Tell the call about the connecting call so async cancels work.
  val newConnection = RealConnection(connectionPool, route)
  call.connectionToCancel = newConnection
  try {
    // 执行连接，是 RealConnection#connect
    newConnection.connect(
        connectTimeout,
        readTimeout,
        writeTimeout,
        pingIntervalMillis,
        connectionRetryEnabled,
        call,
        eventListener
    )
  } finally {
    call.connectionToCancel = null
  }
  // 记录路由
  call.client.routeDatabase.connected(newConnection.route())

  // If we raced another call connecting to this host, coalesce the connections. This makes for 3
  // different lookups in the connection pool!
  if (connectionPool.callAcquirePooledConnection(address, call, routes, true)) {
    val result = call.connection!!
    nextRouteToTry = route
    newConnection.socket().closeQuietly()
    eventListener.connectionAcquired(call, result)
    return result
  }

  synchronized(newConnection) {
    connectionPool.put(newConnection)
    call.acquireConnectionNoEvents(newConnection)
  }

  eventListener.connectionAcquired(call, newConnection)
  return newConnection
}

RealConnection.kt

fun connect(
  connectTimeout: Int,
  readTimeout: Int,
  writeTimeout: Int,
  pingIntervalMillis: Int,
  connectionRetryEnabled: Boolean,
  call: Call,
  eventListener: EventListener
) {
  check(protocol == null) { "already connected" }

  var routeException: RouteException? = null
  val connectionSpecs = route.address.connectionSpecs
  val connectionSpecSelector = ConnectionSpecSelector(connectionSpecs)

  // HTTP 的请求判断
  if (route.address.sslSocketFactory == null) {
    if (ConnectionSpec.CLEARTEXT !in connectionSpecs) {
      throw RouteException(UnknownServiceException(
          "CLEARTEXT communication not enabled for client"))
    }
    val host = route.address.url.host
    if (!Platform.get().isCleartextTrafficPermitted(host)) {
      throw RouteException(UnknownServiceException(
          "CLEARTEXT communication to $host not permitted by network security policy"))
    }
  } else {
    if (Protocol.H2_PRIOR_KNOWLEDGE in route.address.protocols) {
      throw RouteException(UnknownServiceException(
          "H2_PRIOR_KNOWLEDGE cannot be used with HTTPS"))
    }
  }

  while (true) {
    try {
      if (route.requiresTunnel()) {
        // 返回通过 HTTP 代理创建 TLS 隧道的请求（未加密地发送到代理服务器）
        // 默认使用 HTTP_1_1 
        // 先通过建立连接通道（Proxy-Connection: Keep-Alive），保持长连接
        // 调用链: createTunnelRequest -> connectSocket -> Platform.get().connectSocket -> socket.connect() -> SocketImpl.connect()
        // 最终是通过 Socket 进行连接，具体代码就不拷贝了，可以自行看看
        connectTunnel(connectTimeout, readTimeout, writeTimeout, call, eventListener)
        if (rawSocket == null) {
          // We were unable to connect the tunnel but properly closed down our resources.
          break
        }
      } else {
        // 直接连接 socket 处理 HTTP 的请求连接
        connectSocket(connectTimeout, readTimeout, call, eventListener)
      }
      // 建立协议
      // 会先判断 sslSocketFactory 是否为空，为空的话就是普通的 HTTP 请求
      // 再判断 HTTP 版本号
      // 是否使用的是 HTTP_2 协议如果是会通过 startHttp2 执行请求
      // 否则默认还是使用 HTTP_1_1 协议
      // 最终通过 connectTls 建立 TLS 连接
      establishProtocol(connectionSpecSelector, pingIntervalMillis, call, eventListener)
      // 回调连接结束
      eventListener.connectEnd(call, route.socketAddress, route.proxy, protocol)
      break
    } catch (e: IOException) {
      socket?.closeQuietly()
      rawSocket?.closeQuietly()
      socket = null
      rawSocket = null
      source = null
      sink = null
      handshake = null
      protocol = null
      http2Connection = null
      allocationLimit = 1

      // 回调连接失败
      eventListener.connectFailed(call, route.socketAddress, route.proxy, null, e)

      // 异常分发
      if (routeException == null) {
        routeException = RouteException(e)
      } else {
        routeException.addConnectException(e)
      }

      if (!connectionRetryEnabled || !connectionSpecSelector.connectionFailed(e)) {
        throw routeException
      }
    }
  }

  if (route.requiresTunnel() && rawSocket == null) {
    throw RouteException(ProtocolException(
        "Too many tunnel connections attempted: $MAX_TUNNEL_ATTEMPTS"))
  }

  idleAtNs = System.nanoTime()
}

@Throws(IOException::class)
private fun connectTls(connectionSpecSelector: ConnectionSpecSelector) {
  val address = route.address
  val sslSocketFactory = address.sslSocketFactory
  var success = false
  var sslSocket: SSLSocket? = null
  try {
    // 在原始 socket（前面创建的）上通过 sslSocketFactory 包一层
    // Create the wrapper over the connected socket.
    sslSocket = sslSocketFactory!!.createSocket(
        rawSocket, address.url.host, address.url.port, true /* autoClose */) as SSLSocket

    // 配置 socket 密码、TLS 版本和扩展
    // Configure the socket's ciphers, TLS versions, and extensions.
    val connectionSpec = connectionSpecSelector.configureSecureSocket(sslSocket)
    if (connectionSpec.supportsTlsExtensions) {
      Platform.get().configureTlsExtensions(sslSocket, address.url.host, address.protocols)
    }

    // 开始握手
    // Force handshake. This can throw!
    sslSocket.startHandshake()
    // block for session establishment
    val sslSocketSession = sslSocket.session
    val unverifiedHandshake = sslSocketSession.handshake()

    // 验证目标主机是否可以接受套接字的证书
    // Verify that the socket's certificates are acceptable for the target host.
    if (!address.hostnameVerifier!!.verify(address.url.host, sslSocketSession)) {
      val peerCertificates = unverifiedHandshake.peerCertificates
      if (peerCertificates.isNotEmpty()) {
        val cert = peerCertificates[0] as X509Certificate
        throw SSLPeerUnverifiedException("""
            |Hostname ${address.url.host} not verified:
            |    certificate: ${CertificatePinner.pin(cert)}
            |    DN: ${cert.subjectDN.name}
            |    subjectAltNames: ${OkHostnameVerifier.allSubjectAltNames(cert)}
            """.trimMargin())
      } else {
        throw SSLPeerUnverifiedException(
            "Hostname ${address.url.host} not verified (no certificates)")
      }
    }

    // 返回地址的证书 pinner，如果不是 HTTPS 地址，则返回 null
    val certificatePinner = address.certificatePinner!!

    // 根据未验证的 TLS 握手记录新建一个 TLS 握手记录
    handshake = Handshake(unverifiedHandshake.tlsVersion, unverifiedHandshake.cipherSuite,
        unverifiedHandshake.localCertificates) {
      certificatePinner.certificateChainCleaner!!.clean(unverifiedHandshake.peerCertificates,
          address.url.host)
    }

    // 检查证书
    // Check that the certificate pinner is satisfied by the certificates presented.
    certificatePinner.check(address.url.host) {
      handshake!!.peerCertificates.map { it as X509Certificate }
    }

    // 成功，根据平台选择对应的应用层协议
    // Success! Save the handshake and the ALPN protocol.
    val maybeProtocol = if (connectionSpec.supportsTlsExtensions) {
      Platform.get().getSelectedProtocol(sslSocket)
    } else {
      null
    }
    socket = sslSocket
    source = sslSocket.source().buffer()
    sink = sslSocket.sink().buffer()
    // 找不到默认就是 HTTP1.1
    protocol = if (maybeProtocol != null) Protocol.get(maybeProtocol) else Protocol.HTTP_1_1
    success = true
  } finally {
    // 释放资源
    if (sslSocket != null) {
      Platform.get().afterHandshake(sslSocket)
    }
    if (!success) {
      sslSocket?.closeQuietly()
    }
  }
}

ConnectInterceptor 主要工作是判断当前连接是否可用，可用就直接返回，不可用就会从连接池中获取可用连接，如果找不到就切换不同的路由再次从连接池中获取可用的连接，如果还是没找到的话，就重新创建一个新的连接，进行 TLS 和 TCP 握手，最终将新创建的连接放入连接池中。

在连接创建之前，OkHttp 还会判断 HTTP 连接是否需要隧道连接，如果需要的话就添加相应的属性 RealConnection#createTunnelRequest，不需要则直接进行 socket 连接。

在建立协议的过程中，会判断是否为 HTTPS 的连接，不是则正常连接，是会先包一层 TLS，再进行连接。

最后就还有一个 CallServerInterceptor，请求服务器拦截器。

CallServerInterceptor

访问服务器拦截器，对服务器进行网络调用，是链中最后一个拦截器。

CallServerInterceptor.kt

@Throws(IOException::class)
override fun intercept(chain: Interceptor.Chain): Response {
  val realChain = chain as RealInterceptorChain
  // exchange 中 HTTP1.1 codec 对应 Http1ExchangeCodec
  // HTTP2 codec 对应 Http2ExchangeCodec
  // 封装的是创建、写入、刷新等方法
  val exchange = realChain.exchange!!
  val request = realChain.request
  val requestBody = request.body
  val sentRequestMillis = System.currentTimeMillis()

  var invokeStartEvent = true
  var responseBuilder: Response.Builder? = null
  var sendRequestException: IOException? = null
  try {
    // 写入请求头
    exchange.writeRequestHeaders(request)

    // 有请求体则写入（主要是写入和刷新逻辑），否则按没有请求体进行请求
    if (HttpMethod.permitsRequestBody(request.method) && requestBody != null) {
      // If there's a "Expect: 100-continue" header on the request, wait for a "HTTP/1.1 100
      // Continue" response before transmitting the request body. If we don't get that, return
      // what we did get (such as a 4xx response) without ever transmitting the request body.
      if ("100-continue".equals(request.header("Expect"), ignoreCase = true)) {
        exchange.flushRequest()
        responseBuilder = exchange.readResponseHeaders(expectContinue = true)
        exchange.responseHeadersStart()
        invokeStartEvent = false
      }
      if (responseBuilder == null) {
        if (requestBody.isDuplex()) {
          // Prepare a duplex body so that the application can send a request body later.
          exchange.flushRequest()
          val bufferedRequestBody = exchange.createRequestBody(request, true).buffer()
          requestBody.writeTo(bufferedRequestBody)
        } else {
          // Write the request body if the "Expect: 100-continue" expectation was met.
          val bufferedRequestBody = exchange.createRequestBody(request, false).buffer()
          requestBody.writeTo(bufferedRequestBody)
          bufferedRequestBody.close()
        }
      } else {
        exchange.noRequestBody()
        if (!exchange.connection.isMultiplexed) {
          // If the "Expect: 100-continue" expectation wasn't met, prevent the HTTP/1 connection
          // from being reused. Otherwise we're still obligated to transmit the request body to
          // leave the connection in a consistent state.
          exchange.noNewExchangesOnConnection()
        }
      }
    } else {
      exchange.noRequestBody()
    }

    // 如果没有请求体，那就请求刷新到底层 socket 并发出不再传输字节的信号
    if (requestBody == null || !requestBody.isDuplex()) {
      exchange.finishRequest()
    }
  } catch (e: IOException) {
    // 处理 IO 异常
    if (e is ConnectionShutdownException) {
      throw e // No request was sent so there's no response to read.
    }
    if (!exchange.hasFailure) {
      throw e // Don't attempt to read the response; we failed to send the request.
    }
    sendRequestException = e
  }

  try {
    // 处理响应
    if (responseBuilder == null) {
      // 解析来自 HTTP 传输的响应头的字节并返回 Response.Builder
      responseBuilder = exchange.readResponseHeaders(expectContinue = false)!!
      if (invokeStartEvent) {
        exchange.responseHeadersStart()
        invokeStartEvent = false
      }
    }
    // 拿到响应信息，记录 HTTP 响应码
    var response = responseBuilder
        .request(request)
        .handshake(exchange.connection.handshake())
        .sentRequestAtMillis(sentRequestMillis)
        .receivedResponseAtMillis(System.currentTimeMillis())
        .build()
    var code = response.code
    if (code == 100) {
      // 如果收到 100（继续完成请求的话）就再解析一次
      // Server sent a 100-continue even though we did not request one. Try again to read the
      // actual response status.
      responseBuilder = exchange.readResponseHeaders(expectContinue = false)!!
      if (invokeStartEvent) {
        exchange.responseHeadersStart()
      }
      response = responseBuilder
          .request(request)
          .handshake(exchange.connection.handshake())
          .sentRequestAtMillis(sentRequestMillis)
          .receivedResponseAtMillis(System.currentTimeMillis())
          .build()
      code = response.code
    }

    // 解析头部信息结束
    exchange.responseHeadersEnd(response)

    // 获取响应体
    response = if (forWebSocket && code == 101) {
      // Connection is upgrading, but we need to ensure interceptors see a non-null response body.
      response.newBuilder()
          .body(EMPTY_RESPONSE)
          .build()
    } else {
      response.newBuilder()
          .body(exchange.openResponseBody(response))
          .build()
    }
    // 根据响应码做对应的处理
    if ("close".equals(response.request.header("Connection"), ignoreCase = true) ||
        "close".equals(response.header("Connection"), ignoreCase = true)) {
      exchange.noNewExchangesOnConnection()
    }
    if ((code == 204 || code == 205) && response.body?.contentLength() ?: -1L > 0L) {
      throw ProtocolException(
          "HTTP $code had non-zero Content-Length: ${response.body?.contentLength()}")
    }
    // 没问题就返回
    return response
  } catch (e: IOException) {
    // IO 异常处理
    if (sendRequestException != null) {
      sendRequestException.addSuppressed(e)
      throw sendRequestException
    }
    throw e
  }
}

这里有提到的 Http1ExchangeCodec 和 Http2ExchangeCodec 都是用于流读写的，根据 HTTP 版本的不同进行区分使用，BufferedSink（输出流）和 BufferedSource（输入流）是由 okio 提供的工具，在这里主要用于写入请求头和请求体，读取响应头和响应体。

看完这部分的代码，可以知道 CallServerInterceptor 主要是给服务器发起请求并获取数据的，也是在默认的拦截器中最后一个拦截器，获取到服务器数据后，会直接返回给上一个拦截器。责任链最终还是会回到 RetryAndFollowupInterceptor 中返回，而 getResponseWithInterceptorChain 这个方法就能拿到 response，再在同步方法中返回这个 response，整个调用链就完成了。

源码篇小结

API 总结

OkHttpClient: Call 的工厂，可用于发送 HTTP 请求并获取其响应

Request: HTTP 请求

RealCall: OkHttp 的应用层和网络层之间的桥梁（包含高级应用层的连接、请求、响应和流）

RealCall.AsyncCall: 是个 Runnable，用于处理异步请求

Dispatcher: 用于执行请求的策略

RealInterceptorChain: 具体的拦截器链，承载整个拦截器链，最后是网络调用者（用于应用程序拦截器，exchange 必须为空，用于网络拦截器，exchange 必须为非空）

RetryAndFollowUpInterceptor: 重试与重定向拦截器

ExchangeFinder: 尝试查找交换的连接以及随后的任何重试策略（主要用于建立连接）

BridgeInterceptor: 桥拦截器，应用程序代码到网络代码的桥梁，首先根据用户请求构建网络请求，然后继续调用网络，最后根据网络响应构建用户响应

CacheInterceptor: 缓存拦截器，处理来自缓存的请求并将响应写入缓存

ConnectInterceptor: 连接拦截器，打开与目标服务器的连接并继续下一个拦截器，网络可能可用于返回的响应或使用条件 GET 验证缓存的响应

RealConnection: 与远程 web 服务器的连接，能够承载 1 个或多个并发流。连接的生命周期有两个阶段。

  1、在连接时，连接由使用单线程的单个调用拥有，这个阶段，连接不是共享的，也不需要锁定

  2、连接后，连接共享到连接池，此阶段，必须通过在连接上持有锁来保护对连接状态的访问

RealConnectionPool: 连接池，维护连接队列（ConcurrentLinkedQueue）和清理队列（TaskQueue）

CallServerInterceptor: 访问服务器拦截器，对服务器进行网络调用，是链中最后一个拦截器

Http1ExchangeCodec: 用于发送 HTTP/1.1 消息的套接字连接。严格执行以下生命周期。

  1、发送请求头（writeRequest）

  2、打开一个接收器来写入请求体 （newKnownLengthSink 或 newChunkedSink）

  3、写入然后关闭该接收器

  4、读取响应头（readResponseHeaders）

  5、打开源已读取响应正文（newFixedLengthSource 或 newChunkedSource 或 newUnknownLengthSource）

  6、读取并关闭该资源

 没有请求正文的会跳过创建和关闭请求征文

 没有响应正文的可以调用 newFixedLengthSource(0) 且可以跳过读取和关闭该源

Http2ExchangeCodec: 使用 HTTP/2 帧编码请求和响应

调用流程

同步调用

1、通过构造 OkHttpClient 和 Request 构造出 RealCall 实例

2、通过 RealCall#execute() 开始进行同步调用

3、通过 Dispatcher#executed() 将调用存放到同步调用队列（runningSyncCalls，实际是 ArrayQueue）中
3、通过 RealCall#getResponseWithInterceptorChain 开始进入责任链进行网络请求

异步调用

1、通过构造 OkHttpClient 和 Request 构造出 RealCall 实例

2、通过 RealCall#enqueue() 开始进行异步调用

3、通过 responseCallback（Callback）实例化 AsyncCall 并将他传入 Dispatcher#enqueue()

4、通过 Dispatcher#promoteAndExecute() 遍历集合中的 AsyncCall，并以此执行 AsyncCall#executeOn()，并把创建的线程池作为参数传递进去

5、调用线程池的 execute()，执行 AsyncCall#run()

6、在 AsyncCall#run() 中通过 RealCall#getResponseWithInterceptorChain() 开始进入责任链进行网络请求，并通过 responseCallback 进行回调，请求正常回调 onResponse()，出现异常回调 onFailure()

问题

1、OkHttp 实现网络请求的方式

OkHttp 实际上是通过 Socket 进行网络连接的，会根据配置先判断是否需要开启代理隧道（目的是利用 HTTP 来代理请求 HTTPS），需要则开启（connectTunnel）否则会直接建立 TCP 连接（connectSocket）。无论是否需要开启隧道，都会建立一个 TCP 连接（都会调用 connectSocket）。最后会调用 Platform.get().connectSocket()（Socket#connect()）打开一个 TCP 连接

2、为什么 response.body().string() 只能调用一次

通过 source 拿到字节流后，会调用 closeQuietly() 执行关闭，所以只能用一次，可以缓存一份或者自定义拦截器处理 log

3、OkHttp 运用的设计模式
构造者模式（OkHttpClient、Request 对象的创建）

工厂模式（获取 Call 接口的实例）

单例模式（Platform 类型）

策略模式（CacheInterceptor，在响应数据的选择中使用了策略模式，用缓存数据还是网络数据）

责任链模式（拦截器的链式调用）

享元模式（共享技术，支持复用）（Dispatcher 的线程池中，不限量的线程池实现了对象复用）

4、Dispatcher 在异步请求中为什么要分两个 ArrayDeque（runningAsyncCalls 和 readyAsyncCalls）
runningAsyncCalls 用于保存正在执行的请求，readyAsyncCalls 用于保存准备执行的请求，因为 Dispatcher 默认支持最大的并发请求（maxRequests）是 64 个，单个 Host（maxRequestsPerHost）最多执行 5 个并发请求，如果超了，那 Call 会先被放入 readyAsyncCalls 中，当出现空闲的线程时，再将 readyAsyncCalls 中的线程移到 runningAsyncCalls 中执行请求，具体逻辑在（Dispatcher#promoteAndExecute()）中，只要满足正在请求的数量 < 64 && 同一域名正在请求的数量 < 5 就会加入 runningAsyncCalls 中，否则会放到 readAsyncCalls 中。

思考

本来想着是希望通过 OkHttp 的代码来了解一下网络的分层，但是这个库主要还是集中在应用层的逻辑处理，需要注意的是 BridgeInterceptor 这个拦截器，我认为是数据逻辑上的桥（并不属于网络的分层结构中的桥）。但也意识到就算是应用层的开发，里面涉及到的知识点也是很多的（就像建立代理隧道来代理请求 HTTPS 这样的操作很强）。还有在 HTTPS 中握手的抽象 HandShake（也是用于描述完成的握手）、异步调用中为了提高复用的线程池（ThreadPoolExecutor 的使用）等等用法很值得学习。

最后说明一下，因为是了解 OkHttp 的主流程，里面有很多细节的地方并没有详细了解，比如 RealConnection 和 RealConnectionPool 他俩的细节就没大看懂… 如果文中有错误的理解希望读者可以指正，大家一起进步～

参考资料

• 官方文档
• OkHttp的源码解析（一）
• OkHttp的源码解析（二）
• Android 主流开源框架（三）OkHttp 源码解析
• 老生新谈，从OkHttp原理看网络请求