okhttp3源码分析（三）：复用连接池

发表于 | 分类于 | 阅读次数：

前言

向服务器发起请求时，header 部分如果有 Connection: keep-alive，表示客户端和服务器之间保存长连接，这个连接是可以复用的。长连接在 HTTP/1.1 中默认开启。

连接的复用可以提高性能。没有连接复用的话，发起 http 请求时要先建立 TCP 连接，然后传递数据，最后再释放连接。如果同一个客户端在某段时间进行频繁的请求操作，这时频繁的创建和释放连接会导致性能低下。而如果可以连接复用，那么在 timeout 空闲时间内，连接不会关闭，这样就可以减少连接的创建和释放，大幅提高性能。

在 okhttp 中，也有一个复用连接池 ConnectionPool，可以进行连接复用。本文将会从源码分析 okhttp 的复用连接池，以下代码基于 okhttp-3.10.0。

哪里用到了 ConnectionPool

在分析 ConnectionPool前，先看一下在哪里会用到 ConnectionPool。ConnectionPool 在 RealConnection 和 StreamAllocation 中都有用到。

RealConnection 对 Socket 进行了包装，代表一条连接，如果拥有一个 RealConnection 就代表有了一条客户端和服务器之间的通信链路。StreamAllocation 起到桥梁的作用，它负责为一次“请求”寻找“连接”并建立“流”，其中连接对应 RealConnection，流对应 HttpCodec。

ConnectInterceptor 负责和服务器建立连接，在它的 intercept 方法中，可以看到 StreamAllocation 寻找“连接”并建立“流”的过程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@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);
}

StreamAllocation 会通过 findConnection 方法找到 RealConnection，StreamAllocation 在创建 RealConnection 时会传入一个 ConnectionPool对象，该对象是在 StreamAllocation 的构造方法中创建的。

StreamAllocation 的 findConnection 方法和构造方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private final ConnectionPool connectionPool;

public StreamAllocation(ConnectionPool connectionPool, Address address, Call call,
    EventListener eventListener, Object callStackTrace) {
  this.connectionPool = connectionPool;
  
  // ...
}

private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout,
    int pingIntervalMillis, boolean connectionRetryEnabled) throws IOException {
  // ...省略其他代码
  
  RealConnection result = null;
  
  result = new RealConnection(connectionPool, selectedRoute);

  return result;        
}

那么 StreamAllocation 又是在什么时候创建的呢？答案是在 RetryAndFollowUpInterceptor 的 intercept 中：

1
2
3
4
5
6
7
8
9
10
11
12
 @Override 
 public Response intercept(Chain chain) throws IOException {
   // ...

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

// 调用下一拦截器
   response = realChain.proceed(request, streamAllocation, null, null);
   
   //...
 }

可以看到，StreamAllocation 是在 RetryAndFollowUpInterceptor 中创建，并传递给之后的拦截器。

而传入的 ConnectionPool 是在 OkHttpClient 中创建的：

1
2
3
4
5
  connectionPool = new ConnectionPool();
  
public ConnectionPool connectionPool() {
  return connectionPool;
}

小结

通过上面的分析，可以看到 ConnectionPool 只会在 OkHttpClient 中创建一次。然后经由 OkHttpClient 传给 StreamAllocation，再由 StreamAllocation 传给 RealConnection。所以 ConnectionPool 只会在 StreamAllocation 和 RealConnection 中使用到。

ConnectionPool 分析

主要成员变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 属于 CacheThreadPool，用于清除过期的连接
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));

// 最大的空闲连接数
private final int maxIdleConnections;
// 连接的 keepAlive 时间
private final long keepAliveDurationNs;

// 存储着 RealConnection 的双向队列，双向队列同时具有队列和栈的性质
private final Deque<RealConnection> connections = new ArrayDeque<>();

// 记录连接失败的路线
final RouteDatabase routeDatabase = new RouteDatabase();

构造方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
 public ConnectionPool() {
   this(5, 5, TimeUnit.MINUTES);
 }

 public ConnectionPool(int maxIdleConnections, long keepAliveDuration, TimeUnit timeUnit) {

// 最大空闲连接数为 5
   this.maxIdleConnections = maxIdleConnections;
// 连接的 keepAlive 时间设置为 5 分钟
   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);
   }
 }

可以看到，ConnectionPool 默认最大空闲连接数为 5，连接的 keepAlive 时间设置为 5 分钟。

缓存操作

ConnectionPool 提供对 Deque 进行操作的方法有 put、get、connectionBecameIdle 和 evictAll，分别对应加入连接、获取连接、移除连接和移除所有连接。下面分析一下 put 和 get 方法：

put

1
2
3
4
5
6
7
8
9
 void put(RealConnection connection) {
// 清理空闲的连接
   if (!cleanupRunning) {
     cleanupRunning = true;
     executor.execute(cleanupRunnable);
   }
// 将 RealConnection 添加到队列中
   connections.add(connection);
 }

先开启线程清理空闲的连接，然后将 RealConnection 添加到缓存队列中

get

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
 RealConnection get(Address address, StreamAllocation streamAllocation, Route route) {
// 遍历缓存队列
   for (RealConnection connection : connections) {
  // 判断当前的缓存 RealConnection 是否可以作为当前请求的连接
     if (connection.isEligible(address, route)) {
	// 将得到的缓存 RealConnection 和 StreamAllocation 绑定
	// 每一个 StreamAllocation 只和一个 RealConnection 绑定
	// 但一个 RealConnection 可能和多个 StreamAllocation 进行绑定（Http2.0 可以）
	// 因为一个 StreamAllocation 对应一个请求，而一个 RealConnection（连接）上可能有多个请求（Http2.0 使用了多路复用技术）
       streamAllocation.acquire(connection, true);
       return connection;
     }
   }
   return null;
 }

该方法遍历缓存队列，通过 RealConnection 的 isEligible 方法判断当前缓存 RealConnection 是否可以作为当前请求的连接，该方法如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
 public boolean isEligible(Address address, @Nullable Route route) {
 
// HTTP1.1 的情况

// 当前连接已经绑定了其他流，不再接受新的流（在 Http1.x 中一个连接只能对应一个流）
   if (allocations.size() >= allocationLimit || noNewStreams) return false;

// http 和 ssl 配置要相同，才可使用当前连接
   if (!Internal.instance.equalsNonHost(this.route.address(), address)) return false;

// 如果 host 相同，可以使用当前连接
   if (address.url().host().equals(this.route().address().url().host())) {
     return true; // This connection is a perfect match.
   }

   // 省略 Http2.0 的情况
 }

可以看到，能够使用当前连接需要满足一下条件：

  1. 当前连接没有绑定其他流，因为在 Http1.x 中一个连接只能对应一个流
  2. http 和 ssl 配置要相同
  3. host 相同

如果当前缓存连接可用，就将其和 StreamAllocation 绑定。

自动清除空闲连接

在 put 方法中，会通过在线程池执行 cleanupRunnable 来清除空闲连接。该 cleanupRunnable 的定义如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
 private final Runnable cleanupRunnable = new Runnable() {
   @Override 
public void run() {
     while (true) {
	// 进行清理工作
       long waitNanos = cleanup(System.nanoTime());
       if (waitNanos == -1) return;
	// 调用 wait 方法等待一段时间，直到下次清理的开始
       if (waitNanos > 0) {
         long waitMillis = waitNanos / 1000000L;
         waitNanos -= (waitMillis * 1000000L);
         synchronized (ConnectionPool.this) {
           try {
             ConnectionPool.this.wait(waitMillis, (int) waitNanos);
           } catch (InterruptedException ignored) {
           }
         }
       }
     }
   }
 };

先调用 cleanup 方法进行清理并返回下次清理的间隔时间，之后调用 wait 方法等待一段时间，直到下次清理的开始。

接下来看 cleanup 方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 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;
       }
     }

  // 如果最长空闲时间超过了5分钟，或者空闲连接数超过了5，就移除空闲时间最长的连接，并马上再次清理
     if (longestIdleDurationNs >= this.keepAliveDurationNs
         || idleConnectionCount > this.maxIdleConnections) {
       connections.remove(longestIdleConnection);
     } 
  // 如果存在空闲连接，但最长空闲时间没有超5分钟且空闲连接数不超过5，不移除连接，返回最长空闲连接的剩余时间
  else if (idleConnectionCount > 0) {
       return keepAliveDurationNs - longestIdleDurationNs;
     } 
  // 如果不存在空闲连接，只有正在使用的连接，不移除连接，5分钟后再清理
  else if (inUseConnectionCount > 0) {
       return keepAliveDurationNs;
     } 
  // 没有任何连接，清除任务完成，线程执行完毕
  else {
       cleanupRunning = false;
       return -1;
     }
   }

   closeQuietly(longestIdleConnection.socket());

   return 0;
 }

该方法虽然很长，逻辑还是很容易懂得。执行步骤如下：

  1. 遍历各连接，得到正在使用的连接数、空闲的连接数、最长空闲时间和空闲时间最长的连接。
  2. 分四种情况判断是否清除连接以及返回相应间隔时间：
    1. 如果最长空闲时间超过了5分钟，或者空闲连接数超过了5，就移除空闲时间最长的连接，并马上再次清除
    2. 如果存在空闲连接，但最长空闲时间没有超5分钟且空闲连接数不超过5，不移除连接，返回最长空闲连接的剩余时间
    3. 如果不存在空闲连接，只有正在使用的连接，不移除连接，5分钟后再清理
    4. 如果没有任何连接，说明清除任务完成，线程执行完毕

总结

ConnectionPool（连接池）的作用是缓存连接，当然，这个缓存是有时间限制的，空闲时间超过 5 分钟的连接就会被清除。所以连接池中连接的复用是为了避免出现在一段时间内频繁地创建和销毁连接而导致性能下降。

参考

-------------　　　　本文到此结束　　感谢您的阅读　　　　-------------
0%