269      阿裏雲  技術社區[雲棲]

OkHttp 3.7源碼分析（五）——連接池

OkHttp3.7源碼分析文章列表如下：

• OkHttp源碼分析——整體架構
• OkHttp源碼分析——攔截器
• OkHttp源碼分析——任務隊列
• OkHttp源碼分析——緩存策略
• OkHttp源碼分析——多路複用

接下來講下OkHttp的連接池管理，這也是OkHttp的核心部分。通過維護連接池，最大限度重用現有連接，減少網絡連接的創建開銷，以此提升網絡請求效率。

1. 背景

1.1 keep-alive機製

在HTTP1.0中HTTP的請求流程如下：

這種方法的好處是簡單，各個請求互不幹擾。但在複雜的網絡請求場景下這種方式幾乎不可用。例如：瀏覽器加載一個HTML網頁，HTML中可能需要加載數十個資源，典型場景下這些資源中大部分來自同一個站點。按照HTTP1.0的做法，這需要建立數十個TCP連接，每個連接負責一個資源請求。創建一個TCP連接需要3次握手，而釋放連接則需要2次或4次握手。重複的創建和釋放連接極大地影響了網絡效率，同時也增加了係統開銷。

為了有效地解決這一問題，HTTP/1.1提出了Keep-Alive機製：當一個HTTP請求的數據傳輸結束後，TCP連接不立即釋放，如果此時有新的HTTP請求，且其請求的Host通上次請求相同，則可以直接複用為釋放的TCP連接，從而省去了TCP的釋放和再次創建的開銷，減少了網絡延時:

在現代瀏覽器中，一般同時開啟6～8個keepalive connections的socket連接，並保持一定的鏈路生命，當不需要時再關閉；而在服務器中，一般是由軟件根據負載情況(比如FD最大值、Socket內存、超時時間、棧內存、棧數量等)決定是否主動關閉。

1.2 HTTP/2

在HTTP/1.x中，如果客戶端想發起多個並行請求必須建立多個TCP連接，這無疑增大了網絡開銷。另外HTTP/1.x不會壓縮請求和響應報頭，導致了不必要的網絡流量；HTTP/1.x不支持資源優先級導致底層TCP連接利用率低下。而這些問題都是HTTP/2要著力解決的。簡單來說HTTP/2主要解決了以下問題：

• 報頭壓縮：HTTP/2使用HPACK壓縮格式壓縮請求和響應報頭數據，減少不必要流量開銷
• 請求與響應複用：HTTP/2通過引入新的二進製分幀層實現了完整的請求和響應複用，客戶端和服務器可以將HTTP消息分解為互不依賴的幀，然後交錯發送，最後再在另一端將其重新組裝
• 指定數據流優先級：將 HTTP 消息分解為很多獨立的幀之後，我們就可以複用多個數據流中的幀，客戶端和服務器交錯發送和傳輸這些幀的順序就成為關鍵的性能決定因素。為了做到這一點，HTTP/2 標準允許每個數據流都有一個關聯的權重和依賴關係
• 流控製：HTTP/2 提供了一組簡單的構建塊，這些構建塊允許客戶端和服務器實現其自己的數據流和連接級流控製

HTTP/2所有性能增強的核心在於新的二進製分幀層，它定義了如何封裝HTTP消息並在客戶端與服務器之間進行傳輸:

同時HTTP/2引入了三個新的概念：

• 數據流：基於TCP連接之上的邏輯雙向字節流，對應一個請求及其響應。客戶端每發起一個請求就建立一個數據流，後續該請求及其響應的所有數據都通過該數據流傳輸
• 消息：一個請求或響應對應的一係列數據幀
• 幀：HTTP/2的最小數據切片單位

上述概念之間的邏輯關係：

• 所有通信都在一個 TCP 連接上完成，此連接可以承載任意數量的雙向數據流
• 每個數據流都有一個唯一的標識符和可選的優先級信息，用於承載雙向消息
• 每條消息都是一條邏輯 HTTP 消息（例如請求或響應），包含一個或多個幀
• 幀是最小的通信單位，承載著特定類型的數據，例如 HTTP 標頭、消息負載，等等。 來自不同數據流的幀可以交錯發送，然後再根據每個幀頭的數據流標識符重新組裝
• 每個HTTP消息被分解為多個獨立的幀後可以交錯發送，從而在宏觀上實現了多個請求或響應並行傳輸的效果。這類似於多進程環境下的時間分片機製

2. 連接池的使用與分析

無論是HTTP/1.1的Keep-Alive機製還是HTTP/2的多路複用機製，在實現上都需要引入連接池來維護網絡連接。接下來看下OkHttp中的連接池實現。

OkHttp內部通過ConnectionPool來管理連接池，首先來看下ConnectionPool的主要成員：

public final class ConnectionPool {
  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));

  /** The maximum number of idle connections for each address. */
  private final int maxIdleConnections;
  private final long keepAliveDurationNs;
  private final Runnable cleanupRunnable = new Runnable() {
    @Override public void run() {
        ......
    }
  };

  private final Deque<RealConnection> connections = new ArrayDeque<>();
  final RouteDatabase routeDatabase = new RouteDatabase();
  boolean cleanupRunning;
  ......

    /**
    *返回符合要求的可重用連接，如果沒有返回NULL
   */
  RealConnection get(Address address, StreamAllocation streamAllocation, Route route) {
    ......
  }

  /*
  * 去除重複連接。主要針對多路複用場景下一個address隻需要一個連接
  */
  Socket deduplicate(Address address, StreamAllocation streamAllocation) {
    ......
    }

  /*
  * 將連接加入連接池
  */
  void put(RealConnection connection) {
      ......
  }

  /*
  * 當有連接空閑時喚起cleanup線程清洗連接池
  */
  boolean connectionBecameIdle(RealConnection connection) {
      ......
  }

  /**
   * 掃描連接池，清除空閑連接
  */
  long cleanup(long now) {
    ......
  }

  /*
   * 標記泄露連接
  */
  private int pruneAndGetAllocationCount(RealConnection connection, long now) {
    ......
  }
}

相關概念：

• Call：對Http請求的封裝
• Connection/RealConnection:物理連接的封裝，其內部有List<WeakReference<StreamAllocation>>的引用計數
• StreamAllocation: okhttp中引入了StreamAllocation負責管理一個連接上的流，同時在connection中也通過一個StreamAllocation的引用的列表來管理一個連接的流，從而使得連接與流之間解耦。關於StreamAllocation的定義可以看下這篇文章:okhttp源碼學習筆記（二）-- 連接與連接管理
• connections: Deque雙端隊列，用於維護連接的容器
• routeDatabase:用來記錄連接失敗的Route的黑名單，當連接失敗的時候就會把失敗的線路加進去

2.1 實例化

首先來看下ConnectionPool的實例化過程，一個OkHttpClient隻包含一個ConnectionPool，其實例化過程也在OkHttpClient的實例化過程中實現，值得一提的是ConnectionPool各個方法的調用並沒有直接對外暴露，而是通過OkHttpClient的Internal接口統一對外暴露：

public class OkHttpClient implements Cloneable, Call.Factory, WebSocket.Factory {
    static {
    Internal.instance = new Internal() {
      @Override public void addLenient(Headers.Builder builder, String line) {
        builder.addLenient(line);
      }

      @Override public void addLenient(Headers.Builder builder, String name, String value) {
        builder.addLenient(name, value);
      }

      @Override public void setCache(Builder builder, InternalCache internalCache) {
        builder.setInternalCache(internalCache);
      }

      @Override public boolean connectionBecameIdle(
          ConnectionPool pool, RealConnection connection) {
        return pool.connectionBecameIdle(connection);
      }

      @Override public RealConnection get(ConnectionPool pool, Address address,
          StreamAllocation streamAllocation, Route route) {
        return pool.get(address, streamAllocation, route);
      }

      @Override public boolean equalsNonHost(Address a, Address b) {
        return a.equalsNonHost(b);
      }

      @Override public Socket deduplicate(
          ConnectionPool pool, Address address, StreamAllocation streamAllocation) {
        return pool.deduplicate(address, streamAllocation);
      }

      @Override public void put(ConnectionPool pool, RealConnection connection) {
        pool.put(connection);
      }

      @Override public RouteDatabase routeDatabase(ConnectionPool connectionPool) {
        return connectionPool.routeDatabase;
      }

      @Override public int code(Response.Builder responseBuilder) {
        return responseBuilder.code;
      }

      @Override
      public void apply(ConnectionSpec tlsConfiguration, SSLSocket sslSocket, boolean isFallback)       {
        tlsConfiguration.apply(sslSocket, isFallback);
      }

      @Override public HttpUrl getHttpUrlChecked(String url)
          throws MalformedURLException, UnknownHostException {
        return HttpUrl.getChecked(url);
      }

      @Override public StreamAllocation streamAllocation(Call call) {
        return ((RealCall) call).streamAllocation();
      }

      @Override public Call newWebSocketCall(OkHttpClient client, Request originalRequest) {
        return new RealCall(client, originalRequest, true);
      }
    };
     ......
}

這樣做的原因是：

Escalate internal APIs in {@code okhttp3} so they can be used from OkHttp's implementation
packages. The only implementation of this interface is in {@link OkHttpClient}.

Internal的唯一實現在OkHttpClient中，OkHttpClient通過這種方式暴露其API給外部類使用。

2.2 連接池維護

ConnectionPool內部通過一個雙端隊列(dequeue)來維護當前所有連接，主要涉及到的操作包括：

• put：放入新連接
• get：從連接池中獲取連接
• evictAll：關閉所有連接
• connectionBecameIdle：連接變空閑後調用清理線程
• deduplicate：清除重複的多路複用線程

2.2.1 StreamAllocation.findConnection

get是ConnectionPool中最為重要的方法，StreamAllocation在其findConnection方法內部通過調用get方法為其找到stream找到合適的連接，如果沒有則新建一個連接。首先來看下findConnection的邏輯：

private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout,
                                        boolean connectionRetryEnabled) throws IOException {
    Route selectedRoute;
    synchronized (connectionPool) {
      if (released) throw new IllegalStateException("released");
      if (codec != null) throw new IllegalStateException("codec != null");
      if (canceled) throw new IOException("Canceled");

      // 一個StreamAllocation刻畫的是一個Call的數據流動，一個Call可能存在多次請求(重定向，Authenticate等)，所以當發生類似重定向等事件時優先使用原有的連接
      RealConnection allocatedConnection = this.connection;
      if (allocatedConnection != null && !allocatedConnection.noNewStreams) {
        return allocatedConnection;
      }

      // 試圖從連接池中找到可複用的連接
      Internal.instance.get(connectionPool, address, this, null);
      if (connection != null) {
        return connection;
      }

      selectedRoute = route;
    }

    // 獲取路由配置，所謂路由其實就是代理，ip地址等參數的一個組合
    if (selectedRoute == null) {
      selectedRoute = routeSelector.next();
    }

    RealConnection result;
    synchronized (connectionPool) {
      if (canceled) throw new IOException("Canceled");

      //拿到路由後可以嚐試重新從連接池中獲取連接，這裏主要針對http2協議下清除域名碎片機製
      Internal.instance.get(connectionPool, address, this, selectedRoute);
      if (connection != null) return connection;

      //新建連接
      route = selectedRoute;
      refusedStreamCount = 0;
      result = new RealConnection(connectionPool, selectedRoute);
      //修改result連接stream計數，方便connection標記清理
      acquire(result);
    }

    // Do TCP + TLS handshakes. This is a blocking operation.
    result.connect(connectTimeout, readTimeout, writeTimeout, connectionRetryEnabled);
    routeDatabase().connected(result.route());

    Socket socket = null;
    synchronized (connectionPool) {
      // 將新建的連接放入到連接池中
      Internal.instance.put(connectionPool, result);

      // 如果同時存在多個連向同一個地址的多路複用連接，則關閉多餘連接，隻保留一個
      if (result.isMultiplexed()) {
        socket = Internal.instance.deduplicate(connectionPool, address, this);
        result = connection;
      }
    }
    closeQuietly(socket);

    return result;
  }

其主要邏輯大致分為以下幾個步驟：

• 查看當前streamAllocation是否有之前已經分配過的連接，有則直接使用
• 從連接池中查找可複用的連接，有則返回該連接
• 配置路由，配置後再次從連接池中查找是否有可複用連接，有則直接返回
• 新建一個連接，並修改其StreamAllocation標記計數，將其放入連接池中
• 查看連接池是否有重複的多路複用連接，有則清除

2.2.2 ConnectionPool.get

接下來再來看get方法的源碼：

[ConnectionPool.java]
  RealConnection get(Address address, StreamAllocation streamAllocation, Route route) {
    assert (Thread.holdsLock(this));
    for (RealConnection connection : connections) {
      if (connection.isEligible(address, route)) {
        streamAllocation.acquire(connection);
        return connection;
      }
    }
    return null;
  }

其邏輯比較簡單，遍曆當前連接池，如果有符合條件的連接則修改器標記計數，然後返回。這裏的關鍵邏輯在RealConnection.isEligible方法：

[RealConnection.java]
/**
   * Returns true if this connection can carry a stream allocation to {@code address}. If non-null
   * {@code route} is the resolved route for a connection.
   */
  public boolean isEligible(Address address, Route route) {
    // If this connection is not accepting new streams, we're done.
    if (allocations.size() >= allocationLimit || noNewStreams) return false;

    // If the non-host fields of the address don't overlap, we're done.
    if (!Internal.instance.equalsNonHost(this.route.address(), address)) return false;

    // If the host exactly matches, we're done: this connection can carry the address.
    if (address.url().host().equals(this.route().address().url().host())) {
      return true; // This connection is a perfect match.
    }

    // At this point we don't have a hostname match. But we still be able to carry the request if
    // our connection coalescing requirements are met. See also:
    // https://hpbn.co/optimizing-application-delivery/
    // https://daniel.haxx.se/blog/2016/08/18/http2-connection-coalescing/

    // 1. This connection must be HTTP/2.
    if (http2Connection == null) return false;

    // 2. The routes must share an IP address. This requires us to have a DNS address for both
    // hosts, which only happens after route planning. We can't coalesce connections that use a
    // proxy, since proxies don't tell us the origin server's IP address.
    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;

    // 3. This connection's server certificate's must cover the new host.
    if (route.address().hostnameVerifier() != OkHostnameVerifier.INSTANCE) return false;
    if (!supportsUrl(address.url())) return false;

    // 4. Certificate pinning must match the host.
    try {
      address.certificatePinner().check(address.url().host(), handshake().peerCertificates());
    } catch (SSLPeerUnverifiedException e) {
      return false;
    }

    return true; // The caller's address can be carried by this connection.
  }

• 連接沒有達到共享上限
• 非host域必須完全一樣
• 如果此時host域也相同，則符合條件，可以被複用
• 如果host不相同，在HTTP/2的域名切片場景下一樣可以複用，具體細節可以參考：https://hpbn.co/optimizing-application-delivery/

2.2.3 deduplicate

deduplicate方法主要是針對在HTTP/2場景下多個多路複用連接清除的場景。如果當前連接是HTTP/2，那麼所有指向該站點的請求都應該基於同一個TCP連接：

[ConnectionPool.java]
  /**
   * Replaces the connection held by {@code streamAllocation} with a shared connection if possible.
   * This recovers when multiple multiplexed connections are created concurrently.
   */
  Socket deduplicate(Address address, StreamAllocation streamAllocation) {
    assert (Thread.holdsLock(this));
    for (RealConnection connection : connections) {
      if (connection.isEligible(address, null)
          && connection.isMultiplexed()
          && connection != streamAllocation.connection()) {
        return streamAllocation.releaseAndAcquire(connection);
      }
    }
    return null;
  }

put和evictAll比較簡單，在這裏就不寫了，大家自行看源碼。

2.3 自動回收

連接池中有socket回收，而這個回收是以RealConnection的弱引用List<Reference<StreamAllocation>>是否為0來為依據的。ConnectionPool有一個獨立的線程cleanupRunnable來清理連接池，其觸發時機有兩個：

• 當連接池中put新的連接時
• 當connectionBecameIdle接口被調用時

其代碼如下：

while (true) {
  //執行清理並返回下場需要清理的時間
  long waitNanos = cleanup(System.nanoTime());
  if (waitNanos == -1) return;
  if (waitNanos > 0) {
    synchronized (ConnectionPool.this) {
      try {
        //在timeout內釋放鎖與時間片
        ConnectionPool.this.wait(TimeUnit.NANOSECONDS.toMillis(waitNanos));
      } catch (InterruptedException ignored) {
      }
    }
  }
}

這段死循環實際上是一個阻塞的清理任務，首先進行清理(clean)，並返回下次需要清理的間隔時間，然後調用wait(timeout)進行等待以釋放鎖與時間片，當等待時間到了後，再次進行清理，並返回下次要清理的間隔時間...

接下來看下cleanup函數:

[ConnectionPool.java]
long cleanup(long now) {
  int inUseConnectionCount = 0;
  int idleConnectionCount = 0;
  RealConnection longestIdleConnection = null;
  long longestIdleDurationNs = Long.MIN_VALUE;

  //遍曆`Deque`中所有的`RealConnection`，標記泄漏的連接
  synchronized (this) {
    for (RealConnection connection : connections) {
      // 查詢此連接內部StreamAllocation的引用數量
      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) {
      //如果(`空閑socket連接超過5個`
      //且`keepalive時間大於5分鍾`)
      //就將此泄漏連接從`Deque`中移除
      connections.remove(longestIdleConnection);
    } else if (idleConnectionCount > 0) {
      //返回此連接即將到期的時間，供下次清理
      //這裏依據是在上文`connectionBecameIdle`中設定的計時
      return keepAliveDurationNs - longestIdleDurationNs;
    } else if (inUseConnectionCount > 0) {
      //全部都是活躍的連接，5分鍾後再次清理
      return keepAliveDurationNs;
    } else {
      //沒有任何連接，跳出循環
      cleanupRunning = false;
      return -1;
    }
  }

  //關閉連接，返回`0`，也就是立刻再次清理
  closeQuietly(longestIdleConnection.socket());
  return 0;
}

其基本邏輯如下：

• 遍曆連接池中所有連接，標記泄露連接
• 如果被標記的連接滿足(空閑socket連接超過5個&&keepalive時間大於5分鍾)，就將此連接從Deque中移除，並關閉連接，返回0，也就是將要執行wait(0)，提醒立刻再次掃描
• 如果(目前還可以塞得下5個連接，但是有可能泄漏的連接(即空閑時間即將達到5分鍾))，就返回此連接即將到期的剩餘時間，供下次清理
• 如果(全部都是活躍的連接)，就返回默認的keep-alive時間，也就是5分鍾後再執行清理

而pruneAndGetAllocationCount負責標記並找到不活躍連接:

[ConnnecitonPool.java]
//類似於引用計數法，如果引用全部為空，返回立刻清理
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);
    //如果正在被使用，跳過，接著循環
    //是否置空是在上文`connectionBecameIdle`的`release`控製的
    if (reference.get() != null) {
      //非常明顯的引用計數
      i++;
      continue;
    }

    //否則移除引用
    references.remove(i);
    connection.noNewStreams = true;

    //如果所有分配的流均沒了，標記為已經距離現在空閑了5分鍾
    if (references.isEmpty()) {
      connection.idleAtNanos = now - keepAliveDurationNs;
      return 0;
    }
  }

  return references.size();
}

OkHttp的連接池通過計數+標記清理的機製來管理連接池，使得無用連接可以被會回收，並保持多個健康的keep-alive連接。這也是OkHttp的連接池能保持高效的關鍵原因。

最後更新：2017-05-05 10:31:20

  上一篇： OkHttp 3.7源碼分析（四）——緩存策略

  下一篇： 震驚！你不知道的淘寶首頁發展史