[网络协议]OKHttp

1 OkHttp简介

目前主流的Android网络请求框架有OkHttp和Retrofit，不过，Retrofit底层使用的是OKHttp，其自身是不具备网络请求能力的。

OkHttp是由Square公司开发并共享开源的高效网络访问框架，使用简单，它替代了HttpUrlConnection和Apache的HttpClient。

默认情况下OkHttp具备以下特性：

• 支持HTTP/2，允许同一个主机地址的所有请求共享同一个Socket连接
• 连接池减少请求延时
• 透明的GZIP压缩较少响应数据的大小
• 缓存响应内容，避免一些完全重复的请求
• 网络出现问题后，OkHttp保持不变，自动从问题中恢复

下面是Http常见的一些状态码：

• 100~199：指示信息，表示请求已接收，继续处理；
• 200~299：请求成功，表示请求已被成功接收、理解；
• 300~399：重定向，要完成请求必须进行更进一步的操作；
• 400~499：客户端错误，请求有语法错误或请求无法实现；
• 500~599：服务器端错误，服务器未能实现合法的请求；

2 OkHttp请求流程

以下是OkHttp发起请求的大致流程：

下面是OkHttp进行GET请求/POST请求的代码：

// GET请求
private val client = OkHttpClient() // 新建OkHttpClient客户端
fun getMethod(url: String): String {
    val request = Request.Builder().url(url).build() // 新建Request对象
    val response = client.newCall(request).execute() // Response为OkHttp中的响应
    return response.body?.string() ?: ""
}

// POST请求
val JSON = "application/json; charset=utf-8".toMediaType()
fun postMethod(url: String, json: String): String {
   val body = RequestBody.create(JSON, json)
   val request = Request.Builder().url(url).post(body).build()
   val response = client.newCall(request).execute()
   return response.body?.string() ?: ""
}

可以看到，使用OkHttp进行请求时，只需要创建一个OkHttpClient对象和Request对象，然后再调用execute()方法和enqueue()方法即可。其中，execute()方法是同步方法，enqueue()方法是异步方法。

2.1 OkHttpClient

在使用OkHttpClient之前，需要先创建一个OkHttpClient客户端，OkHttpClient的构造方法如下：

open class OkHttpClient internal constructor(
  builder: Builder
) : Cloneable, Call.Factory, WebSocket.Factory {  
  constructor() : this(Builder())
}

可以看到，OkHttpClient使用了建造者模式，Builder里面的可配置参数如下：

open class OkHttpClient internal constructor(
  builder: Builder
) : Cloneable, Call.Factory, WebSocket.Factory {  
  constructor() : this(Builder())
  
  class Builder constructor() {
    internal var dispatcher: Dispatcher = Dispatcher() // 分发器
    internal var connectionPool: ConnectionPool = ConnectionPool() // 连接池
    internal val interceptors: MutableList<Interceptor> = mutableListOf() // 拦截器
    internal val networkInterceptors: MutableList<Interceptor> = mutableListOf()
    internal var eventListenerFactory: EventListener.Factory = EventListener.NONE.asFactory()
    internal var retryOnConnectionFailure = true // 重试连接失败
    internal var authenticator: Authenticator = Authenticator.NONE // 身份验证
    internal var followRedirects = true // 本地重定向
    internal var followSslRedirects = true // 安全套接层重定向
    internal var cookieJar: CookieJar = CookieJar.NO_COOKIES
    internal var cache: Cache? = null
    internal var dns: Dns = Dns.SYSTEM
    internal var proxy: Proxy? = null
    internal var proxySelector: ProxySelector? = null
    internal var proxyAuthenticator: Authenticator = Authenticator.NONE // 代理身份验证
    internal var socketFactory: SocketFactory = SocketFactory.getDefault()
    internal var sslSocketFactoryOrNull: SSLSocketFactory? = null // 安全套接层socket，工厂，用于HTTPS
    internal var x509TrustManagerOrNull: X509TrustManager? = null
    internal var connectionSpecs: List<ConnectionSpec> = DEFAULT_CONNECTION_SPECS // 传输层版本和连接协议
    internal var protocols: List<Protocol> = DEFAULT_PROTOCOLS
    internal var hostnameVerifier: HostnameVerifier = OkHostnameVerifier // 验证确认响应证书，适用HTTPS请求连接的主机
    internal var certificatePinner: CertificatePinner = CertificatePinner.DEFAULT
    internal var certificateChainCleaner: CertificateChainCleaner? = null // 验证确认响应证书，适用HTTPS请求连接的主机名
    internal var callTimeout = 0
    internal var connectTimeout = 10_000
    internal var readTimeout = 10_000
    internal var writeTimeout = 10_000
    internal var pingInterval = 0
    internal var minWebSocketMessageToCompress = RealWebSocket.DEFAULT_MINIMUM_DEFLATE_SIZE
    internal var routeDatabase: RouteDatabase? = null

    internal constructor(okHttpClient: OkHttpClient) : this() {
      this.dispatcher = okHttpClient.dispatcher
      this.connectionPool = okHttpClient.connectionPool
      this.interceptors += okHttpClient.interceptors
      this.networkInterceptors += okHttpClient.networkInterceptors
      this.eventListenerFactory = okHttpClient.eventListenerFactory
      this.retryOnConnectionFailure = okHttpClient.retryOnConnectionFailure
      this.authenticator = okHttpClient.authenticator
      this.followRedirects = okHttpClient.followRedirects
      this.followSslRedirects = okHttpClient.followSslRedirects
      this.cookieJar = okHttpClient.cookieJar
      this.cache = okHttpClient.cache
      this.dns = okHttpClient.dns
      this.proxy = okHttpClient.proxy
      this.proxySelector = okHttpClient.proxySelector
      this.proxyAuthenticator = okHttpClient.proxyAuthenticator
      this.socketFactory = okHttpClient.socketFactory
      this.sslSocketFactoryOrNull = okHttpClient.sslSocketFactoryOrNull
      this.x509TrustManagerOrNull = okHttpClient.x509TrustManager
      this.connectionSpecs = okHttpClient.connectionSpecs
      this.protocols = okHttpClient.protocols
      this.hostnameVerifier = okHttpClient.hostnameVerifier
      this.certificatePinner = okHttpClient.certificatePinner
      this.certificateChainCleaner = okHttpClient.certificateChainCleaner
      this.callTimeout = okHttpClient.callTimeoutMillis
      this.connectTimeout = okHttpClient.connectTimeoutMillis
      this.readTimeout = okHttpClient.readTimeoutMillis
      this.writeTimeout = okHttpClient.writeTimeoutMillis
      this.pingInterval = okHttpClient.pingIntervalMillis
      this.minWebSocketMessageToCompress = okHttpClient.minWebSocketMessageToCompress
      this.routeDatabase = okHttpClient.routeDatabase
    } 
  }
}

2.2 同步请求

同步请求使用的是execute方法，如下所示：

val response = client.newCall(request).execute()

以下是源码：

open class OkHttpClient internal constructor(
  builder: Builder
) : Cloneable, Call.Factory, WebSocket.Factory {

  override fun newCall(request: Request): Call = RealCall(this, request, forWebSocket = false)  
}

class RealCall( // RealCall为真正的请求执行者
  val client: OkHttpClient,
  /** The application's original request unadulterated by redirects or auth headers. */
  val originalRequest: Request,
  val forWebSocket: Boolean
) : Call {
  internal val eventListener: EventListener = client.eventListenerFactory.create(this)

  val call: RealCall
  			get() = this@RealCall

  override fun execute(): Response {
    check(executed.compareAndSet(false, true)) { "Already Executed" } // 每个call只能执行一次

    timeout.enter()
    callStart()
    try {
      client.dispatcher.executed(this) // 通过dispatcher已经进入执行状态
      return getResponseWithInterceptorChain() // 通过一系列的拦截器请求处理和响应处理得到最终的返回结果
    } finally {
      client.dispatcher.finished(this) // 通知dispatcher，已经执行完毕
    }
  }
  
   @Throws(IOException::class)
  internal fun getResponseWithInterceptorChain(): Response {
    // Build a full stack of interceptors.
    val interceptors = mutableListOf<Interceptor>()
    interceptors += client.interceptors // 在配置OkHttpClient时设置的interceptors
    interceptors += RetryAndFollowUpInterceptor(client) // 负责失败重试以及重定向
    interceptors += BridgeInterceptor(client.cookieJar) // 请求时，对必要的Header进行一些添加，接收响应时，移除非必要的Header
    interceptors += CacheInterceptor(client.cache) // 负责读取缓存直接返回、更新缓存
    interceptors += ConnectInterceptor // 负责和服务器建立连接
    if (!forWebSocket) {
      interceptors += client.networkInterceptors // 配置OkHttpClient时设置的networkInterceptors
    }
    // 负责向服务器发送请求数据、从服务器读取响应数据
    interceptors += CallServerInterceptor(forWebSocket)

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

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"
      }
    }

    // Call the next interceptor in the chain.
    // 实例化下一个拦截器对应的RealInterceptorChain对象
    val next = copy(index = index + 1, request = request)
    // 得到当前的拦截器
    val interceptor = interceptors[index]

    // 调用当前拦截器的intercept()方法，并将洗一个拦截器的RealInterceptorChain对象传递下去，最后得到响应
    @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
  }
}

2.3 异步请求

异步请求使用的是enqueue方法，如下所示：

private val client = OkHttpClient()

// GET请求
fun run() {
  val request = Request.Builder()
  .url("http://publicobject.com/helloworld.txt")
  .build()

  client.newCall(request).enqueue(object : Callback {
    override fun onFailure(call: Call, e: IOException) {
      e.printStackTrace()
    }

    override fun onResponse(call: Call, response: Response) {
      response.use {
        if (!response.isSuccessful) throw IOException("Unexpected code $response")

        for ((name, value) in response.headers) {
          println("$name: $value")
        }

        println(response.body!!.string())
      }
    }
  })
}

// POST请求
fun run() {
  val formBody = FormBody.Builder()
  .add("search", "Jurassic Park")
  .build()
  val request = Request.Builder()
  .url("https://en.wikipedia.org/w/index.php")
  .post(formBody)
  .build()

  client.newCall(request).execute().use { response ->
      if (!response.isSuccessful) throw IOException("Unexpected code $response")
           println(response.body!!.string())
  }
}

以下是源码：

open class OkHttpClient internal constructor(
  builder: Builder
) : Cloneable, Call.Factory, WebSocket.Factory {

  override fun newCall(request: Request): Call = RealCall(this, request, forWebSocket = false)  
}

class RealCall( // RealCall为真正的请求执行者
  val client: OkHttpClient,
  /** The application's original request unadulterated by redirects or auth headers. */
  val originalRequest: Request,
  val forWebSocket: Boolean
) : Call {
  
  override fun enqueue(responseCallback: Callback) {
    check(executed.compareAndSet(false, true)) { "Already Executed" }

    callStart()
    client.dispatcher.enqueue(AsyncCall(responseCallback))
  }
  
}

class Dispatcher constructor() {
  private val readyAsyncCalls = ArrayDeque<AsyncCall>() // 正在准备中的异步请求队列
  private val runningAsyncCalls = ArrayDeque<AsyncCall>() // 运行中的异步请求
  private val runningSyncCalls = ArrayDeque<RealCall>()  // 同步请求
  
  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) {
        val existingCall = findExistingCallWithHost(call.host)
        if (existingCall != null) call.reuseCallsPerHostFrom(existingCall)
      }
    }
    promoteAndExecute()
  }
  
  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()

        if (runningAsyncCalls.size >= this.maxRequests) break // Max capacity.
        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
  }
}

2.4 CacheInterceptor网络请求缓存处理

OkHttp的缓存原则是，缓存拦截器会根据请求的信息和缓存的响应的信息来判断是否存在缓存可用，如果有可以使用的缓存，那么就返回该缓存给用户，否则就继续使用责任链模式来从服务器中获取响应。 当获取到响应的时候，又会把响应缓存到磁盘上面。涉及到的代码有：

class CacheInterceptor(internal val cache: Cache?) : Interceptor {

  @Throws(IOException::class)
  override fun intercept(chain: Interceptor.Chain): Response {
    val call = chain.call()
    val cacheCandidate = cache?.get(chain.request()) // 根据request的到cache中缓存的response

    val now = System.currentTimeMillis()

    // request判断缓存的策略，是否使用了网络，缓存或两者都使用
    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 {
      networkResponse = chain.proceed(networkRequest) // 调用下一个拦截器，决定从网络上来的到response
    } finally {
      // If we're crashing on I/O or otherwise, don't leak the cache body.
      if (networkResponse == null && cacheCandidate != null) {
        cacheCandidate.body?.closeQuietly()
      }
    }

    // If we have a cache response too, then we're doing a conditional get.
    // 如果本地已经存在cacheResponse，那么让它和网络的到的networkResponse做比较，决定是否来更新缓存的cacheResponse
    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) // 缓存未经缓存过的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
  }
}

3 OKHttp双任务队列机制

• OKHttp实现异步请求采用了双任务队队列机制，通过Dispatcher来调度任务
• 新加入的异步任务AsyncCall进入等待队列readyAsyncCalls
• 遍历readyAsyncCalls判断当前情况：是否超过最大并发数？是否超过同个主机最大请求数？
• 满足条件直接把AsyncCall加入到正在执行的队列RunningAsyncCalls，并且使用线程池执行新加入的异步任务AsyncCall
• AsyncCall执行结束，再次回到Dispatcher的promoteAndExecute()

4 责任链模式与拦截器

• 责任链模式：为了避免请求发送者与多个请求处理者耦合在一起，于是将所有请求的处理者通过前一个对象记住其下一个对象的引用而成一条链；当有请求发生时，可将这请求沿着这条链传递，直到有对象处理它为止；
• OkHttp拦截器就是基于责任链模式来实现的；

5 OKHttp的拦截器

• OkHttp提供了一系列的拦截器来处理相应的业务。也可以通过自定义拦截器，来实现自己的拦截业务；
• OKHttp中完整的拦截链如下所示，在真正发起请求之前（CallServerInterceptor），将经过各个拦截器处理业务；

6 TCP三次握手和四次挥手

• 三次握手：客户端发送请求建立连接，服务端收到请求后立即回应，客户端收到回应后打开连接并通知服务端，服务端再次收到消息也打开连接；
• 四次挥手：客户端数据发送完成告知服务端申请断开，服务端收到断并回应，客户端继续等待最后数据的传送，服务端业务完成再次发送回应消息并断开连接，客户端收到回应，再次发送一次确认，并断开；

7 Socket连接池复用

• 每次建立连接关闭都要三次握手、四次挥手，显然会造成效率低下，Http协议中有一种叫做KeepAlive机制，它可以在传输数据后仍然保持连接状态，当客户端需要再次传输数据时，直接使用空闲下来的连接而不需要重新建立连接；
• OKHttp默认支持5个并发KeepAlive，链路默认的存活时间为5分钟；