OkHttp-ConnectInterceptor源码解析

Posted 2022-11-24 zhuliyuan丶

ConnectInterceptor源码解析

本文基于okhttp3.10.0

1. 概述

ConnectInterceptor主要是用于建立连接，并再连接成功后将流封装成对象传递给下一个拦截器CallServerInterceptor与远端进行读写操作。

这个过程中会涉及比较多类我们简述下每个类的作用

• StreamAllocation：类似一个工厂用来创建连接RealConnection和与远端通信的流的封装对象HttpCodec
• ConnectionPool：连接池用来存储可用的连接，在条件符合的情况下进行连接复用
• HttpCodec：对输入输出流的封装对象，对于http1和2实现不同
• RealConnection：对tcp连接的封装

2. 源码解析

public final class ConnectInterceptor implements Interceptor 
  public final OkHttpClient client;

  public ConnectInterceptor(OkHttpClient client) 
    this.client = client;
  

  @Override public Response intercept(Chain chain) throws IOException 
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    Request request = realChain.request();
    StreamAllocation streamAllocation = realChain.streamAllocation();

    boolean doExtensiveHealthChecks = !request.method().equals("GET");
    HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks);
    RealConnection connection = streamAllocation.connection();

    return realChain.proceed(request, streamAllocation, httpCodec, connection);
  

可以看到代码非常少，主要就是从realChain中拿出StreamAllocation然后调用StreamAllocation#newStream()获取流的封装对象HttpCodec、StreamAllocation#connection()拿到连接对象RealConnection，然后将它们传递给下一个拦截器。

很显然主要逻辑是在StreamAllocation中实现的，而StreamAllocation是在第一个拦截器retryAndFollowUpInterceptor创建的直到ConnectInterceptor才使用

#RetryAndFollowUpInterceptor
@Override public Response intercept(Chain chain) throws IOException 
		    Request request = chain.request();
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    Call call = realChain.call();
    EventListener eventListener = realChain.eventListener();

    StreamAllocation streamAllocation = new StreamAllocation(client.connectionPool(),
        createAddress(request.url()), call, eventListener, callStackTrace);
    ...

可以看到构造方法一共传入了5个参数，我们只需要关注前三个即可

• client.connectionPool()：okhttp连接池
• createAddress(request.url())：将url解析为Address对象
• call：请求对象Call

2.1 ConnectionPool

连接池通过OkHttpClient实例的connectionPool()方法获取，默认初始化是在OkHttpClient.Builder构造函数

主要功能是用来缓存连接，当符合条件的时候进行连接复用，内部通过一个队列去缓存连接，当超过缓存时间后会自动清理过期连接。

先来看下ConnectionPool构造方法

private final int maxIdleConnections;//最大闲置连接数
private final long keepAliveDurationNs;//每个连接最大缓存时间

public ConnectionPool() 
    this(5, 5, TimeUnit.MINUTES);//默认最大缓存5个闲置连接，过期时间5分钟
  

  public ConnectionPool(int maxIdleConnections, long keepAliveDuration, TimeUnit timeUnit) 
    this.maxIdleConnections = maxIdleConnections;
    this.keepAliveDurationNs = timeUnit.toNanos(keepAliveDuration);

    // Put a floor on the keep alive duration, otherwise cleanup will spin loop.
    if (keepAliveDuration <= 0) 
      throw new IllegalArgumentException("keepAliveDuration <= 0: " + keepAliveDuration);
    
  

默认最大缓存5个连接，过期时间为5分钟，存储连接是通过ConnectionPool#put()方法

private final Deque<RealConnection> connections = new ArrayDeque<>();//缓存队列
boolean cleanupRunning;//开启清理过期连接任务标记位
private static final Executor executor = new ThreadPoolExecutor(0 /* corePoolSize */,
      Integer.MAX_VALUE /* maximumPoolSize */, 60L /* keepAliveTime */, TimeUnit.SECONDS,
      new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp ConnectionPool", true));

void put(RealConnection connection) 
    assert (Thread.holdsLock(this));
    if (!cleanupRunning) 
      cleanupRunning = true;
      executor.execute(cleanupRunnable);//在线程池中开启清理过期连接任务
    
    connections.add(connection);//添加到缓存队列
  

存储连接的时候，如果没开启清理任务则开启清理过期连接任务并缓存新的连接

  private final Runnable cleanupRunnable = new Runnable() 
    @Override public void run() 
      while (true) 
        long waitNanos = cleanup(System.nanoTime());//获取最近一个即将过期连接的倒计时
        if (waitNanos == -1) return;//-1则代表没有缓存连接了直接return
        if (waitNanos > 0) 
          long waitMillis = waitNanos / 1000000L;
          waitNanos -= (waitMillis * 1000000L);
          synchronized (ConnectionPool.this) 
            try 
              ConnectionPool.this.wait(waitMillis, (int) waitNanos);//wait最近一个即将过期连接的倒计时后在进行检测
             catch (InterruptedException ignored) 
            
          
        
      
    
  ;

清理过期连接runnable则是通过cleanup()方法获取最近一个即将过期连接的倒计时，为-1则代表缓存已经清空了直接return退出，否则wait一个即将超时的时间后在进行检查

  long cleanup(long now) 
    int inUseConnectionCount = 0;//使用的连接数
    int idleConnectionCount = 0;//闲置的连接数
    RealConnection longestIdleConnection = null;//闲置时间最长的连接
    long longestIdleDurationNs = Long.MIN_VALUE;

    synchronized (this) 
      for (Iterator<RealConnection> i = connections.iterator(); i.hasNext(); ) 
        RealConnection connection = i.next();//遍历缓存的连接

        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) //闲置最久的连接时间超过5分钟或者闲置连接数大于5
        connections.remove(longestIdleConnection);//移除当前连接
       else if (idleConnectionCount > 0) //闲置连接数大于0
        return keepAliveDurationNs - longestIdleDurationNs;//return缓存最久连接剩余过期时间
       else if (inUseConnectionCount > 0) //如果使用连接数大于0
        return keepAliveDurationNs;//return最大缓存时间
       else //没有缓存连接了
        cleanupRunning = false;//把标记位置为false
        return -1;//返回-1停止检测
      
    

遍历缓存的连接，找到闲置最久的连接，和使用与闲置连接数量

1. 当闲置最久的连接闲置时间大于最大闲置时间或者闲置连接数大于最大闲置连接数，直接移除该连接
2. 当闲置连接数大于0则计算闲置最久的连接过期时间作为倒计时返回
3. 当使用连接数大于0，返回最大缓存时间作为倒计时
4. 没有缓存连接了把标志位cleanupRunning置为false并返回-1停止清理任务。

这里需要注意下的是对于连接是否在使用的判断pruneAndGetAllocationCount()方法

  private int pruneAndGetAllocationCount(RealConnection connection, long now) 
    List<Reference<StreamAllocation>> references = connection.allocations;//拿到RealConnection中StreamAllocation引用
    for (int i = 0; i < references.size(); ) //遍历
      Reference<StreamAllocation> reference = references.get(i);

      if (reference.get() != null) //如果持有StreamAllocation则代表有使用
        i++;
        continue;
      

      references.remove(i);//如果reference.get() == null则代表内存泄露了，移除该引用
      connection.noNewStreams = true;//该连接不再创建新的流

      if (references.isEmpty()) //如果StreamAllocation引用为空
        connection.idleAtNanos = now - keepAliveDurationNs;//将当前闲置时间设置为now - keepAliveDurationNs即满足清除条件会被直接从缓存移除
        return 0;
      
    

    return references.size();//返回当前连接使用数
  

对于连接是否在使用是通过RealConnection中的成员变量allocations来判断，如果有请求正在使用则会将StreamAllocation对象的引用存储到allocations中，具体存储的代码是在StreamAllocation中后面会说到

看完put再来看get

  @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;
  

遍历缓存连接，如果满足条件则复用该连接

  public boolean isEligible(Address address, @Nullable Route route) 
    if (allocations.size() >= allocationLimit || noNewStreams) return false;//如果当前连接流使用数>=最大并发或者不能创建新流了则return false

    if (!Internal.instance.equalsNonHost(this.route.address(), address)) return false;//地址中除了host有不相同的返回false

    if (address.url().host().equals(this.route().address().url().host())) //如果host相同则返回true
      return true; 
    

    if (http2Connection == null) return false;//如果上面条件都不满足并且不是http2则返回false http2不再本次源码讨论范围下面就不分析了
    if (route == null) return false;
    if (route.proxy().type() != Proxy.Type.DIRECT) return false;
    if (this.route.proxy().type() != Proxy.Type.DIRECT) return false;
    if (!this.route.socketAddress().equals(route.socketAddress())) return false;

    if (route.address().hostnameVerifier() != OkHostnameVerifier.INSTANCE) return false;
    if (!supportsUrl(address.url())) return false;

    try 
      address.certificatePinner().check(address.url().host(), handshake().peerCertificates());
     catch (SSLPeerUnverifiedException e) 
      return false;
    

可以看到在满足如下三个条件的时候http1连接可以服用

1. 小于当前连接最大并发数，http1最大并发为1并且可以创建新的流
2. 请求路径中除了host以外的全部相同
3. host也相同

满足以上三个条件即可复用缓存的连接

复用是通过streamAllocation.acquire(connection, true)方法实现

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

    this.connection = connection;
    this.reportedAcquired = reportedAcquired;
    connection.allocations.add(new StreamAllocationReference(this, callStackTrace));
  

即通过成员变量存储connection，并connection.allocations.add(new StreamAllocationReference(this, callStackTrace));给当前连接添加上StreamAllocation引用代表有一个流正在使用该连接。

那么总结下ConnectionPool是用来缓存http连接的，当条件满足的时候复用连接，默认每个连接最大缓存时长5分钟，最大缓存数量5个。

2.2 Address

在RetryAndFollowUpInterceptor#createAddress()方法创建

  private Address createAddress(HttpUrl url) 
    SSLSocketFactory sslSocketFactory = null;
    HostnameVerifier hostnameVerifier = null;
    CertificatePinner certificatePinner = null;
    if (url.isHttps()) //请求是https
      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());
  

当是https的时候SSLSocketFactory、HostnameVerifier、CertificatePinner才会被赋值，后面会通过SSLSocketFactory ！= null这个判断请求是不是https，构造的时候只有host和port是从url中取得，其他都是在OkHttpClient的配置。

  public Address(String uriHost, int uriPort, Dns dns, SocketFactory socketFactory,
      @Nullable SSLSocketFactory sslSocketFactory, @Nullable HostnameVerifier hostnameVerifier,
      @Nullable CertificatePinner certificatePinner, Authenticator proxyAuthenticator,
      @Nullable Proxy proxy, List<Protocol> protocols, List<ConnectionSpec> connectionSpecs,
      ProxySelector proxySelector) 
    this.url = new HttpUrl.Builder()
        .scheme(sslSocketFactory != null ? "https" : "http")
        .host(uriHost)
        .port(uriPort)
        .build();

    if (dns == null) throw new NullPointerException("dns == null");
    this.dns = dns;

    if (socketFactory == null) throw new NullPointerException("socketFactory == null");
    this.socketFactory = socketFactory;

    if (proxyAuthenticator == null) 
      throw new NullPointerException("proxyAuthenticator == null");
    
    this.proxyAuthenticator = proxyAuthenticator;

    if (protocols == null) throw new NullPointerException("protocols == null");
    this.protocols = Util.immutableList(protocols);

    if (connectionSpecs == null) throw new NullPointerException("connectionSpecs == null");
    this.connectionSpecs = Util.immutableList(connectionSpecs);

    if (proxySelector == null) throw new NullPointerException("proxySelector == null");
    this.proxySelector = proxySelector;

    this.proxy = proxy;
    this.sslSocketFactory = sslSocketFactory;
    this.hostnameVerifier = hostnameVerifier;
    this.certificatePinner = certificatePinner;
  

可以看到构造函数只是将传入的属性通过成员变量存储起来。

总结下Address就是将url分解，然后通过成员变量存储不同的数据供后面使用。

2.3 StreamAllocation

上面把两个重要的对象说完了，那么回到ConnectInterceptor#intercept

  @Override public Response intercept(Chain chain) throws IOException 
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    Request request = realChain.request();
    StreamAllocation streamAllocation = realChain.streamAllocation();

    boolean doExtensiveHealthChecks = !request.method().equals("GET");//非get请求需要额外的检查
    HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks);//获取流的封装对象
    RealConnection connection = streamAllocation.connection();//获取连接

    return realChain.proceed(request, streamAllocation, httpCodec, connection);
  

调用StreamAllocation#newStream()获取流的封装对象HttpCodec、调用StreamAllocation#connection()拿到连接对象RealConnection，然后将它们传递