OkHttp(三)责任链

OkHttp(三)

前两篇文章,讲述了OkHttp的基础的使用与请求的调度状况,而今天就让咱们来看看OkHttp的精髓之一-责任链模式。web

责任链模式

前面的文章中咱们看到,当实际进行网络请求时,不管是同步请求仍是异步请求都会使用getResponseWithInterceptorChain() 这个方法,因此咱们先从这个方法开始研究。算法

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)
    
    // 建立责任链
    val chain = RealInterceptorChain(interceptors, transmitter, null, 0, originalRequest, this,
        client.connectTimeoutMillis, client.readTimeoutMillis, client.writeTimeoutMillis)

    var calledNoMoreExchanges = false
    try {
      // 执行责任链
      val response = chain.proceed(originalRequest)
      if (transmitter.isCanceled) {
        response.closeQuietly()
        throw IOException("Canceled")
      }
      return response
    } catch (e: IOException) {
      calledNoMoreExchanges = true
      throw transmitter.noMoreExchanges(e) as Throwable
    } finally {
      if (!calledNoMoreExchanges) {
        transmitter.noMoreExchanges(null)
      }
    }
  }

咱们能够看到,该方法中将拦截器逐一添加集合中,并建立了一个责任链,用chain.proceed()方法来执行请求。promise

OkHttp采用责任链的模式来使每一个功能分开,每一个Interceptor自行完成本身的任务,而且将不属于本身的任务交给下一个,简化了各自的责任和逻辑。浏览器

接下来看看proceed的方法缓存

override fun proceed(request: Request): Response {
    return proceed(request, transmitter, exchange)
  }

  @Throws(IOException::class)
  fun proceed(request: Request, transmitter: Transmitter, exchange: Exchange?): Response {
    if (index >= interceptors.size) throw AssertionError()

    calls++

    ...
    
    // 获取下一个拦截器,链中的拦截器集合index+1
    // Call the next interceptor in the chain.
    val next = RealInterceptorChain(interceptors, transmitter, exchange,
        index + 1, request, call, connectTimeout, readTimeout, writeTimeout)
    val interceptor = interceptors[index]

    @Suppress("USELESS_ELVIS")
    // 执行当前的拦截器-若是在配置okhttpClient,时没有设置intercept默认是先执行:retryAndFollowUpInterceptor 拦截器`
    val response = interceptor.intercept(next) ?: throw NullPointerException(
        "interceptor $interceptor returned null")

    ...

    return response
  }

在该方法中咱们能够看到递归调用了下一个拦截器,当全部拦截器调用完毕后,返回咱们所得的Response。每一个拦截器都重写了intercept()方法,用以执行请求。服务器

责任链的一个执行过程以下图cookie

20190710140954641.png

接下来让咱们分析默认责任链的一个做用,并做出一些源码分析。网络

RetryAndFollowUpInterceptor

其建立过程是在 构建newCall对象时app

private RealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
    this.client = client;
    this.originalRequest = originalRequest;
    this.forWebSocket = forWebSocket;
    this.retryAndFollowUpInterceptor = new RetryAndFollowUpInterceptor(client, forWebSocket);
  }
  ...
  @Override public Call newCall(Request request) {
    return RealCall.newRealCall(this, request, false /* for web socket */);
  }

简单看一下使用的过程
首先建立了transmitter对象,他封装了网络请求相关的信息:链接池,地址信息,网络请求,事件回调,负责网络链接的链接、关闭,释放等操做。框架

var request = chain.request()
    val realChain = chain as RealInterceptorChain
    val transmitter = realChain.transmitter()

然后则进入了网络链接的循环

//计数器 屡次相应的次数是由限制的,不一样浏览器推荐的次数不一样,还特别强调了HTTP 1.0协议推荐5次。
    var followUpCount = 0
    var priorResponse: Response? = null
    while (true) {
      //准备链接
      transmitter.prepareToConnect(request)

      if (transmitter.isCanceled) {
        throw IOException("Canceled")
      }

      var response: Response
      var success = false
      try {
        // 获得最终的结果
        response = realChain.proceed(request, transmitter, null)
        success = true
      } catch (e: RouteException) {
        // The attempt to connect via a route failed. The request will not have been sent.
        //链接地址的异常,判断是否能可以恢复,也就是是否要重试
        if (!recover(e.lastConnectException, transmitter, false, request)) {
          throw e.firstConnectException
        }
        continue
      } catch (e: IOException) {
        // An attempt to communicate with a server failed. The request may have been sent.
        // 链接服务器的异常 判断网络请求是否已经开始
        val requestSendStarted = e !is ConnectionShutdownException
        // 同上
        if (!recover(e, transmitter, requestSendStarted, request)) throw e
        continue
      } finally {
        // The network call threw an exception. Release any resources.
        // 释放资源
        if (!success) {
          transmitter.exchangeDoneDueToException()
        }
      }

      // Attach the prior response if it exists. Such responses never have a body.
      //若是不为空保存到Response中
      if (priorResponse != null) {
        response = response.newBuilder()
            .priorResponse(priorResponse.newBuilder()
                .body(null)
                .build())
            .build()
      }

      val exchange = response.exchange
      val route = exchange?.connection()?.route()
      // 判断返回结果response,是否须要继续完善请求,例如证书验证等等
      val followUp = followUpRequest(response, route)
      
      // 若是不须要继续完善网络请求,返回response
      if (followUp == null) {
        if (exchange != null && exchange.isDuplex) {
          transmitter.timeoutEarlyExit()
        }
        return response
      }
      
      // 若是body内容只能发送一次 直接放回
      val followUpBody = followUp.body
      if (followUpBody != null && followUpBody.isOneShot()) {
        return response
      }

      response.body?.closeQuietly()
      if (transmitter.hasExchange()) {
        exchange?.detachWithViolence()
      }
      
      // 若是已经超过最大的网络请求追加数,释放链接,抛出协议异常
      if (++followUpCount > MAX_FOLLOW_UPS) {
        throw ProtocolException("Too many follow-up requests: $followUpCount")
      }
      // 更新下一次的网络请求对象
      request = followUp
      // 保存上一次的请求结果
      priorResponse = response
    }

而后就是重试阶段recover()的源码了

/**
   * Report and attempt to recover from a failure to communicate with a server. Returns true if
   * `e` is recoverable, or false if the failure is permanent. Requests with a body can only
   * be recovered if the body is buffered or if the failure occurred before the request has been
   * sent.
   */
  private fun recover(
    e: IOException,
    transmitter: Transmitter,
    requestSendStarted: Boolean,
    userRequest: Request
  ): Boolean {
    // The application layer has forbidden retries.
    // 设置了不须要重试
    if (!client.retryOnConnectionFailure) return false

    // We can't send the request body again.
    // body内容只能发送一次
    if (requestSendStarted && requestIsOneShot(e, userRequest)) return false

    // This exception is fatal.
    // 判断异常类型,是否要继续尝试,
    // 不会重试的类型:协议异常、Socketet异常而且网络状况还没开始,ssl认证异常
    if (!isRecoverable(e, requestSendStarted)) return false

    // No more routes to attempt.
    // 已经没有其余可用的路由地址了
    if (!transmitter.canRetry()) return false
    // For failure recovery, use the same route selector with a new connection.
    // 其余放回true
    return true
  }

咱们稍微屡一下上面源码的流程:

  • 首先使用了transmitter对象(重要),用以提供相应的网络链接相关的东西
  • 而后开始链接,而后又有着几种状况

    • 链接成功,且无后续操做(如认证等),直接放回
    • 链接成功,且有后续操做,则进入下一次循环
    • 链接失败,RouteException和IOException异常,利用recover()判断是否重试,不要重试则释放资源,要重试则continue;
    • 链接成功,可是重试的次数超过限度,则有问题(能够本身建立拦截器来修改重试次数)。

BridgeIntecepter

这个拦截器的功能较为的简单,请求以前对响应头作了一些检查,并添加一些头,而后在请求以后对响应作一些处理(gzip解压or设置cookie)。

仍是让咱们看一下源码。

override fun intercept(chain: Interceptor.Chain): Response {
    val userRequest = chain.request()
    val requestBuilder = userRequest.newBuilder()
    
    // 若是咱们有RequestBody,会写一些header信息,如内容长度和内容类型等
    val body = userRequest.body
    if (body != null) {
      ...
    }
    
    // 对一些必要的属性进行补充
    if (userRequest.header("Host") == null) {
      requestBuilder.header("Host", userRequest.url.toHostHeader())
    }

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

    // If we add an "Accept-Encoding: gzip" header field we're responsible for also decompressing
    // the transfer stream.
    // 默认的编码格式gzip
    var transparentGzip = false
    if (userRequest.header("Accept-Encoding") == null && userRequest.header("Range") == null) {
      transparentGzip = true
      requestBuilder.header("Accept-Encoding", "gzip")
    }
    
    // 把以前的cookie存在header里
    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)
    }
    
    // 获得Response
    val networkResponse = chain.proceed(requestBuilder.build())
    
    // 保存新的cookie
    cookieJar.receiveHeaders(userRequest.url, networkResponse.headers)

    val responseBuilder = networkResponse.newBuilder()
        .request(userRequest)
    
    // 若是使用的gzip编码,而且返回的response有body信息,对作相应的处理
    if (transparentGzip &&
        "gzip".equals(networkResponse.header("Content-Encoding"), ignoreCase = true) &&
        networkResponse.promisesBody()) {
      val responseBody = networkResponse.body
      if (responseBody != null) {
        val gzipSource = GzipSource(responseBody.source())
        val strippedHeaders = networkResponse.headers.newBuilder()
            .removeAll("Content-Encoding")
            .removeAll("Content-Length")
            .build()
        responseBuilder.headers(strippedHeaders)
        val contentType = networkResponse.header("Content-Type")
        responseBuilder.body(RealResponseBody(contentType, -1L, gzipSource.buffer()))
      }
    }

    return responseBuilder.build()
  }

CacheIntecepter

在看这个拦截器的源码以前,咱们还得关注一件事情,OkHttp的缓存是怎样缓存的呢?

OkHttp中的Cache类,采用了DiskLruCache,内部使用最近最少使用算法,优先淘汰最近时间内最少次使用的缓存对象,它只有硬存缓存,并无内存缓存,这是他缓存机制的一大缺陷,固然咱们能够经过自定义缓存机制来解决这一问题。

在OkHttp中还存在一个缓存策略CacheStrategy
CacheStrategy的内部工厂类Factory中有一个getCandidate方法,会根据实际的请求生成对应的CacheStrategy类返回,是个典型的简单工厂模式。其内部维护一个request和response,经过指定request和response来告诉CacheInterceptor是使用缓存仍是使用网络请求,亦或二者同时使用。

了解完以后,咱们来看源码:

override fun intercept(chain: Interceptor.Chain): Response {
    1.若是设置缓存而且当前request有缓存,则从缓存Cache中获取当前请求request的缓存response
    val cacheCandidate = cache?.get(chain.request())

    val now = System.currentTimeMillis()
    
    // 2.传入的请求request和获取的缓存response经过缓存策略对象CacheStragy的工厂类get方法根据一些规则获取缓存策略CacheStrategy
    //(这里的规则根据请求的request和缓存的Response的header头部信息生成的,好比是否有noCache标志位,是不是immutable不可变,缓存是否过时等等)
    val strategy = CacheStrategy.Factory(now, chain.request(), cacheCandidate).compute()
    // 3.生成的CacheStrategy有2个变量,networkRequest和cacheRequest,若是networkRequest为Null表示不进行网络请求,若是cacheResponse为null,则表示没有有效缓存 
    val networkRequest = strategy.networkRequest
    val cacheResponse = strategy.cacheResponse

    cache?.trackResponse(strategy)
    
    // 4.缓存不可用,关闭
    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.
    // 5.若是networkRequest和cacheResponse都为Null,则表示不请求网络且缓存为null,返回504,请求失败
    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()
    }

    // If we don't need the network, we're done.
    // 6.若是不请求网络,但存在缓存,则不请求网络,直接返回缓存,结束,不执行下一个拦截器
    if (networkRequest == null) {
      return cacheResponse!!.newBuilder()
          .cacheResponse(stripBody(cacheResponse))
          .build()
    }
    
    // 7.不然,请求网络,并调用下一个拦截器链,将请求转发到下一个拦截器
    var networkResponse: Response? = null
    try {
      networkResponse = chain.proceed(networkRequest)
    } finally {
      // If we're crashing on I/O or otherwise, don't leak the cache body.
      if (networkResponse == null && cacheCandidate != null) {
        cacheCandidate.body?.closeQuietly()
      }
    }
    
    //8.请求网络,而且网络请求返回HTTP_NOT_MODIFIED,说明缓存有效,则合并网络响应和缓存结果,同时更新缓存
    // If we have a cache response too, then we're doing a conditional get.
    if (cacheResponse != null) {
      if (networkResponse?.code == HTTP_NOT_MODIFIED) {
        val response = cacheResponse.newBuilder()
            .headers(combine(cacheResponse.headers, networkResponse.headers))
            .sentRequestAtMillis(networkResponse.sentRequestAtMillis)
            .receivedResponseAtMillis(networkResponse.receivedResponseAtMillis)
            // 清空以前的缓冲
            .cacheResponse(stripBody(cacheResponse))
            // 清空请求到的内容, 由于内容没有改变
            .networkResponse(stripBody(networkResponse))
            .build()

        networkResponse.body!!.close()

        // Update the cache after combining headers but before stripping the
        // Content-Encoding header (as performed by initContentStream()).
        cache!!.trackConditionalCacheHit()
        cache.update(cacheResponse, response)
        return response
      } else {
        cacheResponse.body?.closeQuietly()
      }
    }
    
    //9.若没有缓存,则写入缓存
    val response = networkResponse!!.newBuilder()
        .cacheResponse(stripBody(cacheResponse))
        .networkResponse(stripBody(networkResponse))
        .build()

    if (cache != null) {
      if (response.promisesBody() && CacheStrategy.isCacheable(response, networkRequest)) {
        // Offer this request to the cache.
        val cacheRequest = cache.put(response)
        return cacheWritingResponse(cacheRequest, response)
      }
      // 若是请求的方法不须要缓存,移除缓存,例如post,put
      if (HttpMethod.invalidatesCache(networkRequest.method)) {
        try {
          cache.remove(networkRequest)
        } catch (_: IOException) {
          // The cache cannot be written.
        }
      }
    }

    return response
  }

让咱们简单梳理一下缓存流程

  • 从当前的Request中获取缓存,看是否有缓存
  • 非网络请求时,需结合是否有缓存进行判断,若是有缓存,直接返回;若是没有缓存,放回504
  • 是网络请求时,若是放回304,则作一个小的修补便可;不然根据缓存策略来判断是否要更新缓存(通常要)。

ConnectIntecepter(核心)

获取链接这个过程较为复杂,尽力来梳理这个过程。

首先咱们直接来看这个类的源码,不难发现这个类的源码较为简单,主要核心是transmitter的方法。

override fun intercept(chain: Interceptor.Chain): Response {
    val realChain = chain as RealInterceptorChain
    val request = realChain.request()
    val transmitter = realChain.transmitter()

    // We need the network to satisfy this request. Possibly for validating a conditional GET.
    val doExtensiveHealthChecks = request.method != "GET"
    
    // 利用重试的责任链生成的transmitter类 来获取链接
    val exchange = transmitter.newExchange(chain, doExtensiveHealthChecks)

    return realChain.proceed(request, transmitter, exchange)
  }

而后咱们来看看这个类,transmitter

/** Returns a new exchange to carry a new request and response. */
  internal fun newExchange(chain: Interceptor.Chain, doExtensiveHealthChecks: Boolean): Exchange {
    ...//作一些检查
       
    // 获取链接 分配一个Connection和HttpCodec,为最终的请求作准备
    val codec = exchangeFinder!!.find(client, chain, doExtensiveHealthChecks)
    val result = Exchange(this, call, eventListener, exchangeFinder!!, codec)
    ...
  }
fun find(
    client: OkHttpClient,
    chain: Interceptor.Chain,
    doExtensiveHealthChecks: Boolean
  ): ExchangeCodec {
    val connectTimeout = chain.connectTimeoutMillis()
    val readTimeout = chain.readTimeoutMillis()
    val writeTimeout = chain.writeTimeoutMillis()
    val pingIntervalMillis = client.pingIntervalMillis
    val connectionRetryEnabled = client.retryOnConnectionFailure

    try {
      // 获取链接
      val resultConnection = findHealthyConnection(
          connectTimeout = connectTimeout,
          readTimeout = readTimeout,
          writeTimeout = writeTimeout,
          pingIntervalMillis = pingIntervalMillis,
          connectionRetryEnabled = connectionRetryEnabled,
          doExtensiveHealthChecks = doExtensiveHealthChecks
      )
      // 设置编码,有Http1codec和Http2codec两种方式 后者能够复用链接
      return resultConnection.newCodec(client, chain)
    } catch (e: RouteException) {
      trackFailure()
      throw e
    } catch (e: IOException) {
      trackFailure()
      throw RouteException(e)
    }
// 获取链接
  @Throws(IOException::class)
  private fun findHealthyConnection(
    connectTimeout: Int,
    readTimeout: Int,
    writeTimeout: Int,
    pingIntervalMillis: Int,
    connectionRetryEnabled: Boolean,
    doExtensiveHealthChecks: Boolean
  ): RealConnection {
    while (true) {
      // 查找新链接
      val candidate = findConnection(
          connectTimeout = connectTimeout,
          readTimeout = readTimeout,
          writeTimeout = writeTimeout,
          pingIntervalMillis = pingIntervalMillis,
          connectionRetryEnabled = connectionRetryEnabled
      )

      // If this is a brand new connection, we can skip the extensive health checks.
      // 若是是新链接 则直接使用
      synchronized(connectionPool) {
        if (candidate.successCount == 0) {
          return candidate
        }
      }

      // Do a (potentially slow) check to confirm that the pooled connection is still good. If it
      // isn't, take it out of the pool and start again.
      //判断链接池中链接是否可用,若是不可用,则释放该链接并从链接池中移除,并继续寻找可用链接
      if (!candidate.isHealthy(doExtensiveHealthChecks)) {
        candidate.noNewExchanges()
        continue
      }

      return candidate
    }
  }

接着就是正式获取链接这一步了,咱们从注释中能够看到,首先从已经存在的Connection来选取链接,然后从链接池中寻找,最后才是新建链接。

/**
   * Returns a connection to host a new stream. This prefers the existing connection if it exists,
   * then the pool, finally building a new connection.
   */
  @Throws(IOException::class)
  private fun findConnection(
    connectTimeout: Int,
    readTimeout: Int,
    writeTimeout: Int,
    pingIntervalMillis: Int,
    connectionRetryEnabled: Boolean
  ): RealConnection {
    var foundPooledConnection = false
    var result: RealConnection? = null
    var selectedRoute: Route? = null
    var releasedConnection: RealConnection?
    val toClose: Socket?
    synchronized(connectionPool) {
      if (transmitter.isCanceled) throw IOException("Canceled")
      hasStreamFailure = false // This is a fresh attempt.
·     
      // 对现有链接作一个备份
      releasedConnection = transmitter.connection
      toClose = if (transmitter.connection != null && transmitter.connection!!.noNewExchanges) {
        // 获得要关闭的链接的socket
        transmitter.releaseConnectionNoEvents()
      } else {
        null
      }
      
      // 若是可使用 则使用
      if (transmitter.connection != null) {
        // We had an already-allocated connection and it's good.
        result = transmitter.connection
        releasedConnection = null
      }
      
      // 若是没有能够用的链接,从链接池中查找
      if (result == null) {
        // Attempt to get a connection from the pool.
        // 以URL为key查找
        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() // 使用路由地址,能够是代理地址
        }
      }
    }
    // 关闭以前的socket
    toClose?.closeQuietly()

    ... // 若是上面找到,直接返回
    if (result != null) {
      // If we found an already-allocated or pooled connection, we're done.
      return result!!
    }
    
    // If we need a route selection, make one. This is a blocking operation.
    var newRouteSelection = false
    // 选择一个不空的路由
    if (selectedRoute == null && (routeSelection == null || !routeSelection!!.hasNext())) {
      newRouteSelection = true
      routeSelection = routeSelector.next()
    }

    var routes: List<Route>? = null
    synchronized(connectionPool) {
      if (transmitter.isCanceled) throw IOException("Canceled")

      if (newRouteSelection) {
        // Now that we have a set of IP addresses, make another attempt at getting a connection from
        // the pool. This could match due to connection coalescing.
        routes = routeSelection!!.routes
       
        // 根据IP地址和Route从链接池进行第二次查找
        if (connectionPool.transmitterAcquirePooledConnection(
                address, transmitter, routes, false)) {
          foundPooledConnection = true
          result = transmitter.connection
        }
      }

      if (!foundPooledConnection) {
        if (selectedRoute == null) {
          selectedRoute = routeSelection!!.next()
        }
        // 若是没有找到,再使用下一个路由集合
        // Create a connection and assign it to this allocation immediately. This makes it possible
        // for an asynchronous cancel() to interrupt the handshake we're about to do.
        result = RealConnection(connectionPool, selectedRoute!!)
        connectingConnection = result
      }
    }

    // If we found a pooled connection on the 2nd time around, we're done.
    if (foundPooledConnection) {
      eventListener.connectionAcquired(call, result!!)
      return result!!
    }
    
    // 到这里还没找到链接,那就去建立这个链接
    // Do TCP + TLS handshakes. This is a blocking operation.
    result!!.connect(
        connectTimeout,
        readTimeout,
        writeTimeout,
        pingIntervalMillis,
        connectionRetryEnabled,
        call,
        eventListener
    )
    connectionPool.routeDatabase.connected(result!!.route())

    var socket: Socket? = null
    synchronized(connectionPool) {
      connectingConnection = null
      // Last attempt at connection coalescing, which only occurs if we attempted multiple
      // concurrent connections to the same host.
      // 若是result链接是http2.0链接,http2.0支持一个链接同时发起多个请求,这里作去重判断,防止建立多个
      if (connectionPool.transmitterAcquirePooledConnection(address, transmitter, routes, true)) {
        // We lost the race! Close the connection we created and return the pooled connection.
        result!!.noNewExchanges = true
        socket = result!!.socket()
        result = transmitter.connection

        // It's possible for us to obtain a coalesced connection that is immediately unhealthy. In
        // that case we will retry the route we just successfully connected with.
        nextRouteToTry = selectedRoute
      } else {
        connectionPool.put(result!!)
        transmitter.acquireConnectionNoEvents(result!!)
      }
    }
    socket?.closeQuietly()

    eventListener.connectionAcquired(call, result!!)
    return result!!

在这个源码中,出现了几个新的类,路由route类,地址address类,咱们简单的来看看这两个类,
Address:封装了全部的能够访问的地址信息,在这个类中还添加了代理和dns的相关信息(在OkHttpClient中设置好)proxySelector能够为一个URI设置多个代理,若是地址链接失败还回调connectFailed;proxy设置单独的全局代理,他的优先级高于proxySelecttor;dns用法和proxySelecttor相似,能够返回多个地址。

private Address createAddress(HttpUrl url) {
        SSLSocketFactory sslSocketFactory = null;
        HostnameVerifier hostnameVerifier = null;
        CertificatePinner certificatePinner = null;
        if (url.isHttps()) {
            sslSocketFactory = client.sslSocketFactory();
            hostnameVerifier = client.hostnameVerifier();
            certificatePinner = client.certificatePinner();
        }
 
        return new Address(url.host(), url.port(), client.dns(), client.socketFactory(),
                sslSocketFactory, hostnameVerifier, certificatePinner, client.proxyAuthenticator(),
                client.proxy(), client.protocols(), client.connectionSpecs(), client.proxySelector());
    }

Route路由:对地址Adress的一个封装类
RouteSelector路由选择器:在OKhttp中其实其做用也就是返回一个可用的Route对象

咱们来大概梳理一下流程

  • 首先对当前的流进行一个初步判断,知足则复用
  • 不知足则,对链接池进行第一次的查找,这次查找中,route类为空
connectionPool.transmitterAcquirePooledConnection(address, transmitter, null, false)

查找获得直接复用

  • 查找不到则使用路由进行查找,查找设置的代理和DNS是否能找到相关的代理,若是找到则复用
connectionPool.transmitterAcquirePooledConnection(address, transmitter, routes, false)
  • 上述路线都查找不到,直接新建一个链接,放入链接池中,并把解析的host等信息保存到Connection中,方便下次复用。其中还要多作一步判断,若是是HTTP2同时发起的请求,要进行一个去重的操做。

下图是一个简要的链接步骤。

在这里插入图片描述

CallServerInterceptor

  • 首先得到前面Intecepter获取的信息
  • 而后利用编码器写入header信息
exchange.writeRequestHeaders(request)
  • 判断是否要发送请求体,有请求体时,但指望返回状态码是100时,则不发送。不然利用流封装后发送。
var responseBuilder: Response.Builder? = null
    if (HttpMethod.permitsRequestBody(request.method) && requestBody != null) {
      // If there's a "Expect: 100-continue" header on the request, wait for a "HTTP/1.1 100
      // Continue" response before transmitting the request body. If we don't get that, return
      // what we did get (such as a 4xx response) without ever transmitting the request body.
      if ("100-continue".equals(request.header("Expect"), ignoreCase = true)) {
        exchange.flushRequest()
        responseHeadersStarted = true
        exchange.responseHeadersStart()
        responseBuilder = exchange.readResponseHeaders(true)
      }
      if (responseBuilder == null) {
        if (requestBody.isDuplex()) {
          // Prepare a duplex body so that the application can send a request body later.
          exchange.flushRequest()
          val bufferedRequestBody = exchange.createRequestBody(request, true).buffer()
          requestBody.writeTo(bufferedRequestBody)
        } else {
          // Write the request body if the "Expect: 100-continue" expectation was met.
          val bufferedRequestBody = exchange.createRequestBody(request, false).buffer()
          requestBody.writeTo(bufferedRequestBody)
          bufferedRequestBody.close()
        }
      } else {
        exchange.noRequestBody()
        if (!exchange.connection()!!.isMultiplexed) {
          // If the "Expect: 100-continue" expectation wasn't met, prevent the HTTP/1 connection
          // from being reused. Otherwise we're still obligated to transmit the request body to
          // leave the connection in a consistent state.
          exchange.noNewExchangesOnConnection()
        }
      }
    } else {
      exchange.noRequestBody()
    }
// 建立response,把握手信息,和request等信息保存进去
@Override public Response intercept(Chain chain) throws IOException {
    ...
    // 写入request结束
    httpCodec.finishRequest();
    if (responseBuilder == null) {
      realChain.eventListener().responseHeadersStart(realChain.call());
      // 读取相应response的header信息
      responseBuilder = httpCodec.readResponseHeaders(false);
    }
 
    // 建立response,把握手信息,和request等信息保存进去
    Response response = responseBuilder
        .request(request)
        .handshake(streamAllocation.connection().handshake())
        .sentRequestAtMillis(sentRequestMillis)
        .receivedResponseAtMillis(System.currentTimeMillis())
        .build();
    // 开始判断请求码
    int code = response.code();
    if (code == 100) {
      // 若是是100,直接读取header
      responseBuilder = httpCodec.readResponseHeaders(false);
      response = responseBuilder
              .request(request)
              // 握手
              .handshake(streamAllocation.connection().handshake())
              .sentRequestAtMillis(sentRequestMillis)
              .receivedResponseAtMillis(System.currentTimeMillis())
              .build();
 
      code = response.code();
    }
    ...
    // 判断请求码
    if (forWebSocket && code == 101) {
      // 客户端须要转换协议,这里须要设置一个空的response
      response = response.newBuilder()
          .body(Util.EMPTY_RESPONSE)
          .build();
    } else {
      // 读取网络的body
      response = response.newBuilder()
          .body(httpCodec.openResponseBody(response))
          .build();
    }
    // 若是header请求关闭链接
    if ("close".equalsIgnoreCase(response.request().header("Connection"))
        || "close".equalsIgnoreCase(response.header("Connection"))) {
      // 关闭这个连接
      streamAllocation.noNewStreams();
    }
    // 特殊code判断
    if ((code == 204 || code == 205) && response.body().contentLength() > 0) {
      throw new ProtocolException(
          "HTTP " + code + " had non-zero Content-Length: " + response.body().contentLength());
    }
 
    return response;
  }

若是想要了解具体的读取和写入流程,以我如今使用的Http 2.0为例:
链接:Http2Connection;
流:Http2Stream;
编解码器:Http2Codec;
读操做:Http2Reader;
写操做:Http2Writer;
他们之间的关系:
一、Http2Connection调用Http2Reader和Http2Writer来进行读写;
二、Http2Stream调用Http2Connection进行读写;
三、Http2Codec调用Http2Connection和Http2Stream进行操做;

总结

咱们分三个阶段来简要介绍了OkHttp这个框架,因为如今水平有限,因此会存在疏漏。之后有些有新的发现,则再对其进行补充。

相关文章
相关标签/搜索