okhttp实现分析

本文概要

  1. okhttp拦截器实现
  2. 代理和路由
  3. 连接池实现
  4. 任务调度

1、okhttp拦截器实现

首先看下okhttp的简单使用：

val client = OkHttpClient()

val request = Request.Builder()
    .url("https://publicobject.com/helloworld.txt")
    .build()

client.newCall(request).execute().use { response ->
  if (!response.isSuccessful) throw IOException("Unexpected code $response")
  // print headers
  for ((name, value) in response.headers) {
    println("$name: $value")
  }
  // print body
  println(response.body!!.string())
}

OkHttpClient理解成一个配置类，用于设置一些基础参数（比如：代理，DNS，SSL，超时时间等等），然后通过newCall将这些基础参数和request参数组装成HTTP请求对象RealCall。
再通过execute()方法执行http请求，内部调用了getResponseWithInterceptorChain()方法，这个方法则通过一系列Interceptor嵌套调用得到Response，Response包含了响应Headers和ResponseBody(内部通过okio实现的字节流)，拿到字节流就可以读取http请求数据了。

getResponseWithInterceptorChain() 方法部分代码如下:

fun getResponseWithInterceptorChain(): Response {
    // 请注意Interceptor的添加顺序，proceed方法按照这个顺序调用的拦截器的
    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)
    
    val chain = RealInterceptorChain(interceptors, transmitter, null, 0, originalRequest, this,
        client.connectTimeoutMillis, client.readTimeoutMillis, client.writeTimeoutMillis)
    
    // ...
    val response = chain.proceed(originalRequest)
    // ...
    return response
}

整个调用流程如下图：


image

上图包含了okhttp内置的5个拦截器，主要功能如下：

RetryAndFollowUpInterceptor

在请求抛出异常时负责重试，但需要满足一定条件才会重试。比如ssl握手失败就不会重试，建立连接成功在读写数据出现IOException时会重试。后面会专门介绍重试机制！

BridgeInterceptor

负责桥接应用层和网络层代码，在网络层请求前后做一些准备和善后操作，比如设置请求头（Content-Length，Accept-Encoding，Cookie等），网络层请求完成后对结果ResponseBody进行处理(比如gzip解压)

CacheInterceptor

负责缓存查找处理，如果缓存命中则后面的拦截器就不会执行，直接返回Response

ConnectInterceptor

负责建立HTTP连接：建立新的连接或从连接池中找到可复用的连接。ConnectInterceptor相对其他拦截器功能较多：可复用连接查找，路由查找，域名解析，Socket连接，建立协议版本（HTTP 1.1或HTTP 2），SSL握手等等

CallServerInterceptor

建立连接后跟服务器端交互，主要是写入request和读取response，内部通过Exchange对象完成，Exchange对象实现了对http读写的基本操作（比如写入header，写入body，读取header等），Exchange内部则使用ExchangeCodec格式化写入和读取数据，ExchangeCodec有两个实现，分别是：Http1ExchangeCodec和Http2ExchangeCodec，对应HTTP1和HTTP2

另外除了上述的5个内置拦截器外，okhttp还提供了两类自定义扩展拦截器：Application Interceptors 和 Network Interceptors，在OKHttpClient初始化时可通过addInterceptor()和addNetworkInterceptor()添加，从图中可以看出Application Interceptors是最开始执行的拦截器，是建立TCP连接前调用的，Network Interceptors是在连接建立后调用的，借用官方的图更具体一点：

image

两者的区别主要根据他们在拦截器调用链中的位置来分析，如下：

Application Interceptors

  • 不需要关心是否重定向或者失败重连
  • 应用拦截器只会调用一次，因为它比RetryAndFollowUpInterceptor先执行
  • 应用拦截器是第一个执行的，可以决定是否调用其他拦截器，(不调用Chain.proceed()则表示拦截剩余的拦截器)

Network Interceptors

  • 网络拦截器是在连接拦截器后调用的，可以执行重定向和重连操作
  • 可以查看连接相关信息

拦截器调用链实现

整个okhttp拦截器调用流程还是挺清晰的，拦截器调用是通过RealInterceptorChain.proceed(request)和Inteceptor.intercept(chain)两个方法结合实现。

RealInterceptorChain

  • 内部包含了所有拦截器的列表interceptors和当前要执行的拦截器在列表中的索引index
  • 执行proceed()方法时会取出当前index对应的Interceptor，同时将index加一后保存到一个新的RealInterceptorChain中(除了index不一样，其他数据都一样)，再将这个chain作为参数调用前面取得的Interceptor的拦截方法intercept(chain)
  • 如果intercept(chain)方法内部继续调用chain.proceed()就会执行下一个拦截器，这样类似递归调用使所有拦截器按顺序执行，直到最后一个拦截器（CallServerInterceptor）
  • 如果某个拦截器intercept(chain)方法内部不调用chain.proceed()，那么后面剩余的拦截器就没有机会执行

RealInterceptorChain.proceed() 方法部分代码如下：

fun proceed(request: Request, transmitter: Transmitter, exchange: Exchange?): Response {
    ...
    // index + 1记录下一个待执行的拦截器
    val next = RealInterceptorChain(interceptors, transmitter, exchange,
        index + 1, request, call, connectTimeout, readTimeout, writeTimeout)
    // 取出当前拦截器
    val interceptor = interceptors[index]
    ...
    // 执行当前拦截器
    val response = interceptor.intercept(next) ?: throw NullPointerException(
        "interceptor $interceptor returned null")
    ...
    return response
  }

Interceptor

  • 只有一个方法：fun intercept(chain: Chain): Response
  • 参数chain对象内包含了所有拦截器列表和下一个要执行的拦截器，如果调用chain.proceed()就会执行下一个拦截器

自定义拦截器示例

class MyInterceptor:Interceptor {
    override fun intercept(chain: Interceptor.Chain): Response {
        // do something
        
        // 这里也可以不调用chain.proceed(chain.reqest())，这样chain里面剩余的拦截器就不会执行
        val response = chain.proceed(chain.request())
        
        // do something
        return response
    }
}

到此，okhttp的拦截器实现就这些，拦截器可以拦截(比如CacheInterceptor)，也可以不拦截(比如LogInterceptor)，更多的是提供了一种面向切面编程的能力，在下游实现的前后做一些加工处理，或者直接拦截掉下游的执行！

2、路由和代理

什么是okhttp路由，通过域名发起HTTP请求，首先要将域名转成IP这一过程叫域名解析，也就是DNS协议干的事！okhttp的路由就是通过request的域名或OKHttpClient设置的代理（如过有设置代理）找到合适的IP地址！

几个主要类

Dns

抽象了DNS解析过程，抽象方法为：fun lookup(hostname: String): List<InetAddress>

Route

DNS解析完成后的对象，可以简单的理解为包含了IP地址的对象，内部属性socketAddress为DNS解析后的地址

RouteDatabase

记录了之前连接成功，但读写过程中失败了对应的路由，后续再路由选择时会优先排除这里记录的路由（除非一个可用的都没有）

RouteSelector

路由选择实现，RouteSelector有以下工功能：

  • 收集所有（代理代理由OkHttpClient.Builder.proxy(proxy)和OkHttpClient.Builder.proxySelector(selector)配置， 默认会使用proxy，在proxy不存在的情况下才会使用proxySelector）！
  • 通过域名或代理的域名 调用dns.lookup(socketHost)进行域名解析
  • RouteDatabase会记录请求失败的路由，路由选择时会优先排除这部分路由，除非没有其他路由可用了，就不排除

DNS解析可以通过OkHttpClient.Builder.dns(dns)方法配置，默认是调用系统DNS解析InetAddress.getAllByName(hostname)， 如下：

val SYSTEM: Dns = DnsSystem()
private class DnsSystem : Dns {
  override fun lookup(hostname: String): List<InetAddress> {
    try {
      return InetAddress.getAllByName(hostname).toList()
    } catch (e: NullPointerException) {
      throw UnknownHostException("Broken system behaviour for dns lookup of $hostname").apply {
        initCause(e)
      }
    }
  }
}

3、连接池实现（连接复用）

首先TCP连接复用需要Keep-Alive，所以在BridgeInterceptor中如果用户没有设置Connection，则默认将Connection有设置为: Keep-Alive

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

但实际连接是否可复用还得取决于Server返回的Connection来决定，如果Server返回close，则将TCP连接标识为不可复用，这部分代码在CallServerInterceptor.intercept()方法中:

// 如果response返回 Connection为close，则将连接标识为不可复用
if ("close".equals(response.request.header("Connection"), ignoreCase = true) ||
    "close".equals(response.header("Connection"), ignoreCase = true)) {
  // 内部是将RealConnection标识为不可复用
  exchange.noNewExchangesOnConnection()
}

连接可复用后，就需要对其进行管理，由RealConnectionPool对象实现，主要方法如下:

  • fun put(connection: RealConnection)

    当新的连接建立时，会调用该方法将连接加入到连接池

  • fun transmitterAcquirePooledConnection(
    address: Address,
    transmitter: Transmitter,
    routes: List<Route>?,
    requireMultiplexed: Boolean
    ): Boolean

    该方法为查找可复用的连接，判断条件有：连接是否可复用，连接是否超过最大复用数（HTTP1.1只能为1，HTTP2默认为4），request请求参数匹配（包括：代理，协议版本，ssl，端口号等等）

  • fun connectionBecameIdle(connection: RealConnection): Boolean

    通知连接池，该连接空闲，如果返回true说明该连接已从连接池移除，则需手动关闭连接

  • fun cleanup(now: Long): Long

    清理连接，并返回下一次清理的间隔时间。满足这些条件的连接会从连接池中移除关闭Scoket：空闲时间超过最大值，被标识为不能再复用

连接加入连接池

在RealConnection为单个连接对象，在RealConnection.connect()方法连接成功后会将其加入连接池，源码在ExchangeFinder.findConnection方法中:

synchronized(connectionPool) {
  connectingConnection = null
  // Last attempt at connection coalescing, which only occurs if we attempted multiple
  // concurrent connections to the same host.
  if (connectionPool.transmitterAcquirePooledConnection(address, transmitter, routes, true)) {
    // 省略代码
  } else {
    // 新建的连接加入连接池
    connectionPool.put(result!!)
    transmitter.acquireConnectionNoEvents(result!!)
  }
}
连接复用查找

在建立新的TCP连接前会调用transmitterAcquirePooledConnection方法从连接池查找可复用连接，如果没有查询到可复用的连接才会新建一个连接，代码在ExchangeFinder.findConnection()方法中:

// Attempt to get a connection from the pool.
if (connectionPool.transmitterAcquirePooledConnection(address, transmitter, null, false)) {
  foundPooledConnection = true
  result = transmitter.connection
} else if (nextRouteToTry != null) {
  selectedRoute = nextRouteToTry
  nextRouteToTry = null
} else if (retryCurrentRoute()) {
  selectedRoute = transmitter.connection!!.route()
}
连接复用查找匹配规则

RealConnection内部维护了一个被上层引用的Transmitter对象列表transmitters，transmitters的个数就为复用数，默认一个连接使用HTTP 1.1同时只能有一个request在处理，即transmitters最大size只能为1，当HTTP请求完成时会释放这个连接，即将Transmitter的引用从transmitters列表中移除，这时size为0就可以被其他请求服用了。HttP 2默认最大允许4个request。

连接复用匹配过程主要在RealConnection.isEligible和RealConnection.equalsNonHost方法中：

internal fun isEligible(address: Address, routes: List<Route>?): Boolean {
    // 如果超过最大复用数或者已标识为不可复用，表明不能复用
    if (transmitters.size >= allocationLimit || noNewExchanges) return false

    // 除host以外的参数如果都匹配不上，不可复用
    if (!this.route.address.equalsNonHost(address)) return false

    // 如果host相同，则可复用
    if (address.url.host == this.route().address.url.host) {
      return true // This connection is a perfect match.
    }

    // 1. This connection must be HTTP/2.
    if (http2Connection == null) return false

    // 2. The routes must share an IP address.
    if (routes == null || !routeMatchesAny(routes)) return false

    // 3. This connection's server certificate's must cover the new host.
    if (address.hostnameVerifier !== OkHostnameVerifier) return false
    if (!supportsUrl(address.url)) return false

    // 4. Certificate pinning must match the host.
    try {
      address.certificatePinner!!.check(address.url.host, handshake()!!.peerCertificates)
    } catch (_: SSLPeerUnverifiedException) {
      return false
    }
    return true // The caller's address can be carried by this connection.
}

// 匹配除了host以外的参数
internal fun equalsNonHost(that: Address): Boolean {
return this.dns == that.dns &&
    this.proxyAuthenticator == that.proxyAuthenticator &&
    this.protocols == that.protocols &&
    this.connectionSpecs == that.connectionSpecs &&
    this.proxySelector == that.proxySelector &&
    this.proxy == that.proxy &&
    this.sslSocketFactory == that.sslSocketFactory &&
    this.hostnameVerifier == that.hostnameVerifier &&
    this.certificatePinner == that.certificatePinner &&
    this.url.port == that.url.port
}
连接何时被清理
  • 如果后台返回的Header的Connection字段为Close；说明不能复用连接，则将RealConnection标识为不能再复用（将其内部变量赋值为：noNewExchanges = true）
  • HTTP读写出现异常时也为将noNewExchanges = true
  • RealConnectionPool.cleanup()方法会清理掉空闲超过最大keepAliveDurationNs的连接
  • 在连接被使用完后会调用Transmitter.releaseConnectionNoEvents()方法，会将RealConnection内部保存的引用移除，如果连接已经不可复用了就将其Socket关闭掉！
fun releaseConnectionNoEvents(): Socket? {
  assert(Thread.holdsLock(connectionPool))

  val index = connection!!.transmitters.indexOfFirst { it.get() == this@Transmitter }
  check(index != -1)

  val released = this.connection
  released!!.transmitters.removeAt(index)
  this.connection = null

  if (released.transmitters.isEmpty()) {
    released.idleAtNanos = System.nanoTime()
    if (connectionPool.connectionBecameIdle(released)) {
      return released.socket()
    }
  }
  return null
}
Response.close()方法

另外说一下Response.close()方法，其实并不会关闭Socket，如果真关闭了那还怎么复用，但是close会做一些善后工作，比如用户发起了一个HTTP请求，写入了request，但是没有读取Response就直接调用了close了，这时close()方法会触发读取Server返回的数据操作，并丢弃掉这部分数据来保证该连接可被复用，如果不读取这部分数据，后面复用该连接，就会读取到上一个请求的错误数据！

close()实际调用的如下方法，如果还有未读数据，其中discard()方法会读取未读的数据并丢弃：

// Http1ExchangeCodec.ChunkedSource.close()
override fun close() {
  if (closed) return
  // 这里中discard方法为丢弃未读取的数据
  if (hasMoreChunks &&
      !discard(ExchangeCodec.DISCARD_STREAM_TIMEOUT_MILLIS, MILLISECONDS)) {
    realConnection!!.noNewExchanges() // Unread bytes remain on the stream.
    responseBodyComplete()
  }
  closed = true
}

// Http1ExchangeCodec.FixedLengthSource.close()
override fun close() {
  if (closed) return
  // 这里中discard方法为丢弃未读取的数据
  if (hasMoreChunks &&
      !discard(ExchangeCodec.DISCARD_STREAM_TIMEOUT_MILLIS, MILLISECONDS)) {
    realConnection!!.noNewExchanges() // Unread bytes remain on the stream.
    responseBodyComplete()
  }
  closed = true
}

4、任务调度

任务调度由Dispatcher实现，主要实现了请求异步执行和一些限制，比如同时最多并行64个请求，同一个host最大并行请求为5个，这部分比较简单，主要代码如下：

private fun promoteAndExecute(): Boolean {
  assert(!Thread.holdsLock(this))

  val executableCalls = mutableListOf<AsyncCall>()
  val isRunning: Boolean
  synchronized(this) {
    val i = readyAsyncCalls.iterator()
    while (i.hasNext()) {
      val asyncCall = i.next()

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

      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
}

最后看下okhttp主要对象的类图

image

参考

okhttp源码（4.2.2）

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 199,478评论 5赞 467
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 83,825评论 2赞 376
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 146,482评论 0赞 330
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 53,726评论 1赞 271
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 62,633评论 5赞 359
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,018评论 1赞 275
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,513评论 3赞 390
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,168评论 0赞 254
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,320评论 1赞 294
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,264评论 2赞 317
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,288评论 1赞 328
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 32,995评论 3赞 315
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,587评论 3赞 303
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,667评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,909评论 1赞 255
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 42,284评论 2赞 345
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 41,862评论 2赞 339