OkHttp 源碼解析（二）：創建鏈接

簡介

上一篇文章（OkHttp源碼解析（一）：基本流程）介紹了 OkHttp 的基本流程，包括 Request 的建立、Dispatcher 對 Request 的調度以及 Interceptor 的使用。OkHttp 中默認會添加 RetryAndFollowUpInterceptor、BridgeInterceptor、CacheInterceptor、ConnectInterceptor 以及 CallServerInterceptor 這幾個攔截器。本文主要看一下 RetryAndFollupInterceptor 並引出創建鏈接相關的分析。

RetryAndFollowUpInterceptor

Interceptor 最主要的代碼都在 intercept 中，下面是 RetryAndFollowUpInterceptor#intercept 中的部分代碼：

@Override public Response intercept(Chain chain) throws IOException {
    Request request = chain.request();

    streamAllocation = new StreamAllocation(
        client.connectionPool(), createAddress(request.url()), callStackTrace);  // 1

    int followUpCount = 0;
    Response priorResponse = null;
    while (true) {
      if (canceled) {
        streamAllocation.release();  // 2
        throw new IOException("Canceled");
      }
    ...
    response = ((RealInterceptorChain) chain).proceed(request, streamAllocation, null, null); // 3
    ...
}

上面註釋 1 處建立了一個 StreamAllocation 對象，註釋 2 處 調用了其 release 方法，註釋 3 處則把這個對象傳給了下一個 Interceptor。StreamAlloction 這個類很重要，下面就看一下它的用途。

StreamAlloction

StreamAllocation 從名字上看是流分配器，其實它是統籌管理了幾樣東西，註釋寫的很是清楚：

/**
 * This class coordinates the relationship between three entities:
 *
 * <ul>
 *     <li><strong>Connections:</strong> physical socket connections to remote servers. These are
 *         potentially slow to establish so it is necessary to be able to cancel a connection
 *         currently being connected.
 *     <li><strong>Streams:</strong> logical HTTP request/response pairs that are layered on
 *         connections. Each connection has its own allocation limit, which defines how many
 *         concurrent streams that connection can carry. HTTP/1.x connections can carry 1 stream
 *         at a time, HTTP/2 typically carry multiple.
 *     <li><strong>Calls:</strong> a logical sequence of streams, typically an initial request and
 *         its follow up requests. We prefer to keep all streams of a single call on the same
 *         connection for better behavior and locality.
 * </ul>

簡單來講， StreamAllocation 協調了 3 樣東西：

• Connections ： 物理的 socket 鏈接
• Streams：邏輯上的 HTTP request/response 對。每一個 Connection 有個變量 allocationLimit ，用於定義能夠承載的併發的 streams 的數量。HTTP/1.x 的 Connection 一次只能有一個 stream， HTTP/2 通常能夠有多個。
• Calls ： Streams 的序列。一個初始的 request 可能還會有後續的 request（如重定向）。OkHttp 傾向於讓一個 call 全部的 streams 運行在同一個 connection 上。

StreamAllocation 提供了一些 API 來釋放以上的資源對象。 在 RetryAndFollowUpInterceptor 中建立的 StreamAllocation 對象下一個用到的地方是 ConnectInterceptor，其 intercept 代碼以下：

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

在上面的代碼中， streamAllocation 建立了 httpCodec 以及 connection 對象。 httpCodec 便是上面所說的 Streams，而 connection 則是上面的 Connection，Connection 是一個接口，它的惟一實現類是 RealConnection。

newStream

StreamAllocation 中的 newStream 方法用於尋找新的 RealConnection 以及 HttpCodec，代碼以下：

public HttpCodec newStream(OkHttpClient client, boolean doExtensiveHealthChecks) {
    int connectTimeout = client.connectTimeoutMillis();
    int readTimeout = client.readTimeoutMillis();
    int writeTimeout = client.writeTimeoutMillis();
    boolean connectionRetryEnabled = client.retryOnConnectionFailure();

    try {
      RealConnection resultConnection = findHealthyConnection(connectTimeout, readTimeout,
          writeTimeout, connectionRetryEnabled, doExtensiveHealthChecks);
      HttpCodec resultCodec = resultConnection.newCodec(client, this);

      synchronized (connectionPool) {
        codec = resultCodec;
        return resultCodec;
      }
    } catch (IOException e) {
      throw new RouteException(e);
    }
  }

在 newStream 中，經過 findHealthyConnection 找到可用的 Connection ，並用這個 Connection 生成一個 HttpCodec 對象。 findHealthyConnection 是找到一個健康的鏈接，代碼以下：

private RealConnection findHealthyConnection(int connectTimeout, int readTimeout,
      int writeTimeout, boolean connectionRetryEnabled, boolean doExtensiveHealthChecks)
      throws IOException {
    while (true) {
      RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout,
          connectionRetryEnabled);

      // If this is a brand new connection, we can skip the extensive health checks.
      synchronized (connectionPool) {
      // successCount == 0 表示還未使用過，則可使用
        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;
    }
 }

public boolean isHealthy(boolean doExtensiveChecks) {
    if (socket.isClosed() || socket.isInputShutdown() || socket.isOutputShutdown()) {
      return false;
    }
    ... // 省略 Http2 代碼
    return true;
  }

在一個無限循環中，經過 findConnection 尋找一個 connection，並判斷是否可用，首先若是沒有使用過的確定是健康的可直接返回，不然調用 isHealthy，主要就是判斷 socket 是否關閉。這裏的 socket 是在 findConnection 中賦值的，再看看 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");

      // Attempt to use an already-allocated connection.
      RealConnection allocatedConnection = this.connection;
      if (allocatedConnection != null && !allocatedConnection.noNewStreams) {
        return allocatedConnection;
      }

      // Attempt to get a connection from the pool.
      // 1. 從 ConnectionPool 取得 connection
      Internal.instance.get(connectionPool, address, this, null);
      if (connection != null) {
        return connection;
      }

      selectedRoute = route;
    }

    // If we need a route, make one. This is a blocking operation.
    if (selectedRoute == null) {
      selectedRoute = routeSelector.next();
    }

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

      // Now that we have an IP address, make another attempt at getting a connection from the pool.
      // 2. 有了 ip 地址後再從 connectionpool中取一次
      // This could match due to connection coalescing.
      Internal.instance.get(connectionPool, address, this, selectedRoute);
      if (connection != null) return connection;

      // Create a connection and assign it to this allocation immediately. This makes it possible
      // for an asynchronous cancel() to interrupt the handshake we're about to do.
      route = selectedRoute;
      refusedStreamCount = 0;
      // 3. ConnectionPool 中沒有，新建立一個
      result = new RealConnection(connectionPool, selectedRoute);
      // 3. 將 StreamAllocation 加入到 `RealConnection` 中的一個隊列中
      acquire(result);
    }

    // Do TCP + TLS handshakes. This is a blocking operation.
    // 4. 創建鏈接，在其中建立 socket
    result.connect(connectTimeout, readTimeout, writeTimeout, connectionRetryEnabled);
    routeDatabase().connected(result.route());

    Socket socket = null;
    synchronized (connectionPool) {
      // Pool the connection.
      // 5. 將新建立的 connection 放到 ConnectionPool 中 
      Internal.instance.put(connectionPool, result);

      // If another multiplexed connection to the same address was created concurrently, then
      // release this connection and acquire that one.
      if (result.isMultiplexed()) {
        socket = Internal.instance.deduplicate(connectionPool, address, this);
        result = connection;
      }
    }
    closeQuietly(socket);

    return result;
  }

上面 Connection 的建立大致是如下幾個步驟：

1. 調用 Intenal.get 方法從 ConnectionPool 中獲取一個 Connection，主要根據 url 的 host 判斷，相關代碼在 ConnectionPool 中。
2. 若是沒有而且又獲取了 IP 地址，則再獲取一次。
3. 若是 ConnectionPool 中沒有， 則新建立一個 RealConnection，並調用 acquire 將 StreamAllocation 中加入 RealConnection 中的一個隊列中。
4. 調用 RealConnection#connect 方法創建鏈接，在內部會建立 Socket。
5. 將新建立的 Connection 加入到 ConnectionPool 中。

獲取到了 Connection 以後，再建立一個 HttpCodec 對象。

public HttpCodec newCodec(
      OkHttpClient client, StreamAllocation streamAllocation) throws SocketException {
    if (http2Connection != null) {
      return new Http2Codec(client, streamAllocation, http2Connection);
    } else {
      socket.setSoTimeout(client.readTimeoutMillis());
      source.timeout().timeout(client.readTimeoutMillis(), MILLISECONDS);
      sink.timeout().timeout(client.writeTimeoutMillis(), MILLISECONDS);
      return new Http1Codec(client, streamAllocation, source, sink);
    }
}

根據是 Http1 仍是 Http2 建立對應的 HttpCodec， 其中的 socket 是在 RealConnection 中的 connect 方法建立的。下面具體看看RealConnection。

RealConnection

RealConnection 封裝的是底層的 Socket 鏈接，內部必然有一個 Socket 對象，下面是 RealConnection 內部的變量：

public final class RealConnection extends Http2Connection.Listener implements Connection {
  private static final String NPE_THROW_WITH_NULL = "throw with null exception";
  private final ConnectionPool connectionPool;
  private final Route route;

  // The fields below are initialized by connect() and never reassigned.

  /** The low-level TCP socket. */
  private Socket rawSocket;

  /**
   * The application layer socket. Either an {@link SSLSocket} layered over {@link #rawSocket}, or
   * {@link #rawSocket} itself if this connection does not use SSL.
   */
  private Socket socket;
  private Handshake handshake;
  private Protocol protocol;
  private Http2Connection http2Connection;
  private BufferedSource source;
  private BufferedSink sink;

  // The fields below track connection state and are guarded by connectionPool.

  /** If true, no new streams can be created on this connection. Once true this is always true. */
  public boolean noNewStreams;

  public int successCount;

  /**
   * The maximum number of concurrent streams that can be carried by this connection. If {@code
   * allocations.size() < allocationLimit} then new streams can be created on this connection.
   */
  public int allocationLimit = 1;

  /** Current streams carried by this connection. */
  public final List<Reference<StreamAllocation>> allocations = new ArrayList<>();

  /** Nanotime timestamp when {@code allocations.size()} reached zero. */
  public long idleAtNanos = Long.MAX_VALUE;
  ...
}

• Route 表示的是與服務端創建的路徑，其實內部封裝了 Address，Address 則是封裝了請求的 URL。
• rawSocket 對象表明底層的鏈接，還有一個 socket 是用於 Https， 對於普通的 Http 請求來講，這兩個對象是同樣的。 source 和 sink 則是利用 Okio 封裝 socket 獲得的輸入輸出流。（若是想了解 Okio 的原理，能夠參考我以前的文章：Okio 源碼解析（一）：數據讀取流程）
• noNewStream 對象用於標識這個 Connection 不能再用於 Http 請求了，一旦設置爲 true, 則不會再變。
• allocationLimit 指的是這個 Connection 最多能同時承載幾個 Http 流，對於 Http/1 來講只能是一個。
• allocations 是一個 List 對象，裏面保存着正在使用這個 Connection 的 StreamAllocation 的弱引用，當 StreamAllocation 調用 acquire 時，便會將其弱引用加入這個 List，調用 release 則是移除引用。allocations 爲空說明此 Connection 爲閒置， ConnectionPool 利用這些信息來決定是否關閉這個鏈接。

connect

RealConnection 用於創建鏈接，裏面有相應的 connect 方法：

public void connect(
      int connectTimeout, int readTimeout, int writeTimeout, boolean connectionRetryEnabled) {
    ...
    while (true) {
      try {
        if (route.requiresTunnel()) {
          connectTunnel(connectTimeout, readTimeout, writeTimeout);
        } else {
          // 建立socket，創建鏈接
          connectSocket(connectTimeout, readTimeout);
        }
        // 創建
        establishProtocol(connectionSpecSelector);
        break;
      }
    ...
}

private void connectSocket(int connectTimeout, int readTimeout) throws IOException {
    Proxy proxy = route.proxy();
    Address address = route.address();
    // 建立 socket
    rawSocket = proxy.type() == Proxy.Type.DIRECT || proxy.type() == Proxy.Type.HTTP
        ? address.socketFactory().createSocket()
        : new Socket(proxy);

    rawSocket.setSoTimeout(readTimeout);
    try {
      // 創建鏈接，至關於調用 socket 的 connect 方法
      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;
    }
    
    try {
      // 獲取輸入輸出流
      source = Okio.buffer(Okio.source(rawSocket));
      sink = Okio.buffer(Okio.sink(rawSocket));
    } catch (NullPointerException npe) {
      if (NPE_THROW_WITH_NULL.equals(npe.getMessage())) {
        throw new IOException(npe);
      }
    }
}

若是不是 Https, 則調用 connectSocket，在內部建立 rawSocket 對象，設置超時時間。緊接着 Platform.get().connectSocket 根據不一樣的平臺調用相應的 connect 方法，這樣 rawSocket 就鏈接到服務端了。而後是用 Okio 封裝 rawSocket 的輸入輸出流，這裏的輸入輸出流最終是交給 HttpCodec 進行 Http 報文的寫入都讀取。經過以上步驟，就實現了 Http 請求的鏈接。

總結

本文從 RetryAndFollowupIntercept 中建立 StreamAllocation 對象，到 Connection 中建立 RealConnection 和 HttpCodec，分析了 OkHttp 創建鏈接的基本過程。能夠看出， OkHttp 中的鏈接由
RealConnection 封裝，Http 流的輸入輸出由 HttpCodec 操做，而 StreamAllocation 則統籌管理這些資源。在鏈接的尋找與建立過程，有個關鍵的東西是 ConnectionPool， 即鏈接池。它負責管理全部的 Connection，OkHttp 利用這個鏈接池進行 Connection 的重用以提升網絡請求的效率。本文並無詳細分析 ConnectionPool ，相關內容能夠參見下一篇： OkHttp源碼解析（三）：鏈接池。

若是個人文章對您有幫助，不妨點個贊支持一下(^_^)