在Android世界中前前后后出現(xiàn)過很多網(wǎng)絡(luò)框架，比如：HttpUrlConnection、HttpClient、Volley、AsyncHttpClient、okhttp、Retrofit等。其中目前比較流行的當(dāng)屬okhttp與Retrofit莫屬了，其中Retrofit是基于okhttp的基礎(chǔ)上進行的進一步的封裝得到的，所以對于知其然還要知其所以然的monkey來說了解okhttp的源碼是很必要的，所以下面請跟我一起來看看okhttp的源碼吧。

本篇主要三個部分，分別是：OKHTTP流程分析、連接池實現(xiàn)、Dispatcher詳解。

一、OKHTTP流程分析

想要分析源碼我們需要一個切入口進行跟進，最好的入口不外乎日常使用的流程。我們對okhttp的一般使用方法：

//同步
    OkHttpClient client = new OkHttpClient();

    Request request = new Request.Builder()
            .url(url)
            .build();
    Response response = client.newCall(request).execute();
    if (response.isSuccessful()) {
        //...
    } else {
        //...
    }

//異步
    OkHttpClient client = new OkHttpClient();

    Request request = new Request.Builder()
            .url(url)
            .build();

    client.newCall(request).enqueue(new Callback() {
        @Override public void onFailure(Call call, IOException e) {
           //...
        }

        @Override public void onResponse(Call call, Response response) throws IOException {
            //...
        }
    });

由上可知，先初始化一個OkHttpClient，然后調(diào)用其newCall()函數(shù)返回一個call類。看一下其實現(xiàn)：

  @Override 
  public Call newCall(Request request) {
    return RealCall.newRealCall(this, request, false /* for web socket */);
  }

直接返回了RealCall的newRealCall()，其中RealCall是call接口的實現(xiàn)類。

static RealCall newRealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
    // Safely publish the Call instance to the EventListener.
    RealCall call = new RealCall(client, originalRequest, forWebSocket);
    call.eventListener = client.eventListenerFactory().create(call);
    return call;
  }

通過newCall()最終返回了一個RealCall的實例。之后同步調(diào)用RealCall的 execute()，異步調(diào)用enqueue，我們先來看同步：

  @Override 
  public Response execute() throws IOException {
    try {
      Response result = getResponseWithInterceptorChain();//執(zhí)行request請求
      return result;
    } catch (IOException e) {
      //...
    } finally {
      client.dispatcher().finished(this);//切換下一request執(zhí)行
    }
  }

簡化了以下只剩下最簡單的代碼，可以看到直接調(diào)用了getResponseWithInterceptorChain()

再看異步：

@Override 
  public void enqueue(Callback responseCallback) {
    client.dispatcher().enqueue(new AsyncCall(responseCallback));
  }

  synchronized void enqueue(AsyncCall call) {
    if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
      runningAsyncCalls.add(call);//放到正在執(zhí)行隊列
      executorService().execute(call);//執(zhí)行
    } else {
      readyAsyncCalls.add(call);//放到等待隊列
    }
  }

可以看到生成了一個AsyncCall并在executorService().execute(call);進行了執(zhí)行，看下AsyncCall：

  final class AsyncCall extends NamedRunnable {

    AsyncCall(Callback responseCallback) { }

    @Override 
    protected void execute() {
      try {
        Response response = getResponseWithInterceptorChain();
        //...
      } catch (IOException e) {
        //...
      } finally {
        client.dispatcher().finished(this);
      }
    }
  }

和同步的execute非常相似，都最終調(diào)用了getResponseWithInterceptorChain()，其實同步和異步的區(qū)別就是一個直接執(zhí)行了，一個使用了線程池，具體實現(xiàn)值得學(xué)習(xí)一下，感興趣可以看下源碼，不在贅述。

下面就主要看getResponseWithInterceptorChain()的實現(xiàn)了：

  Response getResponseWithInterceptorChain() throws IOException {
    // Build a full stack of interceptors.
    List<Interceptor> interceptors = new ArrayList<>();
    interceptors.addAll(client.interceptors());
    interceptors.add(retryAndFollowUpInterceptor);
    interceptors.add(new BridgeInterceptor(client.cookieJar()));
    interceptors.add(new CacheInterceptor(client.internalCache()));
    interceptors.add(new ConnectInterceptor(client));
    if (!forWebSocket) {
      interceptors.addAll(client.networkInterceptors());
    }
    interceptors.add(new CallServerInterceptor(forWebSocket));
    Interceptor.Chain chain = new RealInterceptorChain(interceptors, null, null, null, 0,
        originalRequest, this, eventListener, client.connectTimeoutMillis(),
        client.readTimeoutMillis(), client.writeTimeoutMillis());

    return chain.proceed(originalRequest);
  }

其中就做了兩件事：1)創(chuàng)建了一些interceptors；2)新建了RealInterceptorChain并調(diào)用了它的proceed的方法。我們直接看該proceed做了什么

@Override 
  public Response proceed(Request request) throws IOException {
    return proceed(request, streamAllocation, httpCodec, connection);
  }

  public Response proceed(Request request, StreamAllocation streamAllocation, HttpCodec httpCodec,
      RealConnection connection) throws IOException {

    // Call the next interceptor in the chain.
    RealInterceptorChain next = new RealInterceptorChain(interceptors, streamAllocation, httpCodec,
        connection, index + 1, request, call, eventListener, connectTimeout, readTimeout, writeTimeout);
    
    Interceptor interceptor = interceptors.get(index);
    Response response = interceptor.intercept(next);
    
    return response;
  }

可知，其中又新建了只有index + 1不同的RealInterceptorChain并執(zhí)行了上面的interceptors鏈表中某一個interceptor的intercept(next)，我們看下intercept(next)在干嘛(以retryAndFollowUpInterceptor為例)

  @Override 
  public Response intercept(Chain chain) throws IOException {

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

    while (true) {//如果結(jié)果正常則返回，否則使用更新的request進行第二次處理。
      Response response = realChain.proceed(request, streamAllocation, null, null);//直接傳遞給下一個
      -----------------------------------------------------------------------------------------
      Request followUp = followUpRequest(response, streamAllocation.route());//處理重定向等邏輯.

      if (followUp == null) {
        return response;
      }
      //建立新的StreamAllocation，以供followUp 使用。
      streamAllocation = new StreamAllocation(client.connectionPool(),
            createAddress(followUp.url()), call, eventListener, callStackTrace);

      request = followUp;
    }
  }

可見，主要做了三件事：
1）新建StreamAllocation ；用于建立網(wǎng)絡(luò)連接
2）執(zhí)行上面?zhèn)鬟f過來的next.proceed()執(zhí)行下一個interceptor的intercept(),直到interceptors全部被執(zhí)行完。
3）調(diào)用followUpRequest，進行檢查response中是否含有重定向，沒有返回null，有返回新的request。
其中while死循環(huán)的目的就是持續(xù)檢查返回結(jié)果中是否有重定向，直到?jīng)]有在跳出。
注意：其中的分割線，分割線以上都是用來處理request邏輯的，分割線以下都是用來處理response邏輯的，因為realChain.proceed會持續(xù)循環(huán)調(diào)用，直到返回結(jié)果，調(diào)用鏈如下圖。

image.png

總體流程圖：

image.png

其中，總體的調(diào)用流程就是上面的部分，下面介紹下幾個攔截器的作用，分析方法和上面retryAndFollowUpInterceptor一樣，只要看intercept方法就可以了。

retryAndFollowUpInterceptor();//主要負責(zé)重定向攔截相關(guān)
BridgeInterceptor(client.cookieJar());//主要負責(zé)請求相應(yīng)頭添加去除等邏輯
CacheInterceptor(client.internalCache());//主要負責(zé)緩存相關(guān)，使用了diskLruCache()
ConnectInterceptor(client);//主要負責(zé)網(wǎng)絡(luò)連接相關(guān)
CallServerInterceptor(forWebSocket);//最后一個攔截器，負責(zé)與服務(wù)器建立 Socket 連接
client.interceptors();//用戶自定
client.networkInterceptors();//用戶自定義，用戶可以在攔截器的一頭一尾進行自定義功能的數(shù)據(jù)處理，
//并在client初始化時傳入進去即可。

如果只是看流程的到這里就可以結(jié)束了。

二、連接池實現(xiàn)

下面我們看一下連接池ConnectionPool實現(xiàn)，我們都直到網(wǎng)絡(luò)連接是基于TCP/IP基礎(chǔ)的，所以必須要經(jīng)歷三次握手與四次揮手，頻繁的建立連接與斷開連接是很耗時的，所以就建立連接池來實現(xiàn)對連接的最大復(fù)用。

ConnectionPool內(nèi)部維持了一個ArrayDeque來保存連接。

private final Deque<RealConnection> connections = new ArrayDeque<>();

  public ConnectionPool() {
    this(5, 5, TimeUnit.MINUTES);
  }

可維持的最大數(shù)量默認5個，維持時間5分鐘，會創(chuàng)建一個線程定時進行查詢處理超時的鏈接。雖然ConnectionPool在client初始化時就傳入了進來，但是直到ConnectInterceptor時才會調(diào)用進行查找，最終會調(diào)用其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;
  }

其中就是遍歷所有鏈接connections對它進行檢查是否可用，如果符合就返回，Connection代表真實的socket物理連接,如下圖

image.png

下面看下isEligible做了什么：

  /**
   * 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, @Nullable 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.
    }
    //...
    return true; // The caller's address can be carried by this connection.
  }

可見，先是檢查是否超過每個連接所能鏈接的數(shù)量，默認是:1，然后檢查代理、DNS、目標(biāo)主機等是否相同，如果都相同就說明該鏈接可用，否則不可用。

接著看一下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保存到streamAllocation中，并將該streamAllocation添加到connection.allocations鏈表中

public final List<Reference<StreamAllocation>> allocations = new ArrayList<>();

從中我們就可以答題理解connection與StreamAllocation的關(guān)系了：StreamAllocation是一個負責(zé)查找可用連接，完成鏈接回調(diào)的類，它是物理連接connection的邏輯代表，直到ConnectInterceptor調(diào)用時兩者合二為一。

三、Dispatcher詳解

在第一部分有講過同步和異步的處理，其實他們都是通過Dispatcher分發(fā)實現(xiàn)的。了解一下Dispatcher中幾個重要的變量：

  /** 異步等待隊列 */
  private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();

  /** 異步執(zhí)行隊列 */
  private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();

  /** 同步執(zhí)行隊列 */
  private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

Dispatcher就是對這三個變量的維護來實現(xiàn)對request的管理。

  private int maxRequests = 64;
  private int maxRequestsPerHost = 5;

maxRequests 用來限制異步請求runningAsyncCalls 的最大長度，默認為64個；maxRequestsPerHost用來限制runningAsyncCalls 中相同host的最大請求數(shù)量，默認為5個；超過以上兩個限制的request都將放入readyAsyncCalls 隊列中。

同步請求在

Response response = client.newCall(request).execute();

后直接調(diào)用

client.dispatcher().executed(this);//入列

  synchronized void executed(RealCall call) {
    runningSyncCalls.add(call);
  }

加入了runningSyncCalls隊列，等到執(zhí)行結(jié)束后回調(diào)finished方法

client.dispatcher().finished(this);

  void finished(RealCall call) {
    finished(runningSyncCalls, call, false);
  }

異步請求在

 client.newCall(request).enqueue(new Callback())

執(zhí)行結(jié)束后回調(diào)

client.dispatcher().finished(this);

void finished(AsyncCall call) {
    finished(runningAsyncCalls, call, true);
  }

是的，它們最后執(zhí)行的finished是同一個泛型函數(shù)，只是最后一個參數(shù)不同而已。

  private <T> void finished(Deque<T> calls, T call, boolean promoteCalls) {
    synchronized (this) {
    //在runningSyncCalls隊列中將該請求移除
      if (!calls.remove(call)) throw new AssertionError("Call wasn't in-flight!");
      if (promoteCalls) promoteCalls();
    }
  }

可見，同步移除之后直接退出了，異步執(zhí)行了promoteCalls：

  private void promoteCalls() {
    if (runningAsyncCalls.size() >= maxRequests) return; // Already running max capacity.
    if (readyAsyncCalls.isEmpty()) return; // No ready calls to promote.

    for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
      AsyncCall call = i.next();

      if (runningCallsForHost(call) < maxRequestsPerHost) {
        i.remove();
        runningAsyncCalls.add(call);
        executorService().execute(call);
      }

      if (runningAsyncCalls.size() >= maxRequests) return; // Reached max capacity.
    }
  }

可見，直接取出等待隊列中的一個request執(zhí)行。