OKHTTP3源碼2-連接池管理

qqiabc521 6年前發布 | 45K 次閱讀 OkHttp 網絡技術

在《OKHTTP3源碼和設計模式-1》，中整體介紹了 OKHttp3 的源碼架構，重點講解了請求任務的分發管理和線程池以及請求執行過程中的攔截器。這一章我們接著往下走認識一下 OKHttp3 底層連接和連接池工作機制。

RealCall 封裝了請求過程，組織了用戶和內置攔截器，其中內置攔截器 retryAndFollowUpInterceptor -> BridgeInterceptor -> CacheInterceptor 完執行層的大部分邏輯，ConnectInterceptor -> CallServerInterceptor 兩個攔截器開始邁向連接層最終完成網絡請求。

連接層連接器

ConnectInterceptor 的工作很簡單，負責打開連接; CallServerIntercerceptor 是核心連接器鏈上的最后一個連接器，

負責從當前連接中寫入和讀取數據。

連接的打開

/** Opens a connection to the target server and proceeds to the next interceptor. */
        // 打開一個和目標服務器的連接，并把處理交個下一個攔截器
  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();

    // We need the network to satisfy this request. Possibly for validating a conditional GET.
    boolean doExtensiveHealthChecks = !request.method().equals("GET");
            // 打開連接
    HttpCodec httpCodec = streamAllocation.newStream(client, doExtensiveHealthChecks);
    RealConnection connection = streamAllocation.connection();
     // 交個下一個攔截器
   return realChain.proceed(request, streamAllocation, httpCodec, connection);
  }
}

單獨看 ConnectInterceptor 的代碼很簡單，不過連接正在打開的過程需要看看 streamAllocation.newStream(client, doExtensiveHealthChecks)，內部執行過程。還是先整體上了看看 StreamAllocation 這個類的作用。

StreamAllocation

StreamAllocation 處于上層請求和底層連接池直接，協調請求和連接池直接的關系。先來看看 StreamAllocation 對象在哪里創建的? 回到之前文章中介紹的 RetryAndFollowUpInterceptor，這是核心攔截器鏈上的頂層攔截器其中源碼：

@Override 
    public Response intercept(Chain chain) throws IOException {
             Request request = chain.request();
       streamAllocation = new StreamAllocation(
       client.connectionPool(), createAddress(request.url()), callStackTrace);
      ...省略代碼
  }

這里，每一次請求創建了一個 StreamAllocation 對象，那么問題來了？之前我們說過每一個 OkHttpClient 對象只有一個對應的連接池，剛剛又說到 StreamAllocation 打開連接，那么 StreamAllocation 是如何創建連接池的呢？我們很容易就去 StreamAllocation 中找連接池創建的邏輯，但是找不到。連接池創建的地方在 OkHttpClient 中：

public Builder() {
      dispatcher = new Dispatcher();
      protocols = DEFAULT_PROTOCOLS;
      connectionSpecs = DEFAULT_CONNECTION_SPECS;
      eventListenerFactory = EventListener.factory(EventListener.NONE);
      proxySelector = ProxySelector.getDefault();
      cookieJar = CookieJar.NO_COOKIES;
      socketFactory = SocketFactory.getDefault();
      hostnameVerifier = OkHostnameVerifier.INSTANCE;
      certificatePinner = CertificatePinner.DEFAULT;
      proxyAuthenticator = Authenticator.NONE;
      authenticator = Authenticator.NONE;
                // 創建連接池
      connectionPool = new ConnectionPool();
      dns = Dns.SYSTEM;
      followSslRedirects = true;
      followRedirects = true;
      retryOnConnectionFailure = true;
      connectTimeout = 10_000;
      readTimeout = 10_000;
      writeTimeout = 10_000;
      pingInterval = 0;
}

OkHttpClient 默認構造函數的 Builder ，在這里創建了連接池。所以這里我們也可以看到，如果我們對默認連接池不滿，我們是可以直通過 builder 接指定的。

搞懂了 StreamAllocation 和 ConnectionPool 的創建，我們再來看看 StreamAllocation 是怎么打開連接的？直接兜源碼可能有點繞，先給一個粗略流程圖，然后逐點分析。

鏈接池實現

相信大家都有一些 Http 協議的基礎（如果沒有就去補了，不然看不懂）都知道 Http 的下層協議是 TCP。TCP 連接的創建和斷開是有性能開銷的，在 Http1.0 中，每一次請求就打開一個連接，在一些老的舊的瀏覽器上，如果還是基于 Http1.0，體驗會非常差; Http1.1 以后支持長連接，運行一個請求打開連接完成請求后，連接可以不關閉，下次請求時復用此連接，從而提高連接的利用率。當然并不是連接打開后一直開著不關，這樣又會造成連接浪費，怎么管理？

在OKHttp3 的默認實現中，使用一個雙向隊列來緩存所有連接，這些連接中最多只能存在 5 個空閑連接，空閑連接最多只能存活 5 分鐘。

定期清理實現

public final class ConnectionPool {
    /**
    * Background threads are used to cleanup expired connections. There will be at most a single
    * thread running per connection pool. The thread pool executor permits the pool itself to be
    * garbage collected.
   */
        // 后臺定期清理連接的線程池
       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() {
      while (true) {
                   // cleanup 執行清理
        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) {
            }
          }
        }
      }
    }
  };

雙向隊列

// 存儲連接的雙向隊列
private final Deque<RealConnection> connections = new ArrayDeque<>();

放入連接

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

獲取連接

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

StreamAllocation.連接創建和復用

ConnectionPool 的源碼邏輯還是相當比較簡單，主要提供一個雙向列表來存取連接，使用一個定時任務定期清理無用連接。二連接的創建和復用邏輯主要在 StreamAllocation 中。

尋找連接

private RealConnection findHealthyConnection(int connectTimeout, int readTimeout,
  int writeTimeout, boolean connectionRetryEnabled, boolean doExtensiveHealthChecks)
  throws IOException {
while (true) {
       //  核心邏輯在 findConnection（）中
  RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout,
      connectionRetryEnabled);

  // If this is a brand new connection, we can skip the extensive health checks.
  synchronized (connectionPool) {
    if (candidate.successCount == 0) {
      return candidate;
    }
  }

  // Do a (potentially slow) check to confirm that the pooled connection is still good. If it
  // isn't, take it out of the pool and start again.
  if (!candidate.isHealthy(doExtensiveHealthChecks)) {
    noNewStreams();
    continue;
  }

  return candidate;
}
}

findConnection（）：

private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout,
  boolean connectionRetryEnabled) throws IOException {
Route selectedRoute;
synchronized (connectionPool) {
  // 省略部分代碼...
  // Attempt to get a connection from the pool. Internal.instance 就是 ConnectionPool 的實例
  Internal.instance.get(connectionPool, address, this, null);
  if (connection != null) {
           // 復用此連接
    return connection;
  }
          // 省略部分代碼...
            // 創建新新連接
  result = new RealConnection(connectionPool, selectedRoute);
        // 引用計數
  acquire(result);
}

synchronized (connectionPool) {
  // Pool the connection. 放入連接池
  Internal.instance.put(connectionPool, result);
 }
  // 省略部分代碼...
 return result;
}

StreamAllocation 主要是為上層提供一個連接，如果連接池中有復用的連接則復用連接，如果沒有則創建新的。無論是拿到可復用的還是創建新的，都要為此連接計算一下引用計數。

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

 this.connection = connection;
     //  連接使用allocations列表來記錄每一個引用
 connection.allocations.add(new StreamAllocationReference(this, callStackTrace));
}

Realconnection

Realconnection 封裝了底層 socket 連接，同時使用 OKio 來進行數據讀寫， OKio 是 square 公司的另一個獨立的開源項目，大家感興趣可以去深入讀下 OKio 源碼，這里不展開。

/** Does all the work necessary to build a full HTTP or HTTPS connection on a raw socket. */
  private void connectSocket(int connectTimeout, int readTimeout) throws IOException {
    Proxy proxy = route.proxy();
    Address address = route.address();

    rawSocket = proxy.type() == Proxy.Type.DIRECT || proxy.type() == Proxy.Type.HTTP
    ? address.socketFactory().createSocket()
    : new Socket(proxy);

    rawSocket.setSoTimeout(readTimeout);
    try {
              // 打開 socket 連接
      Platform.get().connectSocket(rawSocket, route.socketAddress(), connectTimeout);
    } catch (ConnectException e) {
      ConnectException ce = new ConnectException("Failed to connect to " + route.socketAddress());
      ce.initCause(e);
      throw ce;
    }
           // 使用 OKil 連上 socket 后續讀寫使用 Okio
    source = Okio.buffer(Okio.source(rawSocket));
    sink = Okio.buffer(Okio.sink(rawSocket));
 }

來自：http://www.liuguangli.win/archives/778

本文由用戶 qqiabc521 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。

轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。

本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！

本文地址：http://www.baiduhome.net/lib/view/open1523149630356.html

OkHttp 網絡技術