OkHttpClient源码分析（五）—— ConnectInterceptor和CallServerInterceptor

时间 2019-11-08

标签 okhttpclient 源码分析 connectinterceptor callserverinterceptor 繁體版

原文原文链接

上一篇咱们介绍了缓存拦截器CacheInterceptor，本篇将介绍剩下的两个拦截器: ConnectInterceptor和CallServerInterceptor缓存

ConnectInterceptor

该拦截器主要是负责创建可用的连接，主要做用是打开了与服务器的连接，正式开启了网络请求。查看其intercept()方法:bash

@Override public Response intercept(Chain chain) throws IOException {
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    //从拦截器链中获取StreamAllocation对象
    Request request = realChain.request();
    StreamAllocation streamAllocation = realChain.streamAllocation();
    
    //建立HttpCodec对象
    HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks);
    
    //获取realConnetion
    RealConnection connection = streamAllocation.connection();

    //执行下一个拦截器，返回response
    return realChain.proceed(request, streamAllocation, httpCodec, connection);
  }
复制代码

能够看到intercept中的处理很简单，主要有如下几步操做：服务器

从拦截器链中获取StreamAllocation对象，在讲解第一个拦截器RetryAndFollowUpInterceptor的时候，咱们已经初步了解了StreamAllocation对象，在RetryAndFollowUpInterceptor中仅仅只是建立了StreamAllocation对象，并无进行使用，到了ConnectInterceptor中，StreamAllocation才被真正使用到，该拦截器的主要功能都交给了StreamAllocation处理；网络
执行StreamAllocation对象的 newStream() 方法建立HttpCodec，用于处理编码Request和解码Response；socket
接着经过调用StreamAllocation对象的 connection() 方法获取到RealConnection对象，这个RealConnection对象是用来进行实际的网络IO传输的。ide
调用拦截器链的**proceed()**方法，执行下一个拦截器返回response对象。源码分析

上面咱们已经了解了ConnectInterceptor拦截器的intercept()方法的总体流程，主要的逻辑是在StreamAllocation对象中，咱们先看下它的 newStream() 方法:ui

public HttpCodec newStream(
      OkHttpClient client, Interceptor.Chain chain, boolean doExtensiveHealthChecks) {
    ...
    try {
      //建立RealConnection对象
      RealConnection resultConnection = findHealthyConnection(connectTimeout, readTimeout,
          writeTimeout, connectionRetryEnabled, doExtensiveHealthChecks);
      //建立HttpCodec对象
      HttpCodec resultCodec = resultConnection.newCodec(client, chain, this);
      
      synchronized (connectionPool) {
        codec = resultCodec;
        //返回HttpCodec对象
        return resultCodec;
      }
    } catch (IOException e) {
      throw new RouteException(e);
    }
  }
复制代码

newStream()方法中，主要是建立了RealConnection对象(用于进行实际的网络IO传输)和HttpCodec对象(用于处理编码Request和解码Response)，并将HttpCodec对象返回。this

findHealthyConnection()方法用于建立RealConnection对象:编码

private RealConnection findHealthyConnection(int connectTimeout, int readTimeout,
      int writeTimeout, boolean connectionRetryEnabled, boolean doExtensiveHealthChecks)
      throws IOException {
    while (true) {//while循环
      //获取RealConnection对象
      RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout,
          connectionRetryEnabled);
    
      //同步代码块判断RealConnection对象的successCount是否为0
      synchronized (connectionPool) {
        if (candidate.successCount == 0) {
          //若是为0则返回
          return candidate;
        }
      }

      //对连接池中不健康的连接作销毁处理
      if (!candidate.isHealthy(doExtensiveHealthChecks)) {
        noNewStreams();
        continue;
      }

      return candidate;
    }
  }
复制代码

以上代码主要作的事情有：

开启一个while循环，经过调用findConnection()方法获取RealConnection对象赋值给candidate;
若是candidate 的successCount 为0，直接返回candidate，while循环结束;
调用candidate的isHealthy()方法，进行“健康检查”，若是candidate是一个不“健康”的对象，其中不“健康”指的是Socket没有关闭、或者它的输入输出流没有关闭，则对调用noNewStreams()方法进行销毁处理，接着继续循环。

咱们看下findConnection()方法作了哪些操做：

private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout,
      boolean connectionRetryEnabled) throws IOException {
    ...
    RealConnection result = null;
    ...
    synchronized (connectionPool) {
      ...
      releasedConnection = this.connection;
      toClose = releaseIfNoNewStreams();
      if (this.connection != null) {
        //若是不为 null，则复用，赋值给 result
        result = this.connection;
        releasedConnection = null;
      }
      ...
      //若是result为 null，说明上面找不到能够复用的
      if (result == null) {
        //从链接池中获取，调用其get()方法
        Internal.instance.get(connectionPool, address, this, null);
        if (connection != null) {
          //找到对应的 RealConnection对象
          //更改标志位，赋值给 result
          foundPooledConnection = true;
          result = connection;
        } else {
          selectedRoute = route;
        }
      }
    }
    
    ...
    if (result != null) {
      //已经找到 RealConnection对象，直接返回
      return result;
    }
    
    ...
     //链接池中找不到，new一个
     result = new RealConnection(connectionPool, selectedRoute);
    ...
    
    ...
    //发起请求
    result.connect(
        connectTimeout, readTimeout, writeTimeout, connectionRetryEnabled, call, eventListener);
    ...
    //存进链接池中，调用其put()方法
    Internal.instance.put(connectionPool, result);
    ...
    return result;
  }
复制代码

以上代码主要作的事情有：

StreamAllocation的connection若是能够复用则复用;
若是connection不能复用，则从链接池中获取RealConnection对象，获取成功则返回;
若是链接池里没有，则new一个RealConnection对象;
调用RealConnection的connect()方法发起请求;
将RealConnection对象存进链接池中，以便下次复用;
返回RealConnection对象。

ConnectionPool 链接池介绍

刚才咱们说到从链接池中取出RealConnection对象时调用了Internal的get()方法，存进去的时候调用了其put()方法。其中Internal是一个抽象类，里面定义了一个静态变量instance：

public abstract class Internal {
    ...
    public static Internal instance;
    ...
}
复制代码

instance的实例化是在OkHttpClient的静态代码块中：

public class OkHttpClient implements Cloneable, Call.Factory, WebSocket.Factory {
  ...
  static {
      Internal.instance = new Internal() {
         ...
          @Override public RealConnection get(ConnectionPool pool, Address address,
          StreamAllocation streamAllocation, Route route) {
            return pool.get(address, streamAllocation, route);
         }
         ...
         @Override public void put(ConnectionPool pool, RealConnection connection) {
           pool.put(connection);
         }
      };
  }
  ...
}
复制代码

这里咱们能够看到实际上 Internal 的 get()方法和put()方法是调用了 ConnectionPool 的get()方法和put()方法，这里咱们简单看下ConnectionPool的这两个方法：

private final Deque<RealConnection> connections = new ArrayDeque<>();

@Nullable RealConnection get(Address address, StreamAllocation streamAllocation, Route route) {
    assert (Thread.holdsLock(this));
    for (RealConnection connection : connections) {
      if (connection.isEligible(address, route)) {
        streamAllocation.acquire(connection, true);
        return connection;
      }
    }
    return null;
  }
复制代码

在get()方法中，经过遍历connections(用于存放RealConnection的ArrayDeque队列)，调用RealConnection的isEligible()方法判断其是否可用，若是可用就会调用streamAllocation的acquire()方法，并返回connection。

咱们看下调用StreamAllocation的acquire()方法到底作了什么操做:

public void acquire(RealConnection connection, boolean reportedAcquired) {
    assert (Thread.holdsLock(connectionPool));
    if (this.connection != null) throw new IllegalStateException();

    //赋值给全局变量
    this.connection = connection;
    this.reportedAcquired = reportedAcquired;
    //建立StreamAllocationReference对象并添加到allocations集合中
    connection.allocations.add(new StreamAllocationReference(this, callStackTrace));
  }
复制代码

先是从链接池中获取的RealConnection对象赋值给StreamAllocation的成员变量connection；
建立StreamAllocationReference对象(StreamAllocation对象的弱引用)，并添加到RealConnection的allocations集合中，到时能够经过allocations集合的大小来判断网络链接次数是否超过OkHttp指定的链接次数。

接着咱们查看ConnectionPool 的put()方法:

void put(RealConnection connection) {
    assert (Thread.holdsLock(this));
    if (!cleanupRunning) {
      cleanupRunning = true;
      executor.execute(cleanupRunnable);
    }
    connections.add(connection);
  }
复制代码

put()方法在将链接添加到链接池以前，会先执行清理任务，经过判断cleanupRunning是否在执行，若是当前清理任务没有执行，则更改cleanupRunning标识，并执行清理任务cleanupRunnable。

咱们看下清理任务cleanupRunnable中到底作了哪些操做：

private final Runnable cleanupRunnable = new Runnable() {
    @Override public void run() {
      while (true) {
        //对链接池进行清理，返回进行下次清理的间隔时间。
        long waitNanos = cleanup(System.nanoTime());
        if (waitNanos == -1) return;
        if (waitNanos > 0) {
          long waitMillis = waitNanos / 1000000L;
          waitNanos -= (waitMillis * 1000000L);
          synchronized (ConnectionPool.this) {
            try {
              //进行等待
              ConnectionPool.this.wait(waitMillis, (int) waitNanos);
            } catch (InterruptedException ignored) {
            }
          }
        }
      }
    }
  };
复制代码

能够看到run()方法里面是一个while死循环，其中调用了cleanup()方法进行清理操做，同时会返回进行下次清理的间隔时间，若是返回的时间间隔为-1，则会结束循环，若是不是-1，则会调用wait()方法进行等待，等待完成后又会继续循环执行，具体的清理操做在cleanup()方法中:

long cleanup(long now) {
    //正在使用的链接数
    int inUseConnectionCount = 0;
    //空闲的链接数
    int idleConnectionCount = 0;
    //空闲时间最长的链接
    RealConnection longestIdleConnection = null;
    //最大的空闲时间，初始化为 Long 的最小值，用于记录全部空闲链接中空闲最久的时间
    long longestIdleDurationNs = Long.MIN_VALUE;

    synchronized (this) {
      //for循环遍历connections队列
      for (Iterator<RealConnection> i = connections.iterator(); i.hasNext(); ) {
        RealConnection connection = i.next();

        //若是遍历到的链接正在使用，则跳过，continue继续遍历下一个
        if (pruneAndGetAllocationCount(connection, now) > 0) {
          inUseConnectionCount++;
          continue;
        }

        //当前链接处于空闲，空闲链接数++
        idleConnectionCount++;

        //计算空闲时间
        long idleDurationNs = now - connection.idleAtNanos;
        //空闲时间若是超过最大空闲时间
        if (idleDurationNs > longestIdleDurationNs) {
          //从新赋值最大空闲时间
          longestIdleDurationNs = idleDurationNs;
          //赋值空闲最久的链接
          longestIdleConnection = connection;
        }
      }

      if (longestIdleDurationNs >= this.keepAliveDurationNs
          || idleConnectionCount > this.maxIdleConnections) {
        //若是最大空闲时间超过空闲保活时间或空闲链接数超过最大空闲链接数限制
        //则移除该链接
        connections.remove(longestIdleConnection);
      } else if (idleConnectionCount > 0) {
        //若是存在空闲链接
        //计算出线程清理的时间即(保活时间-最大空闲时间)，并返回
        return keepAliveDurationNs - longestIdleDurationNs;
      } else if (inUseConnectionCount > 0) {
         //没有空闲链接，返回keepAliveDurationNs
        return keepAliveDurationNs;
      } else {
        //链接池中没有链接存在，返回-1
        cleanupRunning = false;
        return -1;
      }
    }

    //关闭空闲时间最长的链接
    closeQuietly(longestIdleConnection.socket());

    return 0;
  }
复制代码

cleanup()方法经过for循环遍历connections队列，记录最大空闲时间和空闲时间最长的链接；若是存在超过空闲保活时间或空闲链接数超过最大空闲链接数限制的链接，则从connections中移除，最后执行关闭该链接的操做。

主要是经过pruneAndGetAllocationCount()方法判断链接是否处于空闲状态：

private int pruneAndGetAllocationCount(RealConnection connection, long now) {
    List<Reference<StreamAllocation>> references = connection.allocations;
    for (int i = 0; i < references.size(); ) {
      Reference<StreamAllocation> reference = references.get(i);

      if (reference.get() != null) {
        i++;
        continue;
      }

      ...
      
      references.remove(i);
      connection.noNewStreams = true;
      
      ...
      
      if (references.isEmpty()) {
        connection.idleAtNanos = now - keepAliveDurationNs;
        return 0;
      }
    }

    return references.size();
  }
复制代码

该方法经过for循环遍历RealConnection的allocations集合，若是当前遍历到的StreamAllocation被使用就遍历下一个，不然就将其移除，若是移除后列表为空，则返回0，因此若是方法的返回值为0则说明当前链接处于空闲状态，若是返回值大于0则说明链接正在使用。

CallServerInterceptor

接下来说解最后一个拦截器CallServerInterceptor了，查看intercept()方法:

@Override public Response intercept(Chain chain) throws IOException {
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    //相关对象的获取 
    HttpCodec httpCodec = realChain.httpStream();
    StreamAllocation streamAllocation = realChain.streamAllocation();
    RealConnection connection = (RealConnection) realChain.connection();
    Request request = realChain.request();

    ...
    
    //写入请求头
    httpCodec.writeRequestHeaders(request);

    Response.Builder responseBuilder = null;
    if (HttpMethod.permitsRequestBody(request.method()) && request.body() != null) {
      //判断是否有请求体
      if ("100-continue".equalsIgnoreCase(request.header("Expect"))) {
        //询问服务器是否愿意接收请求体
        httpCodec.flushRequest();//刷新请求
        realChain.eventListener().responseHeadersStart(realChain.call());
        responseBuilder = httpCodec.readResponseHeaders(true);
      }

      if (responseBuilder == null) {
        //服务器愿意接收请求体
        //写入请求体
        ...
      } else if (!connection.isMultiplexed()) {
        streamAllocation.noNewStreams();
      }
    }

    //结束请求
    httpCodec.finishRequest();

    if (responseBuilder == null) {
      realChain.eventListener().responseHeadersStart(realChain.call());
      //根据服务器返回的数据构建 responseBuilder对象
      responseBuilder = httpCodec.readResponseHeaders(false);
    }

    //构建 response对象
    Response response = responseBuilder
        .request(request)
        .handshake(streamAllocation.connection().handshake())
        .sentRequestAtMillis(sentRequestMillis)
        .receivedResponseAtMillis(System.currentTimeMillis())
        .build();

    ...
    
    //设置 response的 body
    response = response.newBuilder()
          .body(httpCodec.openResponseBody(response))
          .build();
   
   //若是请求头中 Connection对应的值为 close，则关闭链接
    if ("close".equalsIgnoreCase(response.request().header("Connection"))
        || "close".equalsIgnoreCase(response.header("Connection"))) {
      streamAllocation.noNewStreams();
    }
    
    ...
    
    return response;
  }
复制代码

以上代码具体的流程：

从拦截器链中获取到保存的相关对象；
调用HttpCodec的writeRequestHeaders()方法写入请求头;
判断是否须要写入请求体，先是判断请求方法，若是知足，请求头经过携带特殊字段Expect: 100-continue来询问服务器是否愿意接收请求体；
结束请求；
根据服务器返回的数据构建response对象；
关闭链接；
返回response;

好了，到这里OkHttpClient源码分析就结束了，相信看完本套源码解析会加深你对OkHttpClient的认识，同时也学到了其巧妙的代码设计思路，在阅读源码的过程当中，咱们的编码能力也逐步提高，若是想要写更加优质的代码，阅读源码是一件颇有帮助的事。