OkHttp源码解析07 - getResponseWithInterceptorChain

待我称王封你为后i 2023-10-10 10:59 11阅读 0赞

@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)
    
      //创建Interceptor的链条
      //用来开始工作，管理工作
      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)
        }
      }
    }

来看，OkHttp的核心代码  
每一个Interceptor都有前置、中置、后置三个动作，(最后一个只有前置和后置动作)。  
![在这里插入图片描述][20200414081326707.png]  
可以看到，这里拼接了Interceptor集合，并传入RealInterceptorChain链条中。  
最后调用`val response = chain.proceed(originalRequest)`，开始执行链条。

来看RealInterceptorChain\#proceed

@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"
          }
        }
    
        // 复制一份RealInterceptorChain，index+1，用作给下一个拦截器
        val next = copy(index = index + 1, request = request)
        //index初始传入为0，即取出第一个interceptor
        val interceptor = interceptors[index]
    
    	//调用 nterceptor.intercept
        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
      }

接着来看各个拦截器。

#### RetryAndFollowUpInterceptor ####

会对连接做⼀些初始化⼯作，并且负责在请求失败时的重试，以及重定向的⾃动后续请求。它的存  
在，可以让重试和重定向对于开发者是⽆感知的

@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) {
        
          //前置工作
          //newExchangeFinder为true时，会新new一个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) {
        
              //判断是否可以恢复
              if (!recover(e.lastConnectException, call, request, requestSendStarted = false)) {
        
                throw e.firstConnectException.withSuppressed(recoveredFailures)
              } else {
        
                recoveredFailures += e.firstConnectException
              }
              newExchangeFinder = false
              //进行重试
              continue
            } catch (e: IOException) {
        
              //判断是否可以恢复
              if (!recover(e, call, request, requestSendStarted = e !is ConnectionShutdownException)) {
        
                throw e.withSuppressed(recoveredFailures)
              } else {
        
                recoveredFailures += e
              }
              newExchangeFinder = false
              //进行重试
              continue
            }
    
            // Attach the prior response if it exists. Such responses never have a body.
            if (priorResponse != null) {
        
              response = response.newBuilder()
                  .priorResponse(priorResponse.newBuilder()
                      .body(null)
                      .build())
                  .build()
            }
    
            val exchange = call.interceptorScopedExchange
            //是否进行重定向
            val followUp = followUpRequest(response, exchange)
    
            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)
          }
        }
      }

#### BridgeInterceptor ####

桥接你准备好的请求和即将要发布的请求，为最终的请求报文做完整性补充，以及初步解读响应报文

比如 Content-Type、Content-Length、Transfer-Encoding、Host、Content-Encoding等Header，自动帮你添加到Header

#### CacheInterceptor ####

缓存响应报文以及在新请求中应用之前缓存过的响应报文（如果可用）

> 请求之前，先看看有没有可用缓存  
> 请求之后，如果可以缓存，就缓存下来

#### ConnectInterceptor \[最关键\] ####

获取可用的 TCP(HTTP) 连接和 TLS (HTTPS) 连接，这个连接可能是当场建立，也可能是之前缓存的

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)
        //中置工作 
        //这里将exchange和request传入到CallServerInterceptor进行实质的请求与响应的 I/O 操作了
        return connectedChain.proceed(realChain.request)
        //没有后置工作
      } 
    }

先来看initExchange，初始化Exchange并返回

internal fun initExchange(chain: RealInterceptorChain): Exchange {
        
    	synchronized(connectionPool) {
        
    	  check(!noMoreExchanges) {
         "released" }
    	  check(exchange == null)
    	}
    	
    	//找到ExchangeCodec (code & decoder 编码解码器)
    	val codec = exchangeFinder!!.find(client, chain)
    	val result = Exchange(this, eventListener, exchangeFinder!!, codec)
    	this.interceptorScopedExchange = result
    	
    	synchronized(connectionPool) {
        
    	  this.exchange = result
    	  this.exchangeRequestDone = false
    	  this.exchangeResponseDone = false
    	  return result
    	}
    }

先来看`exchangeFinder!!.find(client, chain)`

fun find(
    client: OkHttpClient,
      chain: RealInterceptorChain
    ): ExchangeCodec {
        
      try {
        
      	//找到可用的健康连接
        val resultConnection = findHealthyConnection(
            connectTimeout = chain.connectTimeoutMillis,
            readTimeout = chain.readTimeoutMillis,
            writeTimeout = chain.writeTimeoutMillis,
            pingIntervalMillis = client.pingIntervalMillis,
            connectionRetryEnabled = client.retryOnConnectionFailure,
            doExtensiveHealthChecks = chain.request.method != "GET"
        )
        //用这个连接创建一个codec
        return resultConnection.newCodec(client, chain)
      } catch (e: RouteException) {
        
        trackFailure(e.lastConnectException)
        throw e
      } catch (e: IOException) {
        
        trackFailure(e)
        throw RouteException(e)
      }
    }

findHealthyConnection最终会调用`ExchangeFinder#findConnection`  
这里会通过5种方式获取

*  检查没有可用连接，有可用连接直接使用
 *  尝试从池里拿一个连接 (只拿不多路复用的连接)
 *  再次尝试从池里拿一个连接 (传入routes，用来处理Http2的连接合并，多路复用和不多路复用的都拿)
 *  创建RealConnection，自己进行连接
 *  再次尝试从池中拿一个连接 (只拿多路复用的连接)

在`findConnection`方法中，我们来看下`result.connect()`，是怎么进行连接的。

fun connect(
        connectTimeout: Int,
        readTimeout: Int,
        writeTimeout: Int,
        pingIntervalMillis: Int,
        connectionRetryEnabled: Boolean,
        call: Call,
        eventListener: EventListener
      ) {
        
        check(protocol == null) {
         "already connected" }
    
        var routeException: RouteException? = null
        val connectionSpecs = route.address.connectionSpecs
        val connectionSpecSelector = ConnectionSpecSelector(connectionSpecs)
    
        if (route.address.sslSocketFactory == null) {
        
          if (ConnectionSpec.CLEARTEXT !in connectionSpecs) {
        
            throw RouteException(UnknownServiceException(
                "CLEARTEXT communication not enabled for client"))
          }
          val host = route.address.url.host
          if (!Platform.get().isCleartextTrafficPermitted(host)) {
        
            throw RouteException(UnknownServiceException(
                "CLEARTEXT communication to $host not permitted by network security policy"))
          }
        } else {
        
          if (Protocol.H2_PRIOR_KNOWLEDGE in route.address.protocols) {
        
            throw RouteException(UnknownServiceException(
                "H2_PRIOR_KNOWLEDGE cannot be used with HTTPS"))
          }
        }
    
        while (true) {
        
          try {
        
          	//有时我们会去建立一个HTTP连接，但是我们要访问HTTPS的URL，那么就需要创建一个Tunnel
            if (route.requiresTunnel()) {
        
              connectTunnel(connectTimeout, readTimeout, writeTimeout, call, eventListener)
              if (rawSocket == null) {
        
                // We were unable to connect the tunnel but properly closed down our resources.
                break
              }
            } else {
        
              //建立TCP连接
              connectSocket(connectTimeout, readTimeout, call, eventListener)
            }
            //建立HTTP2(如果是HTTP2的话)
            //startHttp2
            //ConnetTls
            establishProtocol(connectionSpecSelector, pingIntervalMillis, call, eventListener)
            eventListener.connectEnd(call, route.socketAddress, route.proxy, protocol)
            break
          } catch (e: IOException) {
        
            socket?.closeQuietly()
            rawSocket?.closeQuietly()
            socket = null
            rawSocket = null
            source = null
            sink = null
            handshake = null
            protocol = null
            http2Connection = null
            allocationLimit = 1
    
            eventListener.connectFailed(call, route.socketAddress, route.proxy, null, e)
    
            if (routeException == null) {
        
              routeException = RouteException(e)
            } else {
        
              routeException.addConnectException(e)
            }
    
            if (!connectionRetryEnabled || !connectionSpecSelector.connectionFailed(e)) {
        
              throw routeException
            }
          }
        }
    
        if (route.requiresTunnel() && rawSocket == null) {
        
          throw RouteException(ProtocolException(
              "Too many tunnel connections attempted: $MAX_TUNNEL_ATTEMPTS"))
        }
    
        idleAtNs = System.nanoTime()
      }

#### CallServerInterceptor ####

它负责实质的请求与响应的 I/O 操作，即往 Socket ⾥写⼊请求数据，和从 Socket ⾥读取响应数据。

public Response intercept(Chain chain) throws IOException {
        
            final RealInterceptorChain realChain = (RealInterceptorChain) chain;
    
        // 获取对象
        Call call = realChain.call();
        final HttpCodec httpCodec = realChain.httpStream();
        StreamAllocation streamAllocation = realChain.streamAllocation();
        RealConnection connection = (RealConnection) realChain.connection();
        Request request = realChain.request();
    
        long sentRequestMillis = System.currentTimeMillis();
    
        realChain.eventListener().requestHeadersStart(call);
        // 发送状态行和头部数据到服务端
        httpCodec.writeRequestHeaders(request);
        realChain.eventListener().requestHeadersEnd(call, request);
    
        Response.Builder responseBuilder = null;
        // 判断是否有请求实体的请求（不是GET请求）
        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.
            // 如果头部信息添加了"Expect: 100-continue"，这个请求头字段的作用是在发送RequestBody前向服务器确认是否接受RequestBody，如果服务器不接受也就没有发送的必要了。
            // 有这个字段，相当于一次简单的握手操作，会等待拿到服务器返回的ResponseHeaders之后再继续，如果服务器接收RequestBody，会返回null。
            if ("100-continue".equalsIgnoreCase(request.header("Expect"))) {
        
                httpCodec.flushRequest();
                realChain.eventListener().responseHeadersStart(call);
                // 服务器返回
                responseBuilder = httpCodec.readResponseHeaders(true);
            }
    
            // 服务器同意接收，开始向流中写入RequestBody
            if (responseBuilder == null) {
        
                if (request.body() instanceof DuplexRequestBody) {
        
                    // Prepare a duplex body so that the application can send a request body later.
                    httpCodec.flushRequest();
                    CountingSink requestBodyOut = new CountingSink(httpCodec.createRequestBody(request, -1L));
                    ((DuplexRequestBody) request.body()).foldSink(requestBodyOut);
                } else {
        
                    // Write the request body if the "Expect: 100-continue" expectation was met.
                    realChain.eventListener().requestBodyStart(call);
                    long contentLength = request.body().contentLength();
                    CountingSink requestBodyOut =
                            new CountingSink(httpCodec.createRequestBody(request, contentLength));
                    BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut);
    
                    // 写入数据
                    request.body().writeTo(bufferedRequestBody);
                    bufferedRequestBody.close();
                    realChain.eventListener().requestBodyEnd(call, requestBodyOut.successfulCount);
                }
            } else if (!connection.isMultiplexed()) {
        
                streamAllocation.noNewStreams();
            }
        }
    
        if (!(request.body() instanceof DuplexRequestBody)) {
        
            // 结束请求，调用了sink.flush();
            httpCodec.finishRequest();
        }
    
        // 读取头部信息、状态码等
        if (responseBuilder == null) {
        
            realChain.eventListener().responseHeadersStart(call);
            responseBuilder = httpCodec.readResponseHeaders(false);
        }
    
        // 构建Response, 写入原请求，握手情况，请求时间，得到的响应时间
        responseBuilder
                .request(request)
                .handshake(streamAllocation.connection().handshake())
                .sentRequestAtMillis(sentRequestMillis)
                .receivedResponseAtMillis(System.currentTimeMillis());
        Internal.instance.initCodec(responseBuilder, httpCodec);
        Response response = responseBuilder.build();
    
        // 返回的状态码
        int code = response.code();
        if (code == 100) {
         // 再请求一次
            responseBuilder = httpCodec.readResponseHeaders(false);
    
            responseBuilder
                    .request(request)
                    .handshake(streamAllocation.connection().handshake())
                    .sentRequestAtMillis(sentRequestMillis)
                    .receivedResponseAtMillis(System.currentTimeMillis());
            Internal.instance.initCodec(responseBuilder, httpCodec);
            response = responseBuilder.build();
    
            code = response.code();
        }
    
        realChain.eventListener().responseHeadersEnd(call, response);
    
        // ResponseBody
        if (forWebSocket && code == 101) {
        
            // Connection is upgrading, but we need to ensure interceptors see a non-null response body.
            // 设置一个空的Body
            response = response.newBuilder()
                    .body(Util.EMPTY_RESPONSE)
                    .build();
        } else {
        
            // 读取Body信息
            response = response.newBuilder()
                    .body(httpCodec.openResponseBody(response))
                    .build();
        }
    
        // 如果设置了连接关闭，则断开连接
        if ("close".equalsIgnoreCase(response.request().header("Connection"))
                || "close".equalsIgnoreCase(response.header("Connection"))) {
        
            streamAllocation.noNewStreams();
        }
    
        //HTTP 204(no content) 代表响应报文中包含若干首部和一个状态行，但是没有实体的主体内容。
        //HTTP 205(reset content) 表示响应执行成功，重置页面（Form表单），方便用户下次输入
        //这里做了同样的处理，就是抛出协议异常。
        if ((code == 204 || code == 205) && response.body().contentLength() > 0) {
        
            throw new ProtocolException(
                    "HTTP " + code + " had non-zero Content-Length: " + response.body().contentLength());
        }
    
        return response;
    }

#### interceptor和NetworkInterceptor的区别 ####

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.add(new CallServerInterceptor(forWebSocket));

从上可以看到，interceptor和networkInterceptor有位置上的区别，interceptor拿到的是比较干净的Request，没有经过BridgeInterceptor等interceptor的处理，改东西都要放到interceptor(networkInterceptor已在发送之后了，所以该Request没有意义，且Connect-length什么都会出现异常)，networkInterceptor一般不会用到，只有在网络调试的时候(拿到纯净的，最直接的Response)，才会用到，比如第三方库: stetho

[20200414081326707.png]: https://img-blog.csdnimg.cn/20200414081326707.png