当前位置: 首页 > 最新文章 > 正文

OkHttp 用了那么久,你竟然不知道它的原理?

最近,很多朋友反映大厂的面试喜欢挖底层知识,像OkHttp这些都是必问的问题。本文就以请求使用为入口,来深入学习下OkHttp。= null ... }}OkhttpClient可以通过构造者配置参数来构建,也可以直接实例化,直接实例化其实也是内部调用构造者,只是传入的是默认builder。最近,很多朋友反映大厂的面试喜欢挖底层知识,像OkHttp这些都是必问的问题。这里就给大

admin

最近,很多朋友反映大厂的面试喜欢挖底层知识,像OkHttp这些都是必问的问题。本文就以请求使用为入口,来深入学习下OkHttp。= null ... }}OkhttpClient可以通过构造者配置参数来构建,也可以直接实例化,直接实例化其实也是内部调用构造者,只是传入的是默认builder。

最近,很多朋友反映大厂的面试喜欢挖底层知识,像OkHttp这些都是必问的问题。这里就给大家分享一篇非常有帮助的技术文吧。

OkHttp是一个高效的HTTP库:

  • 支持HTTP / 2,允许对同一主机的所有请求共享一个套接字
  • 通过连接池可减少请求延迟(如果HTTP / 2不可用)
  • 支持GZIP压缩减少数据流量
  • 响应缓存可以完全避免网络重复请求
  • 静默恢复处理常见的连接问题

本文就以请求使用为入口,来深入学习下OkHttp。

OkHttp 用了那么久,你竟然不知道它的原理?

一、请求流程分析

1. 同步请求

Okhttp同步GET请求使用:

// 新建一个Okhttp客户端(也可以通过OkHttpClient.Builder来构造)OkHttpClient client = new OkHttpClient();// 构造一个请求对象Request request = new Request.Builder().url(url).build();// 执行同步请求,返回响应Response response = client.newCall(request).execute();// 从响应体中获取数据String str = response.body().string();

先来瞧瞧构建OkhttpClient的源码:

open class OkHttpClient internal constructor(  builder: Builder) : Cloneable, Call.Factory, WebSocket.Factory {    //若直接实例化OkHttpClient,则调用主构造函数以默认Builder作为参数    constructor() : this(Builder())    // 通过builder中的值赋值    @get:JvmName("dispatcher") val dispatcher: Dispatcher = builder.dispatcher    @get:JvmName("connectionPool") val connectionPool: ConnectionPool = builder.connectionPool    ...    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        //cookie        internal var cookieJar: CookieJar = CookieJar.NO_COOKIES        //磁盘缓存        internal var cache: Cache? = null        //dns        internal var dns: Dns = Dns.SYSTEM        //代理设置        internal var proxy: Proxy? = null        ...    }}

OkhttpClient可以通过构造者配置参数来构建,也可以直接实例化,直接实例化其实也是内部调用构造者,只是传入的是默认builder。

再来看看OkhttpClient的newCall方法

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

发现返回的是RealCall,接下来去RealCall中看后续的execute执行方法

override fun execute(): Response {  //确认call没有执行过并置executed为true,否则抛出异常  check(executed.compareAndSet(false, true)) { "Already Executed" }  timeout.enter()  callStart()  try {    //标记执行中    client.dispatcher.executed(this)    //通过拦截器链获取网络响应    return getResponseWithInterceptorChain()  } finally {    //标记执行结束    client.dispatcher.finished(this)  }}

看起来比较精简,通过拦截器链获取网络响应,然后返回响应(拦截器链路在后续拦截器分析)。

2. 异步请求

Okhttp异步GET请求使用:

OkHttpClient client = new OkHttpClient();Request request = new Request.Builder().url(url).build();// 执行异步请求,通过回调返回响应client.newCall(request).enqueue(new Callback() {      @Override      public void onFailure(@NotNull Call call, @NotNull IOException e) {}      @Override      public void onResponse(@NotNull Call call, @NotNull Response response) throws IOException {        // 从回调中通过响应体获取数据        String str = response.body().string();      }});

异步请求流程大致和同步请求相似,但是最后的执行方法是enqueue,并传入回调对象。

我们来看看源码:

override fun enqueue(responseCallback: Callback) {  //确认call没有执行过并置executed为true,否则抛出异常  check(executed.compareAndSet(false, true)) { "Already Executed" }  //监听回调  callStart()  //调用Dispatcher的enqueue方法,并传入一个AsyncCall对象  client.dispatcher.enqueue(AsyncCall(responseCallback))}

内部调用了客户端的分发器的enqueue方法,并把AsyncCall(responseCallback)作为参数传入,AsyncCall是继承自Runnable,且是RealCall的内部类,我们先看Dispatcher.enqueue()方法

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)      ...    }    // 执行    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        //检查请求是否超过一个Host对应的最大请求数        if (asyncCall.callsPerHost.get() >= this.maxRequestsPerHost) continue        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()方法      asyncCall.executeOn(executorService)    }    return isRunning  }}

可以从上面代码看出,就是将符合条件的调用从readyAsyncCalls队列提升到runningAsyncCalls,并调用 AsyncCall的executeOn() 方法,把线程池传入。

我们来看看AsyncCall:

  inner class AsyncCall(    private val responseCallback: Callback  ) : Runnable {    ...    fun executeOn(executorService: ExecutorService) {      client.dispatcher.assertThreadDoesntHoldLock()      var success = false      try {        //使用线程池执行自己的run()方法        executorService.execute(this)        success = true      } catch (e: RejectedExecutionException) {        ...        //失败回调        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 {          //和同步请求一样,通过拦截器链获取网络响应          val response = getResponseWithInterceptorChain()          signalledCallback = true          //回调成功          responseCallback.onResponse(this@RealCall, response)        } catch (e: IOException) {          if (signalledCallback) {            ...          } else {            //回调失败            responseCallback.onFailure(this@RealCall, e)          }        } catch (t: Throwable) {          cancel()          if (!signalledCallback) {            ...            //回调失败            responseCallback.onFailure(this@RealCall, canceledException)          }          throw t        } finally {          //标记结束          client.dispatcher.finished(this)        }      }    }  }

使用线程池来执行自己,接下来就看run()方法,发现和同步请求一样,通过拦截器链获取网络响应,再调用回调对象的回调方法返回响应。

二、拦截器分析

请求大致流程知道了,我们来看看重头戏,拦截器链里面做了什么操作。

@Throws(IOException::class)internal fun getResponseWithInterceptorChain(): Response {  // 建立一个拦截器列表  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(      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的proceed方法:

  @Throws(IOException::class)  override fun proceed(request: Request): Response {    ...    // 复制一个RealInterceptorChain,用于调用链中的下一个拦截器    val next = copy(index = index + 1, request = request)    val interceptor = interceptors[index]    @Suppress("USELESS_ELVIS")    // 调用下一个拦截器的intercept方法,获取response返回给上一个拦截器    val response = interceptor.intercept(next) ?: throw NullPointerException(        "interceptor $interceptor returned null")    ...    return response  }

接下来我们来具体看看各个拦截器的作用

1. RetryAndFollowUpInterceptor

RetryAndFollowUpInterceptor 处理错误恢复和重定向,它会判断错误是否满足条件进行重试,还有根据返回的响应判断是否需要重定向请求。

class RetryAndFollowUpInterceptor(private val client: OkHttpClient) : Interceptor {  @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) {      // 初始化ExchangeFinder(后续ConnectInterceptor会用到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) {          ...          // 满足条件则重试          continue        } catch (e: IOException) {          ...          // 满足条件则重试          continue        }        // 赋上重定向之前的响应(响应体置空)        if (priorResponse != null) {          response = response.newBuilder()              .priorResponse(priorResponse.newBuilder()                  .body(null)                  .build())              .build()        }        val exchange = call.interceptorScopedExchange        //判断是否需重定向,若需则返回重定向请求        val followUp = followUpRequest(response, exchange)        //不需要重定向则直接返回response        if (followUp == null) {          if (exchange != null && exchange.isDuplex) {            call.timeoutEarlyExit()          }          closeActiveExchange = false          return response        }        val followUpBody = followUp.body        // 若该请求只可传输一次,则返回响应        if (followUpBody != null && followUpBody.isOneShot()) {          closeActiveExchange = false          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)      }    }  }  ...}

2. BridgeInterceptor

BridgeInterceptor 桥接应用层和网络层的代码,对用户的请求进行加工(如对请求头进行设置添加),也对网络响应做相应的处理(如解压服务端返回的 gzip 压缩数据)。

class BridgeInterceptor(private val cookieJar: CookieJar) : Interceptor {  @Throws(IOException::class)  override fun intercept(chain: Interceptor.Chain): Response {    // 获取用户请求    val userRequest = chain.request()    // 真正发送的网络请求的构建者    val requestBuilder = userRequest.newBuilder()    // 用户请求的请求体    val body = userRequest.body    // 对请求头的设置    ...    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())    cookieJar.receiveHeaders(userRequest.url, networkResponse.headers)    val responseBuilder = networkResponse.newBuilder()        .request(userRequest)    // 若因配置问题,服务端返回gzip压缩的数据,则做相应的解压缩    if (transparentGzip &&        "gzip".equals(networkResponse.header("Content-Encoding"), ignoreCase = true) &&        networkResponse.promisesBody()) {      val responseBody = networkResponse.body      if (responseBody != null) {        // GzipSource对象,用于解压        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()  }}

3. CacheInterceptor

CacheInterceptor 承担着缓存的查找与保存的职责。根据策略判断是使用缓存还是走网络请求,对于返回的响应,满足条件则进行缓存。

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())    val now = System.currentTimeMillis()    // 检查缓存策略    val strategy = CacheStrategy.Factory(now, chain.request(), cacheCandidate).compute()    // 若还需发送网络请求,则networkRequest不为空    val networkRequest = strategy.networkRequest    // 若存在可用缓存,则cacheResponse不为空    val cacheResponse = strategy.cacheResponse    ...    // 如果我们被禁止使用网络,并且无可用缓存,则返回失败    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 (networkRequest == null) {      return cacheResponse!!.newBuilder()          .cacheResponse(stripBody(cacheResponse))          .build().also {            listener.cacheHit(call, it)          }    }    ...    var networkResponse: Response? = null    try {      // 若无缓存可用,则执行下一个拦截器,获取响应      networkResponse = chain.proceed(networkRequest)    } finally {      if (networkResponse == null && cacheCandidate != null) {        cacheCandidate.body?.closeQuietly()      }    }    // 如果我们还有缓存响应,且网络响应code为304,则更新缓存响应,并返回    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()        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)) {        // 将网络响应写入缓存,并返回        val cacheRequest = cache.put(response)        return cacheWritingResponse(cacheRequest, response).also {          if (cacheResponse != null) {            listener.cacheMiss(call)          }        }      }      // 根据请求方法判断是否为无效请求,是则从缓存移除相对应响应      if (HttpMethod.invalidatesCache(networkRequest.method)) {        try {          cache.remove(networkRequest)        } catch (_: IOException) {          // cache无法被写        }      }    }    return response  }}

4. ConnectInterceptor

ConnectInterceptor 主要是给网络请求提供一个连接,并交给下一个拦截器处理,这里还没有发送请求到服务器获取响应。在获取连接对象的时候,使用了连接池ConnectionPool来复用连接。

object ConnectInterceptor : Interceptor {  @Throws(IOException::class)  override fun intercept(chain: Interceptor.Chain): Response {    val realChain = chain as RealInterceptorChain    // 查找新连接或池里的连接以承载即将到来的请求和响应    val exchange = realChain.call.initExchange(chain)    val connectedChain = realChain.copy(exchange = exchange)    return connectedChain.proceed(realChain.request)  }}

ConnectInterceptor 看似代码很少,其实代码都在深处,看下initExchange方法

internal fun initExchange(chain: RealInterceptorChain): Exchange {  ...  // codec(ExchangeCodec) 是一个连接所用的编码解码器,用于编码HTTP请求和解码HTTP响应  val codec = exchangeFinder.find(client, chain)  // result(Exchange)是封装这个编码解码器的一个工具类,用于管理ExchangeCodec,处理实际的 I/O  val result = Exchange(this, eventListener, exchangeFinder, codec)  ...  return result}

ExchangeCodec持有连接,可通过其编码请求到服务端和获取服务端的响应并解码,我们依方法进入到最深处,看看是连接是如何获取的(代码已做简化处理)。

private fun findConnection(): RealConnection {  // 1、复用当前连接  val callConnection = call.connection   if (callConnection != null) {      //检查这个连接是否可用和可复用      if (callConnection.noNewExchanges || !sameHostAndPort(callConnection.route().address.url)) {        toClose = call.releaseConnectionNoEvents()      }    return callConnection  } //2、从连接池中获取可用连接  if (connectionPool.callAcquirePooledConnection(address, call, null, false)) {    val result = call.connection!!    eventListener.connectionAcquired(call, result)    return result  }  //3、从连接池中获取可用连接,通过一组路由routes(涉及知识点Http2多路复用)  if (connectionPool.callAcquirePooledConnection(address, call, routes, false)) {      val result = call.connection!!      return result    }  route = localRouteSelection.next()  // 4、创建新连接,进行tcp连接  val newConnection = RealConnection(connectionPool, route)  newConnection.connect  // 5、再获取一次连接,在新建连接过程中可能有其他竞争连接被创建了,如可用防止浪费  if (connectionPool.callAcquirePooledConnection(address, call, routes, true)) {    val result = call.connection!!     // 关闭刚刚创建的新连接    newConnection.socket().closeQuietly()    return result  }  //6、还是要使用创建的新连接,放入连接池,并返回  connectionPool.put(newConnection)  return newConnection}

5. CallServerInterceptor

CallServerInterceptor 是真正向服务器发起请求并获取响应的,它是拦责任链的最后一个拦截器,拿到响应后返回给上一个拦截器。代码已做简化(省略了很多条件判断和处理)。

class CallServerInterceptor(private val forWebSocket: Boolean) : Interceptor {  @Throws(IOException::class)  override fun intercept(chain: Interceptor.Chain): Response {    val realChain = chain as RealInterceptorChain    // ConnectInterceptor获取到的,持有编码解码器    val exchange = realChain.exchange!!    val request = realChain.request    val requestBody = request.body    val sentRequestMillis = System.currentTimeMillis()    var responseBuilder: Response.Builder? = null    try {      // 写入请求头      exchange.writeRequestHeaders(request)      if (HttpMethod.permitsRequestBody(request.method) && requestBody != null) {        if (...) {            // 写入请求体            val bufferedRequestBody = exchange.createRequestBody(request, false).buffer()            requestBody.writeTo(bufferedRequestBody)            bufferedRequestBody.close()        } else {            ...        }      } else {        // 无请求体        exchange.noRequestBody()      }    } catch (e: IOException) {...}    try {      if (responseBuilder == null) {        // 读取响应头        responseBuilder = exchange.readResponseHeaders(expectContinue = false)!!        if (invokeStartEvent) {          exchange.responseHeadersStart()          invokeStartEvent = false        }      }      // 构建响应      var response = responseBuilder          .request(request)          .handshake(exchange.connection.handshake())          .sentRequestAtMillis(sentRequestMillis)          .receivedResponseAtMillis(System.currentTimeMillis())          .build()      var code = response.code      // 读取响应体      response = if (forWebSocket && code == 101) {        response.newBuilder()            .body(EMPTY_RESPONSE)            .build()      } else {        response.newBuilder()            .body(exchange.openResponseBody(response))            .build()      }      ...      return response    } catch (e: IOException) {      ...    }  }}

总结

以上就是对OkHttp的源码解析,可以看出它是一个结构清晰的优质源码库,各个模块通过设计模式解耦。

总结下流程:首先通过OkHttpClient对象调用newCall方法得到RealCall实例,再通过调用RealCall的execute方法或enqueue方法,这两个方法最终都会调用到getResponseWithInterceptorChain方法,运用责任链模式,开始一层层传入各个拦截器,每个拦截器都有着自己的职责,最终在CallServerInterceptor发出请求并获取响应,然后层层返回响应。


上一篇: 慕课网课程分享(慕课网项目) 下一篇:Eclipse搭建SSH(Struts2+Spring+Hibernate)框架教程
返回顶部