跳转至

Okhttp源码分析

这篇笔记保留了早期 Java/Kotlin 源码摘录。新增的设计模式分析与 UML统一对照 OkHttp 4.9.3，重点解释对象的职责、协作与模式边界；旧摘录中的简化表述可结合文末的阅读提醒对照。

基本流程

  1. 创建RequestBody
  2. 创建Request
  3. 创建OkhttpClient
  4. 调用newCall创建Call对象
  5. 执行异步或同步操作。
  6. 创建Socket连接
  7. 发送请求并处理返回结果

创建RequestBody

RequestBody主要通过writeTo方法将请求内容写入到BufferedSink。RequestBody提供了3个create静态方法来创建RequestBody，此外RequestBody还包含两个子类。

image-20211114170629029

FormBody用于表单提交，MultipartBody用于多内容提交。

创建Request

Request对象包含如下信息

1
2
3
4
5
final HttpUrl url; //请求连接
final String method; //请求方法
final Headers headers; //请求头
final @Nullable RequestBody body; //请求体
final Map<Class<?>, Object> tags; //tags

创建OkHttpClient

OkHttpClient可配置的信息包括如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
final Dispatcher dispatcher; //调度器主要负责异步请求
final @Nullable Proxy proxy; //代理
final List<Protocol> protocols; //支持的协议 默认支持HTTP1.1 和HTTP2.0
final List<ConnectionSpec> connectionSpecs; //传输层版本和连接协议
final List<Interceptor> interceptors; //拦截器
final List<Interceptor> networkInterceptors;//网络拦截器
final EventListener.Factory eventListenerFactory; //
final ProxySelector proxySelector; //代理选择器
final CookieJar cookieJar;
final @Nullable Cache cache; //缓存
final @Nullable InternalCache internalCache; //内部缓存
final SocketFactory socketFactory; //工厂
final SSLSocketFactory sslSocketFactory; //安全套接层
final CertificateChainCleaner certificateChainCleaner; //验证确认响应书，适用HTTPS 请求连接的主机名
final HostnameVerifier hostnameVerifier;//主机名字确认
final CertificatePinner certificatePinner; //证书链
final Authenticator proxyAuthenticator; //代理身份验证
final Authenticator authenticator; //本地身份验证
final ConnectionPool connectionPool; //链接池 复用连接
final Dns dns; //DNS
final boolean followSslRedirects; //安全套接层重定向
final boolean followRedirects; //本地重定向
final boolean retryOnConnectionFailure; //重试连接失败
final int callTimeout; //默认0
final int connectTimeout; //连接超时时间 默认10s
final int readTimeout;  //读取超时时间默认10s
final int writeTimeout; //写入超时时间 默认10s
final int pingInterval;

调用newCall创建Call对象

image-20211114170500713

OkHttpClient继承Call.Factory。

1
2
3
interface Factory {
  Call newCall(Request request);
}

OkHttpClient的newCall方法调用

1
2
//每次调用一个newCall都会创建一个新的RealCall对象
override fun newCall(request: Request): Call = RealCall(this, request, forWebSocket = false)

执行网络请求

image-20211114173043097

Call支持执行网络请求的方法enqueue和execute()。enqueue执行异步请求，execute()执行同步请求。

1
2
3
4
5
override fun enqueue(responseCallback: Callback) {
  check(executed.compareAndSet(false, true)) { "Already Executed" }
  callStart()
  client.dispatcher.enqueue(AsyncCall(responseCallback))
}

enqueue()

Dispatcher类

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
internal fun enqueue(call: AsyncCall) {
  synchronized(this) {
    //添加到准备请求的队列中
    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) {
      //寻找相同主机的Call
      val existingCall = findExistingCallWithHost(call.host)
      //所有的AsyncCall公用一个对象
      if (existingCall != null) call.reuseCallsPerHostFrom(existingCall)
    }
  }
  promoteAndExecute()
}

promoteAndExecute()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
private fun promoteAndExecute(): Boolean {
  this.assertThreadDoesntHoldLock()
  val executableCalls = mutableListOf<AsyncCall>()
  val isRunning: Boolean
  synchronized(this) {
    val i = readyAsyncCalls.iterator()
    while (i.hasNext()) {
      val asyncCall = i.next()
      //如果正在运行的Call的数量大于64
      if (runningAsyncCalls.size >= this.maxRequests) break // Max capacity. 
       //如果每个域名的请求数超过5
      if (asyncCall.callsPerHost.get() >= this.maxRequestsPerHost) continue // Host max capacity.
      //从准备数组中移除
      i.remove()
      //个数累加
      asyncCall.callsPerHost.incrementAndGet()
      executableCalls.add(asyncCall)
      runningAsyncCalls.add(asyncCall)
    }
    isRunning = runningCallsCount() > 0
  }
  //
  for (i in 0 until executableCalls.size) {
    val asyncCall = executableCalls[i]
      asyncCall.executeOn(executorService)
  }

  return isRunning
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
//创建线程池
@get:Synchronized
@get:JvmName("executorService") val executorService: ExecutorService
get() {
  if (executorServiceOrNull == null) {
    executorServiceOrNull = ThreadPoolExecutor(0, Int.MAX_VALUE, 60, TimeUnit.SECONDS,
                                               SynchronousQueue(), threadFactory("$okHttpName Dispatcher", false))
  }
  return executorServiceOrNull!!
}

executeOn()

AsyncCall的executeOn方法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
fun executeOn(executorService: ExecutorService) {
  client.dispatcher.assertThreadDoesntHoldLock()
  var success = false
  try {
    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!
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
override fun run() {
  threadName("OkHttp ${redactedUrl()}") {
    var signalledCallback = false
      timeout.enter()
      try {
        val response = getResponseWithInterceptorChain()
          signalledCallback = true
          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)
      }
  }
}

拦截器

RealInterceptorChain的proceed会创建一个InterceptorChain，并调用拦截器的intercept方法。除了CallServerInterceptor外，所有拦截器的intercept都会调用创建的InterceptorChain的proceed方法。

getResponseWithInterceptorChain()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@Throws(IOException::class)
internal fun getResponseWithInterceptorChain(): Response {
  // Build a full stack of interceptors.
  val interceptors = mutableListOf<Interceptor>()
  interceptors += client.interceptors
  interceptors += RetryAndFollowUpInterceptor(client)
  interceptors += BridgeInterceptor(client.cookieJar)
  interceptors += CacheInterceptor(client.cache)
  interceptors += ConnectInterceptor
  if (!forWebSocket) {
    interceptors += client.networkInterceptors
  }
  interceptors += CallServerInterceptor(forWebSocket)
  //创建chain并把所有的拦截器传递给chain
  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 {//执行chain的proceed方法
    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)
    }
  }
}

proceed()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@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"
    }
  }
  // Call the next interceptor in the chain.
  //调用copy方法创建一个chain
  val next = copy(index = index + 1, request = request)
  //获取下一个拦截器
  val interceptor = interceptors[index]
  @Suppress("USELESS_ELVIS")
  //调用拦截器的intercept方法 在interceptor中会调用next的proceed执行下一个拦截器
  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
}

缓存分析

Cache

缓存核心类是Cache对象。

Cache的构造函数。

1
2
//缓存目录 和最大尺寸
constructor(directory: File, maxSize: Long) : this(directory, maxSize, FileSystem.SYSTEM)

Cache提供了两个方法用于存储和获取缓存。

put方法将缓存存储到磁盘上。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
//返回一个CacheRequest对象
internal val cache = DiskLruCache(
    fileSystem = fileSystem,
    directory = directory,
    appVersion = VERSION,
    valueCount = ENTRY_COUNT,
    maxSize = maxSize,
    taskRunner = TaskRunner.INSTANCE
)

internal fun put(response: Response): CacheRequest? {
  val requestMethod = response.request.method //获取请求方法
  if (HttpMethod.invalidatesCache(response.request.method)) {
    try {
      remove(response.request)
    } catch (_: IOException) {
      // The cache cannot be written.
    }
    return null
  }
  //如果不是GET 请求直接返回null 
  if (requestMethod != "GET") {
    // Don't cache non-GET responses. We're technically allowed to cache HEAD requests and some
    // POST requests, but the complexity of doing so is high and the benefit is low.
    return null
  }
  //Vary头的值为*的情况
  if (response.hasVaryAll()) {
    return null
  }
  val entry = Entry(response) //创建Entry对象
  var editor: DiskLruCache.Editor? = null
  try {
    //先调用key方法生成key
    //调用DiskLruCache的edit方法获取Editor
    editor = cache.edit(key(response.request.url)) ?: return null
    entry.writeTo(editor) //写入本地
    return RealCacheRequest(editor)
  } catch (_: IOException) {
    abortQuietly(editor)
    return null
  }
}

get方法获取缓存

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
internal fun get(request: Request): Response? {
  val key = key(request.url)
  val snapshot: DiskLruCache.Snapshot = try {
    cache[key] ?: return null
  } catch (_: IOException) {
    return null // Give up because the cache cannot be read.
  }
  val entry: Entry = try {
    Entry(snapshot.getSource(ENTRY_METADATA))
  } catch (_: IOException) {
    snapshot.closeQuietly()
    return null
  }
  val response = entry.response(snapshot)
  if (!entry.matches(request, response)) {
    response.body?.closeQuietly()
    return null
  }
  return response
}

CacheInterceptor

Cache这两个方法都被CacheInterceptor对象调用。所以处理缓存的核心逻辑都在CacheInterceptor中。

缓存逻辑

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
if (cache != null) {
    //有body 并且可以被缓存
    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.
        }
    }
}

isCacheable方法用来判断是否可以缓存。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
fun isCacheable(response: Response, request: Request): Boolean {
      // Always go to network for uncacheable response codes (RFC 7231 section 6.1), This
      // implementation doesn't support caching partial content.
    when (response.code) {
        HTTP_OK,
        HTTP_NOT_AUTHORITATIVE,
        HTTP_NO_CONTENT,
        HTTP_MULT_CHOICE,
        HTTP_MOVED_PERM,
        HTTP_NOT_FOUND,
        HTTP_BAD_METHOD,
        HTTP_GONE,
        HTTP_REQ_TOO_LONG,
        HTTP_NOT_IMPLEMENTED,
        StatusLine.HTTP_PERM_REDIRECT -> {
          // These codes can be cached unless headers forbid it.
        }

        HTTP_MOVED_TEMP,
        StatusLine.HTTP_TEMP_REDIRECT -> {
          // These codes can only be cached with the right response headers.
          // http://tools.ietf.org/html/rfc7234#section-3
          // s-maxage is not checked because OkHttp is a private cache that should ignore s-maxage.
          if (response.header("Expires") == null &&
              response.cacheControl.maxAgeSeconds == -1 &&
              !response.cacheControl.isPublic &&
              !response.cacheControl.isPrivate) {
            return false
          }
        }

        else -> {
          // All other codes cannot be cached.
          return false
        }
      }

      // A 'no-store' directive on request or response prevents the response from being cached.
    return !response.cacheControl.noStore && !request.cacheControl.noStore
}

这里面加了一堆判断，这里需要先了解一下HTTP的缓存原理。才能搞清楚这些判断的逻辑。

Http缓存原理

在HTTP 1.0时代，响应使用Expires头标识缓存的有效期，其值是一个绝对时间，比如Expires:Thu,31 Dec 2020 23:59:59 GMT。当客户端再次发出网络请求时可比较当前时间 和上次响应的expires时间进行比较，来决定是使用缓存还是发起新的请求。

使用Expires头最大的问题是它依赖客户端的本地时间，如果用户自己修改了本地时间，就会导致无法准确的判断缓存是否过期。

因此，从HTTP 1.1 开始使用Cache-Control头表示缓存状态，它的优先级高于Expires，常见的取值为下面的一个或多个。

  • private，默认值，标识那些私有的业务逻辑数据，比如根据用户行为下发的推荐数据。该模式下网络链路中的代理服务器等节点不应该缓存这部分数据，因为没有实际意义。
  • public 与private相反，public用于标识那些通用的业务数据，比如获取新闻列表，所有人看到的都是同一份数据，因此客户端、代理服务器都可以缓存。
  • no-cache 可进行缓存，但在客户端使用缓存前必须要去服务端进行缓存资源有效性的验证，即下文的对比缓存部分，我们稍后介绍。
  • max-age 表示缓存时长单位为秒，指一个时间段，比如一年，通常用于不经常变化的静态资源。
  • no-store 任何节点禁止使用缓存。

强制缓存

在上述缓存头规约基础之上，强制缓存是指网络请求响应header标识了Expires或Cache-Control带了max-age信息，而此时客户端计算缓存并未过期，则可以直接使用本地缓存内容，而不用真正的发起一次网络请求。

协商缓存

强制缓存最大的问题是，一旦服务端资源有更新，直到缓存时间截止前，客户端无法获取到最新的资源（除非请求时手动添加no-store头），另外大部分情况下服务器的资源无法直接确定缓存失效时间，所以使用对比缓存更灵活一些。

使用**Last-Modify** / **If-Modify-Since**头实现协商缓存，具体方法是服务端响应头添加Last-Modify头标识资源的最后修改时间，单位为秒，当客户端再次发起请求时添加If-Modify-Since头并赋值为上次请求拿到的Last-Modify头的值。

服务端收到请求后自行判断缓存资源是否仍然有效，如果有效则返回状态码304同时body体为空，否则下发最新的资源数据。客户端如果发现状态码是304，则取出本地的缓存数据作为响应。

使用这套方案有一个问题，那就是资源文件使用最后修改时间有一定的局限性：

  1. Last-Modify单位为秒，如果某些文件在一秒内被修改则并不能准确的标识修改时间。
  2. 资源修改时间并不能作为资源是否修改的唯一依据，比如资源文件是Daily Build的，每天都会生成新的，但是其实际内容可能并未改变。

建立连接

image-20211114174719516

OkHttp通过ConnectInterceptor建立连接。在ConnectInterceptor的intercept方法中，通过调用RealCall的initExchange方法初始化一个Exchange对象。Exchange对象负责交换数据。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
object ConnectInterceptor : Interceptor {
  @Throws(IOException::class)
  override fun intercept(chain: Interceptor.Chain): Response {
    val realChain = chain as RealInterceptorChain
    //调用RealCall的initExchange初始化Exchange
    val exchange = realChain.call.initExchange(chain)
    //创建一个新的Chain
    val connectedChain = realChain.copy(exchange = exchange)
    return connectedChain.proceed(realChain.request)
  }
}

initExchange()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/** Finds a new or pooled connection to carry a forthcoming request and response. */
internal fun initExchange(chain: RealInterceptorChain): Exchange {
  synchronized(this) {
    check(expectMoreExchanges) { "released" }
    check(!responseBodyOpen)
    check(!requestBodyOpen)
  }
  //ExchangeFinder
  val exchangeFinder = this.exchangeFinder!!
  //ExchangeFinder调用find方法获取ExchangeCodec
  val codec = exchangeFinder.find(client, chain)
  //创建Exchange
  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
}

find()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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"
    )
    //调用RealConnection的newCodec获取ExchangeCodec
    return resultConnection.newCodec(client, chain)
  } catch (e: RouteException) {
    trackFailure(e.lastConnectException)
    throw e
  } catch (e: IOException) {
    trackFailure(e)
    throw RouteException(e)
  }
}

findHealthyConnection()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@Throws(IOException::class)
private fun findHealthyConnection(
  connectTimeout: Int,
  readTimeout: Int,
  writeTimeout: Int,
  pingIntervalMillis: Int,
  connectionRetryEnabled: Boolean,
  doExtensiveHealthChecks: Boolean
): RealConnection {
  while (true) {
    //调用findConnection方法获取RealConnection
    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")
  }
}

findConnection()

  1. 如果call里面的connection不为空，则复用call里面的connection。
  2. 如果call中的connection为空，则从连接池中获取一个。
  3. 如果缓存池里也没有就创建一个
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@Throws(IOException::class)
private fun findConnection(
  connectTimeout: Int,
  readTimeout: Int,
  writeTimeout: Int,
  pingIntervalMillis: Int,
  connectionRetryEnabled: Boolean
): RealConnection {
  //如果取消抛异常
  if (call.isCanceled()) throw IOException("Canceled")
  // Attempt to reuse the connection from the call.
  //复用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.
    //如果call里面的connection不为空
    if (call.connection != null) {
      check(toClose == null)
      return callConnection
    }
    // 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
  // 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()
  }
  // Connect. Tell the call about the connecting call so async cancels work.
  //创建连接
  val newConnection = RealConnection(connectionPool, route)
  call.connectionToCancel = newConnection
  try {
    //建立连接
    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)
    //将connection赋值给call
    call.acquireConnectionNoEvents(newConnection)
  }
  eventListener.connectionAcquired(call, newConnection)
  return newConnection
}

OkHttp连接复用

OkHttp 连接复用主要是通过 RealConnectionPool 来实现的，其内部定义了一个 ConcurrentLinkedQueue 来存储创建的 RealConnection。当获取连接时，优先判断 Call 中的` Connection是否为空，如果不为空则直接复用，为空则从连接池中获取，当从连接池中获取不到的时候才调用RealConnection构造函数创建一个新的连接并添加到 RealConnectionPool 中。RealConnectionPool 提供了两个可配置的参数：最大空闲连接数和存活时长，这两个参数默认是 5 个和 5 分钟，当然我们也可以在OKHttpClient 中自己配置。当向 RealConnectionPool 中添加连接和连接变成空闲时，内部都会遍历队列中所有的连接，如果最大空闲连接数或者存活时长都大于设定的则移除存空闲时长最长的连接，该方法会多次执行，直到最大空闲连接数和存活时长都小于设定的值。

image-20211115105736461

ConnectionPool

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class ConnectionPool internal constructor(
  internal val delegate: RealConnectionPool
) {
  constructor(
    maxIdleConnections: Int,
    keepAliveDuration: Long,
    timeUnit: TimeUnit
  ) : this(RealConnectionPool(//创建RealConnectionPool
      taskRunner = TaskRunner.INSTANCE, //获取TaskRunner
      maxIdleConnections = maxIdleConnections,
      keepAliveDuration = keepAliveDuration,
      timeUnit = timeUnit
  ))
  //最大空闲连接数5 最长空闲时间5分钟
  constructor() : this(5, 5, TimeUnit.MINUTES)

  /** Returns the number of idle connections in the pool. */
  fun idleConnectionCount(): Int = delegate.idleConnectionCount()

  /** Returns total number of connections in the pool. */
  fun connectionCount(): Int = delegate.connectionCount()

  /** Close and remove all idle connections in the pool. */
  fun evictAll() {
    delegate.evictAll()
  }
}

清理连接

image-20211115114825724

执行网络请求

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
/** This is the last interceptor in the chain. It makes a network call to the server. */
public final class CallServerInterceptor implements Interceptor {
  private final boolean forWebSocket;

  public CallServerInterceptor(boolean forWebSocket) {
    this.forWebSocket = forWebSocket;
  }

  @Override public Response intercept(Chain chain) throws IOException {
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    Exchange exchange = realChain.exchange();
    Request request = realChain.request();

    long sentRequestMillis = System.currentTimeMillis();

    exchange.writeRequestHeaders(request); //写入请求头

    boolean responseHeadersStarted = false;
    Response.Builder responseBuilder = null;
    if (HttpMethod.permitsRequestBody(request.method()) && request.body() != 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".equalsIgnoreCase(request.header("Expect"))) {
        exchange.flushRequest();
        responseHeadersStarted = true;
        exchange.responseHeadersStart();
        responseBuilder = exchange.readResponseHeaders(true);
      }
      //写入请求体
      if (responseBuilder == null) {
        if (request.body().isDuplex()) {
          // Prepare a duplex body so that the application can send a request body later.
          exchange.flushRequest();
          BufferedSink bufferedRequestBody = Okio.buffer(
              exchange.createRequestBody(request, true));
          request.body().writeTo(bufferedRequestBody);
        } else {
          // Write the request body if the "Expect: 100-continue" expectation was met.
          BufferedSink bufferedRequestBody = Okio.buffer(
              exchange.createRequestBody(request, false));
          request.body().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();
    }

    if (request.body() == null || !request.body().isDuplex()) {
      exchange.finishRequest();
    }

    if (!responseHeadersStarted) {
      exchange.responseHeadersStart();
    }

    if (responseBuilder == null) {
      responseBuilder = exchange.readResponseHeaders(false);
    }

    Response response = responseBuilder
        .request(request)
        .handshake(exchange.connection().handshake())
        .sentRequestAtMillis(sentRequestMillis)
        .receivedResponseAtMillis(System.currentTimeMillis())
        .build();

    int code = response.code();
    if (code == 100) {
      // server sent a 100-continue even though we did not request one.
      // try again to read the actual response
      response = exchange.readResponseHeaders(false)
          .request(request)
          .handshake(exchange.connection().handshake())
          .sentRequestAtMillis(sentRequestMillis)
          .receivedResponseAtMillis(System.currentTimeMillis())
          .build();

      code = response.code();
    }

    exchange.responseHeadersEnd(response);

    if (forWebSocket && code == 101) {
      // Connection is upgrading, but we need to ensure interceptors see a non-null response body.
      response = response.newBuilder()
          .body(Util.EMPTY_RESPONSE)
          .build();
    } else {
      response = response.newBuilder()
          .body(exchange.openResponseBody(response))
          .build();
    }

    if ("close".equalsIgnoreCase(response.request().header("Connection"))
        || "close".equalsIgnoreCase(response.header("Connection"))) {
      exchange.noNewExchangesOnConnection();
    }

    if ((code == 204 || code == 205) && response.body().contentLength() > 0) {
      throw new ProtocolException(
          "HTTP " + code + " had non-zero Content-Length: " + response.body().contentLength());
    }

    return response;
  }
}

参考

设计模式分析

本节对照 OkHttp 4.9.3 固定版本源码分析：parent-4.9.3，提交 0a27e9ff30c6129f4fdeb1f99c07c9b009bcc537。旧文保留的 Java/Kotlin 摘录不保证来自同一个版本；以下图中的类、字段和调用关系统一以这个版本为准。

UML 省略无关成员、泛型和异常；internal 成员用 ~ 简化表示模块内可见，嵌套类会标出外层名称。涉及 JDK、Okio 或可选模块的类型会单独说明，不虚构源码中不存在的角色类。

模式或设计思想 具体落点 解决的问题
建造者 OkHttpClient.Builder、Request.Builder 分步组织可选配置，构造客户端和请求。
工厂方法 Call.Factory.newCall()、OkHttpClient 把调用对象的创建隐藏在统一接口之后。
责任链 Interceptor、RealInterceptorChain 按顺序组合请求处理，并允许短路或重试下游。
策略 Dns、DnsSystem、DnsOverHttps 在路由流程不变的情况下替换域名解析算法。
命令思想 RealCall.AsyncCall、Dispatcher 把执行动作封装成可排队、调度的对象。
装饰者 Exchange.RequestBodySink、ResponseBodySource 在数据流读写外增加计数、校验和完成通知。
原型 Call.clone()、RealCall.clone() 从已有调用描述创建一个新的调用状态实例。
观察者式事件通知 EventListener.Factory、EventListener 将网络生命周期监控与请求实现分离。

建造者：分别组装客户端和请求

OkHttpClient 的配置包括连接池、调度器、拦截器、DNS、缓存、超时等；Request 则需要 URL、方法、请求头和请求体。Builder 将这些输入的收集过程与最终对象创建分开，避免调用者面对过长的构造参数列表。

类 模式角色 职责
OkHttpClient.Builder 客户端的具体建造者 提供默认配置及 addInterceptor()、dns()、connectTimeout() 等配置入口。
OkHttpClient 客户端产品 保存配置和共享组件，作为后续创建 Call 的入口。
Request.Builder 请求的具体建造者 收集 URL、方法和请求体等信息，校验后创建 Request。
Request 请求产品 描述一次请求；还未包含执行状态和网络连接。 
classDiagram
    direction TB
    class ClientBuilder["OkHttpClient.Builder"] {
        +addInterceptor(Interceptor interceptor) Builder
        +dns(Dns dns) Builder
        +connectTimeout(long timeout, TimeUnit unit) Builder
        +build() OkHttpClient
    }
    class OkHttpClient {
        +newBuilder() Builder
        +newCall(Request request) Call
    }
    class RequestBuilder["Request.Builder"] {
        +url(String url) Builder
        +method(String method, RequestBody body) Builder
        +build() Request
    }
    class Request {
        +newBuilder() Builder
    }
    ClientBuilder ..> OkHttpClient : build 创建
    OkHttpClient ..> ClientBuilder : newBuilder 复制配置
    RequestBuilder ..> Request : build 创建
    Request ..> RequestBuilder : newBuilder 复制描述
    OkHttpClient ..> Request : newCall 参数

调用者按需设置 Builder，最后调用 build()。Request.Builder.build() 检查 URL 是否缺失；方法与请求体是否匹配等条件在配置方法中校验。OkHttpClient.Builder 提供默认组件，构造出的客户端会复制拦截器列表并完成 TLS 等配置的初始化和检查。

newBuilder() 可从已有对象继续配置：派生客户端默认共享原来的 Dispatcher、连接池等组件；请求的新 Builder 复制请求头等结构，却不会深复制 RequestBody 或 tag 对象。Builder 的意义是组织构建过程，并不是承诺构造整个对象图的独立副本。这里也没有专门的 Director 类，配置步骤由应用代码组织。

源码：OkHttpClient 与 Builder、Request 与 Builder。

工厂方法：通过 Call.Factory 创建调用

使用方只要依赖 Call.Factory，就能根据 Request 获得 Call，无须知道 RealCall 的构造细节。OkHttpClient 实现 newCall()，返回与自身配置关联的新 RealCall。

类或接口 模式角色 职责
Call.Factory 创建者接口 定义 newCall(Request)，允许使用方注入不同创建实现。
OkHttpClient 具体创建者 用当前客户端、原始请求和普通 HTTP 标志构造 RealCall。
Call 产品接口 对外提供执行、取消、查询状态与克隆等操作。
RealCall 具体产品 保存请求及调用状态，并组织真正的执行过程。
Request 创建参数 向工厂描述要执行什么请求。 
classDiagram
    direction TB
    class CallFactory["Call.Factory"] {
        <<interface>>
        +newCall(Request request) Call
    }
    class OkHttpClient {
        +newCall(Request request) Call
    }
    class Call {
        <<interface>>
        +execute() Response
        +enqueue(Callback callback)
        +cancel()
    }
    class RealCall {
        +OkHttpClient client
        +Request originalRequest
    }
    class Request
    CallFactory <|.. OkHttpClient
    Call <|.. RealCall
    CallFactory ..> Call : 返回抽象产品
    OkHttpClient ..> RealCall : 创建
    RealCall --> OkHttpClient : client
    RealCall --> Request : originalRequest

newCall() 每次创建独立的调用对象，此时不会发送 HTTP 请求。网络工作由之后的 execute() 或 enqueue() 启动。一个客户端可以复用配置与资源，却可以创建许多独立 Call。

这是通过实现创建者接口提供工厂方法的形式，不依赖一个包含大量业务步骤的抽象父类。Call.Factory 是嵌套接口，OkHttpClient 实现它，并不继承 Call。这里创建的是一种产品抽象 Call，也不能因客户端还支持 WebSocket 创建就直接认定为抽象工厂。

源码：Call.Factory 与 Call 契约、OkHttpClient.newCall()、RealCall 的构造与执行入口。

责任链：拦截器共同完成请求与响应处理

拦截器接收 Interceptor.Chain，可以在调用 proceed(request) 前修改请求，在其返回后检查或包装响应，也可以在允许的阶段直接返回响应。RealInterceptorChain 保存拦截器列表和当前位置，并为下一步创建新的链上下文。

类或接口 模式角色 职责
Interceptor 处理者接口 通过 intercept(chain) 处理请求和响应。
Interceptor.Chain 链调用接口 暴露当前请求与 proceed()，连接后续处理流程。
RealInterceptorChain 分派上下文 保存 interceptors、index、请求及 Exchange 等状态，调用当前位置的拦截器。
CacheInterceptor 具体处理者 决定直接使用缓存、条件验证或继续访问网络。
CallServerInterceptor 末端处理者 通过 Exchange 写入请求并读取响应，不继续分派下一个拦截器。
RealCall 流程发起者 组装完整链并发起第一次 proceed()。 
classDiagram
    direction TB
    class RealCall {
        ~getResponseWithInterceptorChain() Response
    }
    class Chain["Interceptor.Chain"] {
        <<interface>>
        +request() Request
        +proceed(Request request) Response
    }
    class RealInterceptorChain {
        -List interceptors
        -int index
        ~copy() RealInterceptorChain
        +proceed(Request request) Response
    }
    class Interceptor {
        <<interface>>
        +intercept(Interceptor.Chain chain) Response
    }
    class CacheInterceptor {
        +intercept(Interceptor.Chain chain) Response
    }
    class CallServerInterceptor {
        +intercept(Interceptor.Chain chain) Response
    }
    Chain <|.. RealInterceptorChain
    Interceptor <|.. CacheInterceptor
    Interceptor <|.. CallServerInterceptor
    RealCall ..> RealInterceptorChain : 组装并启动
    RealInterceptorChain --> "1..*" Interceptor : 按 index 调用
    Interceptor ..> Chain : 接收并按需继续

完整顺序是：应用拦截器 → 重试与重定向 → Bridge → Cache → Connect → 网络拦截器 → CallServer。图中仅展开缓存和末端拦截器，省略其他实现；WebSocket 调用不会加入用户网络拦截器。

proceed() 使用 copy(index = index + 1, request = request) 创建下一个链对象，再调用当前 interceptor.intercept(next)。请求沿调用栈向后传递，响应随返回过程逐层回到前面的拦截器。具体拦截器不保存 next 字段，后继位置由 Chain 管理。

这不是“必须每层调用一次”的统一规则：

  • **应用拦截器**可以短路，也可以多次调用下游；再次调用之前必须关闭前一次响应体。
  • **缓存拦截器**命中可用缓存时可以直接返回；禁止网络且无可用缓存时也会直接返回 504。并非除了 CallServer 以外的所有拦截器都必然调用 proceed()。
  • **重试与重定向拦截器**可能多次进入下游，因此一次 Call 可以包含多次网络交换。
  • **网络拦截器**受到 Exchange 阶段的约束：必须保持同一主机和端口，并且恰好调用一次 proceed()；RealInterceptorChain 会检查这些条件。

和 Retrofit 的工厂列表查找相比，那里通常是“首个匹配者返回产品”；这里多个处理者可以参与同一次请求，并在响应返回时继续处理。责任链可以有不同实现形态，不能仅凭一个循环或类名判断它们完全相同。

源码：链的组装、RealInterceptorChain.proceed()、CacheInterceptor、RetryAndFollowUpInterceptor、CallServerInterceptor。

策略：替换 DNS 解析算法

路由选择需要一组候选 IP 地址，但不应把具体解析方式写死。OkHttp 用 Dns.lookup(hostname) 表达这个变化点：系统 DNS 和 DNS over HTTPS 都提供同一个契约，路由代码只消费解析结果。

类或接口 模式角色 职责
Dns 策略接口 根据主机名返回按尝试顺序排列的 IP 地址列表。
DnsSystem 默认具体策略 Dns 伴生对象中的私有实现，通过 InetAddress.getAllByName() 解析。
DnsOverHttps 可选具体策略 来自同仓库的 okhttp-dnsoverhttps 模块，通过 HTTPS 查询 DNS。
Address 策略的持有者 保存当前目标地址所使用的 Dns 等配置。
RouteSelector 策略的使用上下文 需要本地解析时调用 address.dns.lookup()，据结果构造候选套接字地址。
OkHttpClient 配置入口 从 Builder 接收 Dns；由 RealCall.createAddress() 将 client.dns 传入 Address。 
classDiagram
    direction TB
    class OkHttpClient {
        +Dns dns
    }
    class Address {
        +Dns dns
    }
    class RouteSelector {
        -Address address
    }
    class Dns {
        <<interface>>
        +lookup(String hostname) List
    }
    class DnsSystem {
        +lookup(String hostname) List
    }
    class DnsOverHttps {
        +lookup(String hostname) List
    }
    OkHttpClient --> Dns : 配置策略
    Address --> Dns : 保存策略
    RouteSelector --> Address : address
    RouteSelector ..> Dns : lookup
    Dns <|.. DnsSystem
    Dns <|.. DnsOverHttps

应用通过 OkHttpClient.Builder.dns() 选择策略，默认是 Dns.SYSTEM。在需要本地解析的连接路径中，RouteSelector 发出 DNS 生命周期事件，调用策略取得 IP 列表，再交给后续连接尝试。SOCKS 代理使用未解析的套接字地址，由代理处理解析，不能推断所有请求都会走本地 Dns.lookup()。

改变 DNS 实现不需要改写连接流程，但替代实现仍需满足线程安全和返回值契约。DnsOverHttps 属于可选模块，不是核心客户端默认启用的策略。认证器 Authenticator、Cookie 存储 CookieJar 也提供类似的可替换行为接口。

源码：Dns 与 DnsSystem、RouteSelector 的解析分支、Address、DnsOverHttps。

命令思想：把异步请求变为可调度的任务

Call 将请求描述与执行状态封装成对象；异步路径进一步创建 RealCall.AsyncCall，通过 Runnable 把“执行请求并交付回调”的动作交给调度器与线程池。调用的提交和真正执行因此可以分离。

类或接口 对应角色 职责
Runnable 可执行命令接口 JDK 提供的 run() 契约。
RealCall.AsyncCall 具体命令对象 保存回调并访问外层 RealCall，执行拦截器链、交付结果和通知调度完成。
Dispatcher 排队和调度者 维护待执行、执行中队列及异步并发限制，选择任务提交。
ExecutorService 实际执行者 接收 Runnable，在线程池中运行任务。
RealCall 执行逻辑的承载者 保存请求与调用状态，提供拦截器链执行入口。
Callback 结果接收接口 接收响应或失败通知，不是用于网络各阶段监控的 EventListener。 
classDiagram
    direction TB
    class Dispatcher {
        -ArrayDeque readyAsyncCalls
        -ArrayDeque runningAsyncCalls
        ~enqueue(AsyncCall call)
        -promoteAndExecute() boolean
        ~finished(AsyncCall call)
    }
    class Runnable {
        <<interface>>
        +run()
    }
    class AsyncCall["RealCall.AsyncCall"] {
        -Callback responseCallback
        +executeOn(ExecutorService executorService)
        +run()
    }
    class ExecutorService {
        <<interface>>
        +execute(Runnable task)
    }
    class RealCall {
        +enqueue(Callback callback)
        ~getResponseWithInterceptorChain() Response
    }
    class Callback {
        <<interface>>
        +onResponse(Call call, Response response)
        +onFailure(Call call, IOException error)
    }
    Runnable <|.. AsyncCall
    Dispatcher --> "0..*" AsyncCall : 排队与限流
    Dispatcher --> ExecutorService : 持有线程池
    AsyncCall ..> ExecutorService : executeOn 提交自身
    ExecutorService ..> Runnable : 执行 run
    AsyncCall --> RealCall : inner class 的外层对象
    AsyncCall --> Callback : responseCallback
    RealCall ..> Dispatcher : 提交任务

enqueue() 创建 AsyncCall 后交给 Dispatcher。promoteAndExecute() 根据总量及每主机限制选择可运行任务，将其移到执行中队列，然后在锁外调用 executeOn(executorService)。run() 执行请求并交付结果，在 finally 中通知 Dispatcher 释放名额并尝试推进下一批任务。

4.9.3 默认异步限制为总量 64、每主机 5；这是**异步任务调度限制**，不是连接池大小，也不是同步请求总数上限。execute() 在调用线程直接执行，只向 Dispatcher 登记与移除运行记录，不会进入同一个异步等待队列。线程池拒绝任务时，也会报告失败并进行调度清理，不能假设所有失败回调都在正常工作线程中发生。

这里借用命令模式解释请求对象化、排队和执行解耦；源码没有提供 GoF 示例常见的 undo() 或命令历史。cancel() 是尽力中断未完成的调用，并不能撤销服务器已经处理的副作用。

源码：RealCall 的同步与异步入口、AsyncCall、Dispatcher。

装饰者：在流读写外增加计数和生命周期管理

Exchange 从协议编解码器取得原始 Sink 或 Source 后，会添加包装层。包装对象仍然提供同一种流接口，同时记录传输字节数、检查部分长度条件，并将完成或失败情况通知调用流程。

类或接口 模式角色 职责
okio.Sink、okio.Source 组件接口 分别提供写入和读取字节的操作。
okio.ForwardingSink、okio.ForwardingSource 转发装饰者基类 持有同接口的 delegate，默认把读写、关闭和超时操作转交给它。
Exchange.RequestBodySink 请求侧具体装饰者 统计写入、检查长度、处理部分 I/O 异常并报告完成。
Exchange.ResponseBodySource 响应侧具体装饰者 统计读取、发出开始与结束事件，报告读取或关闭失败。
Exchange 包装层的创建者与收尾协作者 创建包装对象，接收 bodyComplete() 通知并调用 RealCall 的完成处理。 
classDiagram
    direction TB
    class Sink["okio.Sink"] {
        <<interface>>
        +write(Buffer source, long byteCount)
        +close()
    }
    class ForwardingSink["okio.ForwardingSink"] {
        <<abstract>>
        +Sink delegate
    }
    class RequestBodySink["Exchange.RequestBodySink"] {
        +write(Buffer source, long byteCount)
        +flush()
        +close()
    }
    class Source["okio.Source"] {
        <<interface>>
        +read(Buffer sink, long byteCount) long
        +close()
    }
    class ForwardingSource["okio.ForwardingSource"] {
        <<abstract>>
        +Source delegate
    }
    class ResponseBodySource["Exchange.ResponseBodySource"] {
        +read(Buffer sink, long byteCount) long
        +close()
    }
    Sink <|.. ForwardingSink
    ForwardingSink <|-- RequestBodySink
    ForwardingSink --> Sink : delegate
    Source <|.. ForwardingSource
    ForwardingSource <|-- ResponseBodySource
    ForwardingSource --> Source : delegate

请求侧先将协议编解码器返回的 Sink 包成 RequestBodySink，再加缓冲并交给 RequestBody.writeTo()；实际写入沿“RequestBody.writeTo() → 缓冲 Sink → RequestBodySink → 原始 Sink”调用。响应侧将原始 Source 包成 ResponseBodySource，再缓冲并交给 RealResponseBody。两个包装类都是 Exchange 的 inner 类，可以通过外层对象进行收尾；Exchange 创建它们，但没有专门的字段持续持有包装对象。

请求侧在成功写入后增加计数，关闭时检查已知长度等条件；响应侧在读取、EOF 或关闭等时机推进完成处理。complete() 用标志防止重复通知，Exchange.bodyComplete() 再更新请求/响应体事件并调用 RealCall.messageDone()。因此这不只是一个统计包装层，也参与调用资源的生命周期管理。

RequestBodySink 装饰的是 Sink，并不继承 RequestBody；ResponseBodySource 装饰的是 Source，并不继承 ResponseBody。部分异常检查在委托调用的捕获范围之外，响应包装器遇到 EOF 的分支也没有单独校验“是否已读满声明长度”，因此不应夸大为统一捕获所有错误或包办全部协议校验。

这两个转发基类来自 OkHttp 4.9.3 依赖的 Okio 2.8.0，不是 OkHttp 自己声明的类型。

源码：Exchange 的包装与完成处理、RealResponseBody、依赖版本配置、Okio ForwardingSink、Okio ForwardingSource。

原型：新建调用状态，复用原始请求描述

同一个 Call 的 execute() 和 enqueue() 都受执行标志保护，不能再次执行。clone() 则提供从现有调用创建新 Call 的统一契约，调用者不需要知道 RealCall 的构造方式。

类或接口 模式角色 职责
Call 原型接口 声明 clone(): Call，提供可再次执行的新调用入口。
RealCall 具体原型 通过构造器创建新的调用状态对象。
OkHttpClient 共享配置与资源入口 克隆前后使用同一个客户端。
Request、RequestBody 共享的请求描述与内容 克隆使用同一 originalRequest，因此也不会复制请求体的数据源或写入状态。 
classDiagram
    direction TB
    class Call {
        <<interface>>
        +clone() Call
    }
    class RealCall {
        +OkHttpClient client
        +Request originalRequest
        -AtomicBoolean executed
        -boolean canceled
        +clone() RealCall
    }
    class OkHttpClient
    class Request {
        +RequestBody body
    }
    class RequestBody {
        <<abstract>>
        +isOneShot() boolean
        +isDuplex() boolean
    }
    Call <|.. RealCall
    RealCall ..> RealCall : clone 创建新实例
    RealCall --> OkHttpClient : 共享 client
    RealCall --> Request : 共享 originalRequest
    Request --> "0..1" RequestBody : body

RealCall.clone() 实际返回 RealCall(client, originalRequest, forWebSocket)，并未调用 super.clone()。新的对象重新初始化执行标志、取消状态、超时和交换状态，也会重新调用事件监听器工厂；已有网络交换不会被复制。客户端与原始请求则继续共享。

新 Call 可以被执行，不代表原请求体一定可以重放。 RequestBody.isOneShot() 表示内容至多接受一次 writeTo()，clone 不会因此拒绝创建，也不会重新打开已经消耗的数据源。自动失败恢复会结合请求是否已经开始发送及一次性标志决定能否重试；手动克隆同样需要调用者确认内容可再次提供。

isDuplex() 表示请求和响应可以交错读写，与 isOneShot() 是两个独立标志。该版本双工传输要求 HTTP/2；不能简单写成“所有双工请求都禁止克隆或重试”。共享同一个请求体时，它的异步写入状态和并发使用能力也没有被复制或重置。

前面的 newBuilder() 则用于派生配置或请求描述：虽然同样从已有实例出发，它返回的是 Builder，仍需继续构建产品。两类 API 都有共享边界，不能把它们统称为深克隆整个请求运行环境。

源码：Call.clone() 契约、RealCall.clone()、RequestBody 的一次性和双工说明、重试与后续请求判断、HTTP/1 的双工限制。

观察者式事件通知：记录一次调用的生命周期

网络监控通常关心 DNS、连接、TLS、请求体和响应体等阶段。如果直接把统计逻辑写进每个阶段，就会让网络实现与具体监控系统耦合。EventListener 用统一事件接口接收这些通知，应用可以选择关心的事件进行记录。

类或接口 对应角色 职责
EventListener 事件观察接口的抽象类 提供 callStart()、dnsStart()、responseBodyEnd() 等默认空实现方法，子类按需覆写。
EventListener.Factory 监听器创建接口 在创建 Call 时提供其生命周期使用的监听器。
OkHttpClient 监听配置持有者 保存 eventListenerFactory，由 Builder 配置。
RealCall 监听器作用域的持有者 构造时调用工厂并保存一个监听器引用，发送调用开始、结束和取消等事件。
Exchange、RouteSelector 事件产生者 接收同一 Call 的监听器引用，分别通知传输和 DNS 等阶段事件。 
classDiagram
    direction TB
    class OkHttpClient {
        +EventListener.Factory eventListenerFactory
    }
    class ListenerFactory["EventListener.Factory"] {
        <<interface>>
        +create(Call call) EventListener
    }
    class RealCall {
        ~EventListener eventListener
    }
    class EventListener {
        <<abstract>>
        +callStart(Call call)
        +dnsStart(Call call, String domainName)
        +responseBodyEnd(Call call, long byteCount)
        +callEnd(Call call)
    }
    class Exchange {
        ~EventListener eventListener
    }
    class RouteSelector {
        -EventListener eventListener
    }
    OkHttpClient --> ListenerFactory : eventListenerFactory
    RealCall ..> ListenerFactory : 构造时 create
    ListenerFactory ..> EventListener : 返回
    RealCall --> EventListener : 保存并通知
    Exchange --> EventListener : 使用同一调用的监听器
    RouteSelector --> EventListener : 使用同一调用的监听器

创建发生在 RealCall 的属性初始化阶段：client.eventListenerFactory.create(this)，不是第一次执行时，也不是每次重试时。后续创建 Exchange、ExchangeFinder 和 RouteSelector 时，会把同一 Call 的监听器向下传递。一项 Call 内的重试可能重复产生部分阶段事件，但仍然属于这个 Call 的监听范围。

“每个 Call 调用一次工厂”不保证“每个 Call 都得到一个新监听器实例”。Builder.eventListener(listener) 会把已有实例包装成始终返回它的工厂；默认的 EventListener.NONE 也是共享的空实现。若要记录每个调用独立的可变数据，应由工厂提供合适的实例作用域。

它具有观察者式的通知解耦，但核心实现是**每 Call 一个监听器引用**，没有 addObserver/removeObserver、动态订阅列表或遍历广播。EventListener.Factory 也不继承 EventListener。图中关联表示使用关系，不表示监听器必须被某个 Call 独占。

事件方法在触发事件的线程上调用，没有独立统一的通知线程；取消可能与其他事件并发。它和 Callback 的用途也不同：EventListener 用于过程监控，Callback 用于交付异步结果。

源码：EventListener 及其 Factory、RealCall 的监听器初始化、Builder 的监听配置、asFactory()、ExchangeFinder 的传递。

连接池与旧摘录的阅读边界

连接复用是这里另一个重要设计，但更适合按**连接资源池**理解。RealConnectionPool 管理 RealConnection，获取时检查连接是否允许复用、地址与路由是否匹配、是否要求多路复用；清理任务根据空闲时长和空闲数量淘汰连接。连接包含 socket、协议与使用状态，不能只因为“多个调用共享”就直接认定为 GoF 享元。

需要区分四种对象：Request 描述要做什么，Call 保存本次调用状态，Exchange 表示一次网络交换，RealConnection 表示可复用的连接。一项 Call 可能因重试或重定向经历多项 Exchange；多个 Call 也可能先后复用连接，HTTP/2 下还可以在符合条件时共享同一条连接上的不同流。

默认的“5 个、5 分钟”指连接池的**最大空闲连接数与空闲保活时长**，不表示最多只能创建 5 个连接。连接池的自动空闲清理由 TaskRunner 的任务队列安排；不能把所有清理描述成调用 put() 的线程立即循环关闭连接。异步并发限制则由前面的 Dispatcher 管理。

容易按名称或表面结构误判的地方 对照源码应如何理解
建议全局复用 OkHttpClient 是资源复用建议。客户端允许创建多个实例，不是强制单例类。
BridgeInterceptor 负责用户请求与 HTTP 传输细节的衔接，如补请求头、Cookie、透明 gzip；名称不足以证明 GoF 桥接结构。
CacheStrategy 保存 networkRequest、cacheResponse 等缓存决策结果。不能仅凭 Strategy 后缀认为它自身就是可替换算法接口；前文用 Dns 展示更明确的策略结构。
所有拦截器都要继续链 缓存命中或应用层合成响应可以短路；网络拦截器才有特定的恰好一次约束。
Callback、EventListener 都是回调 前者交付异步请求结果，后者观察生命周期；它们不是同一个订阅机制。
Java 片段与 Kotlin 片段混合 旧摘录中的字段、构造方式未必属于同一版本；新增 UML 以本节固定的 4.9.3 源码为准。

源码：ConnectionPool、RealConnectionPool、BridgeInterceptor、CacheStrategy。

评论