天天看點
傳回

開源架構源碼解析系列（1）——進入OkHttp的世界1.請求部分源碼解析2 Dispatcher分析3.攔截器解析4.CallServerInterceptor 攔截器

以前曾經寫過一篇關于Okhttp的使用的文章深入解析OkHttp3，通過這篇文章可以了解OkHttp的各種基本用法，光會使用并不算好漢，我們還要深入了解源碼，學習優秀的設計思想，本篇我就帶大家一起分析源碼，基于Okhttp 3.10.0版本。

1.請求部分源碼解析

1.1 回顧請求的基本用法

1.1.1 發送同步請求

Request request = new Request.Builder().url(url).build();
        try {
            //同步請求
            Call call = mOkHttpClient.newCall(request);
            Response response = call.execute();
            String json = response.body().string();
            Log.d(TAG, json);

        } catch (IOException e) {
            e.printStackTrace();
        }

1.1.2 發送異步請求

//異步請求
        Call call = mOkHttpClient.newCall(request);
        call.enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                Log.d(TAG, "onFailure:" + e.getMessage());
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
                String json = response.body().string();
                Log.d(TAG, json);
            }
        });

1.2 同步請求源碼分析

首先，我們要了解，無論是同步請求還是異步請求，我們都需要先編寫以下代碼：

OkHttpClient mOkHttpClient = new OkHttpClient();
Request request = new Request.Builder().url(url).build();
Call call = mOkHttpClient.newCall(request);

Call是請求的關鍵對象，是通過調用Call的execute方法之後，就會進入請求的邏輯

Response response = call.execute();

1.2.1 OkHttpClient# newCall

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

我們可以看到OkHttpClient# newCall會調用RealCall.newRealCall方法

1.2.2 RealCall # newRealCall

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

檢視一下構造函數

private RealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
    this.client = client;
    this.originalRequest = originalRequest;
    this.forWebSocket = forWebSocket;
    this.retryAndFollowUpInterceptor = new RetryAndFollowUpInterceptor(client, forWebSocket);
  }

RealCall的建立過程中會持有OkHttpClient，請求的 Request還有建立了一個攔截器RetryAndFollowUpInterceptor（這個後面會詳細說明），同時建立eventListener 。

1.2.3 Call# execute()

建立好Call之後，調用execute()方法就開始了請求的流程，Call是一個借口，是以我們要檢視它的實作類RealCall。

1.2.4 RealCall# execute()

@Override public Response execute() throws IOException {
    synchronized (this) {
      if (executed) throw new IllegalStateException("Already Executed");
      executed = true;
    }
    captureCallStackTrace();
    eventListener.callStart(this);
    try {
      client.dispatcher().executed(this);
      Response result = getResponseWithInterceptorChain();
      if (result == null) throw new IOException("Canceled");
      return result;
    } catch (IOException e) {
      eventListener.callFailed(this, e);
      throw e;
    } finally {
      client.dispatcher().finished(this);
    }
  }

這裡重點在于如下兩句

client.dispatcher().executed(this);
 Response result = getResponseWithInterceptorChain();

調用OkHttpClient持有的Dispatcher對象執行call，Dispatcher是非常重要的一環，後面詳細介紹。

1.2.5 Dispatcher# execute()

synchronized void executed(RealCall call) {
    runningSyncCalls.add(call);
  }
           
private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

将同步請求的RealCall 添加到同步的隊列中。

1.2.6 RealCall# getResponseWithInterceptorChain()

通過RealCall# getResponseWithInterceptorChain()方法就可以擷取請求傳回的Response，傳回到調用者

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

這裡主要是一系列攔截器的添加操作，然後調用Interceptor.Chain的proceed方法去執行請求

chain.proceed(originalRequest)

攔截器又是另一個非常重要的環節，後面重點提到。

1.2.7總結同步請求

同步請求邏輯相對簡單，通過Call# execute()最終會調用 RealCall# execute()，然後通過分發器Dispatcher将任務添加到同步隊列中，然後通過一系列攔截器操作後進行請求，最後傳回Response，全程都在主線程中運作，是阻塞式的。

1.3 異步請求源碼分析

1.3.1 Call#enqueue

異步請求，會調用Call#enqueue方法，因為是異步，是以需要傳遞一個Callback回調，Call#enqueue中調用了RealCall#enqueue

1.3.2 RealCall#enqueue

@Override public void enqueue(Callback responseCallback) {
    synchronized (this) {
      if (executed) throw new IllegalStateException("Already Executed");
      executed = true;
    }
    captureCallStackTrace();
    eventListener.callStart(this);
    client.dispatcher().enqueue(new AsyncCall(responseCallback));
  }

RealCall#enqueue中同樣會調用到Dispatcher中，隻是調用enqueue方法，同時new AsyncCall将Callback 包一層.AsyncCall是RealCall的内部類，從中可以擷取RealCall的Request等成員。

1.3.2 Dispatcher#enqueue

synchronized void enqueue(AsyncCall call) {
    if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
      runningAsyncCalls.add(call);
      executorService().execute(call);
    } else {
      readyAsyncCalls.add(call);
    }
  }
           
/** Running asynchronous calls. Includes canceled calls that haven't finished yet. */
  private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();

  /** Running synchronous calls. Includes canceled calls that haven't finished yet. */
  private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

主要做了幾件事：

1.判斷正在執行的異步任務隊列中任務數是否小于maxRequests，且正在執行的任務的host小于 maxRequestsPerHost，這兩個值的大小為：

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

同時滿足條件，則将AsyncCall 添加到runningAsyncCalls隊列中，runningAsyncCalls是異步任務的隊列，否則添加到readyAsyncCalls等待隊列中。

2.調用Dispatcher#executorService方法，擷取Android系統提供的線程池

public synchronized ExecutorService executorService() {
    if (executorService == null) {
      executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
          new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
    }
    return executorService;
  }

**注意：**這裡設定線程池的最大容量為 Integer.MAX_VALUE，但其實受限于maxRequests，是以最多容量也就64而已。

3.通過ExecutorService執行AsyncCall任務，可想而知AsyncCall一定是實作了Runnable接口。

1.3.3 NamedRunnable#run

AsyncCall繼承自NamedRunnable，是以當AsyncCall任務執行時，會執行NamedRunnable#run

@Override public final void run() {
    String oldName = Thread.currentThread().getName();
    Thread.currentThread().setName(name);
    try {
      execute();
    } finally {
      Thread.currentThread().setName(oldName);
    }
  }

主要邏輯在execute()方法

1.3.4 AsyncCall#execute

@Override protected void execute() {
      boolean signalledCallback = false;
      try {
        Response response = getResponseWithInterceptorChain();
        if (retryAndFollowUpInterceptor.isCanceled()) {
          signalledCallback = true;
          responseCallback.onFailure(RealCall.this, new IOException("Canceled"));
        } else {
          signalledCallback = true;
          responseCallback.onResponse(RealCall.this, response);
        }
      } catch (IOException e) {
        if (signalledCallback) {
          // Do not signal the callback twice!
          Platform.get().log(INFO, "Callback failure for " + toLoggableString(), e);
        } else {
          eventListener.callFailed(RealCall.this, e);
          responseCallback.onFailure(RealCall.this, e);
        }
      } finally {
        client.dispatcher().finished(this);
      }
    }

可以看到還是會和同步請求一樣，調用getResponseWithInterceptorChain()執行各種攔截器，傳回Response，無論是同步還是異步最後都會執行Dispatcher#inished()方法，這個後面會提到。

1.3.5 總結異步請求

Call#enqueue會調用到 Dispatcher#enqueue，然後判斷是否符合最大請求數maxRequests（64），最大請求Host數maxRequestsPerHost （5），符合條件的添加到異步任務隊列runningAsyncCalls，通過線程池執行任務，否則添加到等待隊列readyAsyncCalls。

2 Dispatcher分析

前面分析同步和異步請求的時候，都提到Dispatcher，我們這裡重新總結一下：

1.維護了3個隊列，同步請求執行隊列runningSyncCalls，異步請求執行隊列runningAsyncCalls，異步請求等待隊列readyAsyncCalls，3個隊列的添加邏輯前面已經提過。

/** Ready async calls in the order they'll be run. */
  private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();

  /** Running asynchronous calls. Includes canceled calls that haven't finished yet. */
  private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();

  /** Running synchronous calls. Includes canceled calls that haven't finished yet. */
  private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

2.維護了異步請求的線程池，異步執行任務通過線程池進行任務執行

public synchronized ExecutorService executorService() {
    if (executorService == null) {
      executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
          new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
    }
    return executorService;
  }

3.RealCall中請求任務執行完後，進行回收，我們知道Call會被封裝成RealCall，但無論同步還是異步執行完成後，都會調用以下代碼

finally {
        client.dispatcher().finished(this);
      }

2.1 Dispatcher#finished

分為同步和異步的finished，我們先看同步的代碼：

2.1.1 同步finished

/** Used by {@code Call#execute} to signal completion. */
  void finished(RealCall call) {
    finished(runningSyncCalls, call, false);
  }

注意這裡傳入的第三個參數為false

private <T> void finished(Deque<T> calls, T call, boolean promoteCalls) {
    int runningCallsCount;
    Runnable idleCallback;
    synchronized (this) {
      if (!calls.remove(call)) throw new AssertionError("Call wasn't in-flight!");
      if (promoteCalls) promoteCalls();
      runningCallsCount = runningCallsCount();
      idleCallback = this.idleCallback;
    }

    if (runningCallsCount == 0 && idleCallback != null) {
      idleCallback.run();
    }
  }

1.首先将隊列中call進行移除

2.如果promoteCalls為true還會調用 promoteCalls()方法

3.計算runningCallsCount，即為同步和異步執行隊列的size總和

public synchronized int runningCallsCount() {
    return runningAsyncCalls.size() + runningSyncCalls.size();
  }

4.當runningCallsCount為0時說明已經沒有任務了，進行回調

if (runningCallsCount == 0 && idleCallback != null) {
      idleCallback.run();
    }
           
void finished(AsyncCall call) {
    finished(runningAsyncCalls, call, true);
  }

2.1.2 異步finished

異步代碼如下：

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

和同步的差別是，傳入的promoteCalls為true，是以當執行finished時會比同步多執行一個promoteCalls()方法

Dispatcher#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.
    }
  }

邏輯很清晰，就是當異步執行隊列readyAsyncCalls有空閑位置時，周遊等待隊列，将readyAsyncCalls的任務取出加入readyAsyncCalls，然後線程池對任務進行執行。

3.攔截器解析

3.1 攔截器執行順序

在之前分析同步和異步任務的時候，分析過getResponseWithInterceptorChain()方法執行後就會傳回請求結果Response

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.首先将調用者自定義的攔截器都放入interceptors集合的最前面，然後是分别添加okhttp中必須的幾個攔截器，後面我們會一一分析

2.建立攔截器的鍊RealInterceptorChain,将interceptors傳入

從上述代碼可以看出攔截器調用的先後順序依次是

client.interceptors()–>RetryAndFollowUpInterceptor–>BridgeInterceptor–>CacheInterceptor–>ConnectInterceptor–>client.networkInterceptors()–>CallServerInterceptor

這裡使用到了非常經典的設計模式，就是責任鍊模式，reques自上而下下傳遞執行，然後Response至下而上傳回

開源架構源碼解析系列（1）——進入OkHttp的世界1.請求部分源碼解析2 Dispatcher分析3.攔截器解析4.CallServerInterceptor 攔截器

這裡3個參數是為 null的

3.2 RealInterceptorChain#proceed 

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

關鍵部分代碼如下：

// 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);

又建立了一個RealInterceptorChain，然後擷取interceptors中第一位的攔截器開始執行，這裡index=0。然後就會按照順序執行各攔截器。

3.3 RetryAndFollowUpInterceptor

如果沒有自定義攔截器情況下，首先會走到RetryAndFollowUpInterceptor的intercept方法

@Override public Response intercept(Chain chain) throws IOException {
    Request request = chain.request();
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    Call call = realChain.call();
    EventListener eventListener = realChain.eventListener();

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

    int followUpCount = 0;
    Response priorResponse = null;
    while (true) {
      if (canceled) {
        streamAllocation.release();
        throw new IOException("Canceled");
      }

      Response response;
      boolean releaseConnection = true;
      try {
        response = realChain.proceed(request, streamAllocation, null, null);
        releaseConnection = false;
      } catch (RouteException e) {
       	//省略
      } catch (IOException e) {
      	//省略
      } finally {
        // We're throwing an unchecked exception. Release any resources.
        if (releaseConnection) {
          streamAllocation.streamFailed(null);
          streamAllocation.release();
        }
      }

      // Attach the prior response if it exists. Such responses never have a body.
      if (priorResponse != null) {
        response = response.newBuilder()
            .priorResponse(priorResponse.newBuilder()
                    .body(null)
                    .build())
            .build();
      }

      Request followUp = followUpRequest(response, streamAllocation.route());

      if (followUp == null) {
        if (!forWebSocket) {
          streamAllocation.release();
        }
        return response;
      }

      closeQuietly(response.body());

      if (++followUpCount > MAX_FOLLOW_UPS) {
        streamAllocation.release();
        throw new ProtocolException("Too many follow-up requests: " + followUpCount);
      }

      if (followUp.body() instanceof UnrepeatableRequestBody) {
        streamAllocation.release();
        throw new HttpRetryException("Cannot retry streamed HTTP body", response.code());
      }

      if (!sameConnection(response, followUp.url())) {
        streamAllocation.release();
        streamAllocation = new StreamAllocation(client.connectionPool(),
            createAddress(followUp.url()), call, eventListener, callStackTrace);
        this.streamAllocation = streamAllocation;
      } else if (streamAllocation.codec() != null) {
        throw new IllegalStateException("Closing the body of " + response
            + " didn't close its backing stream. Bad interceptor?");
      }

      request = followUp;
      priorResponse = response;
    }
  }

1.建立StreamAllocation

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

主要傳入OkHttpClient中的ConnectionPool，還有通過請求request.url()建立出Address對象，主要是HTTP請求中一些SSLSocket，host認證，Dns等

  1. realChain.proceed 
try {
        response = realChain.proceed(request, streamAllocation, null, null);
        releaseConnection = false;
      }

realChain執行proceed方法，此時streamAllocation已經有值傳入，此時再次進入RealInterceptorChain#proceed方法中

public Response proceed(Request request, StreamAllocation streamAllocation, HttpCodec httpCodec,
      RealConnection connection) throws IOException {
    if (index >= interceptors.size()) throw new AssertionError();

    calls++;

 	//省略
    // 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，從interceptors中取出下一個攔截器，執行下一個攔截器的邏輯

**總結一下：**這裡責任鍊模式，每一個攔截器執行時都會建立一個攔截器鍊RealInterceptorChain，index也會随之增加1，這樣在 interceptors.get(index)中就會取出下一個攔截器，一直向下執行到沒有攔截器為止，同時每一個攔截的Response是下一個攔截器執行的傳回的結果

開源架構源碼解析系列（1）——進入OkHttp的世界1.請求部分源碼解析2 Dispatcher分析3.攔截器解析4.CallServerInterceptor 攔截器

RetryAndFollowUpInterceptor最重要的是建立了StreamAllocation

3.4 BridgeInterceptor攔截器

BridgeInterceptor#intercept

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

    RequestBody body = userRequest.body();
    if (body != null) {
      MediaType contentType = body.contentType();
      if (contentType != null) {
        requestBuilder.header("Content-Type", contentType.toString());
      }

      long contentLength = body.contentLength();
      if (contentLength != -1) {
        requestBuilder.header("Content-Length", Long.toString(contentLength));
        requestBuilder.removeHeader("Transfer-Encoding");
      } else {
        requestBuilder.header("Transfer-Encoding", "chunked");
        requestBuilder.removeHeader("Content-Length");
      }
    }

    if (userRequest.header("Host") == null) {
      requestBuilder.header("Host", hostHeader(userRequest.url(), false));
    }

    if (userRequest.header("Connection") == null) {
      requestBuilder.header("Connection", "Keep-Alive");
    }

    // If we add an "Accept-Encoding: gzip" header field we're responsible for also decompressing
    // the transfer stream.
    boolean transparentGzip = false;
    if (userRequest.header("Accept-Encoding") == null && userRequest.header("Range") == null) {
      transparentGzip = true;
      requestBuilder.header("Accept-Encoding", "gzip");
    }

    List<Cookie> cookies = cookieJar.loadForRequest(userRequest.url());
    if (!cookies.isEmpty()) {
      requestBuilder.header("Cookie", cookieHeader(cookies));
    }

    if (userRequest.header("User-Agent") == null) {
      requestBuilder.header("User-Agent", Version.userAgent());
    }

    Response networkResponse = chain.proceed(requestBuilder.build());

    HttpHeaders.receiveHeaders(cookieJar, userRequest.url(), networkResponse.headers());

    Response.Builder responseBuilder = networkResponse.newBuilder()
        .request(userRequest);

    if (transparentGzip
        && "gzip".equalsIgnoreCase(networkResponse.header("Content-Encoding"))
        && HttpHeaders.hasBody(networkResponse)) {
      GzipSource responseBody = new GzipSource(networkResponse.body().source());
      Headers strippedHeaders = networkResponse.headers().newBuilder()
          .removeAll("Content-Encoding")
          .removeAll("Content-Length")
          .build();
      responseBuilder.headers(strippedHeaders);
      String contentType = networkResponse.header("Content-Type");
      responseBuilder.body(new RealResponseBody(contentType, -1L, Okio.buffer(responseBody)));
    }

    return responseBuilder.build();
  }

BridgeInterceptor攔截器的作用主要是添加一些網絡請求的必備參數，例如Content-Type，Content-Length，Host，ConnectionAccept-Encoding，Cookie，User-Agent等，如果有使用gzip的話，還會進行gzip的處理

3.5 CacheInterceptor攔截器

3.5.1 CacheInterceptor#intercept 

@Override public Response intercept(Chain chain) throws IOException {
    Response cacheCandidate = cache != null
        ? cache.get(chain.request())
        : null;

    long now = System.currentTimeMillis();

    CacheStrategy strategy = new CacheStrategy.Factory(now, chain.request(), cacheCandidate).get();
    Request networkRequest = strategy.networkRequest;
    Response cacheResponse = strategy.cacheResponse;

    if (cache != null) {
      cache.trackResponse(strategy);
    }

    if (cacheCandidate != null && cacheResponse == null) {
      closeQuietly(cacheCandidate.body()); // The cache candidate wasn't applicable. Close it.
    }

    // If we're forbidden from using the network and the cache is insufficient, fail.
    if (networkRequest == null && cacheResponse == null) {
      return new Response.Builder()
          .request(chain.request())
          .protocol(Protocol.HTTP_1_1)
          .code(504)
          .message("Unsatisfiable Request (only-if-cached)")
          .body(Util.EMPTY_RESPONSE)
          .sentRequestAtMillis(-1L)
          .receivedResponseAtMillis(System.currentTimeMillis())
          .build();
    }

    // If we don't need the network, we're done.
    if (networkRequest == null) {
      return cacheResponse.newBuilder()
          .cacheResponse(stripBody(cacheResponse))
          .build();
    }

    Response networkResponse = null;
    try {
      networkResponse = chain.proceed(networkRequest);
    } finally {
      // If we're crashing on I/O or otherwise, don't leak the cache body.
      if (networkResponse == null && cacheCandidate != null) {
        closeQuietly(cacheCandidate.body());
      }
    }

    // If we have a cache response too, then we're doing a conditional get.
    if (cacheResponse != null) {
      if (networkResponse.code() == HTTP_NOT_MODIFIED) {
        Response response = cacheResponse.newBuilder()
            .headers(combine(cacheResponse.headers(), networkResponse.headers()))
            .sentRequestAtMillis(networkResponse.sentRequestAtMillis())
            .receivedResponseAtMillis(networkResponse.receivedResponseAtMillis())
            .cacheResponse(stripBody(cacheResponse))
            .networkResponse(stripBody(networkResponse))
            .build();
        networkResponse.body().close();

        // Update the cache after combining headers but before stripping the
        // Content-Encoding header (as performed by initContentStream()).
        cache.trackConditionalCacheHit();
        cache.update(cacheResponse, response);
        return response;
      } else {
        closeQuietly(cacheResponse.body());
      }
    }

    Response response = networkResponse.newBuilder()
        .cacheResponse(stripBody(cacheResponse))
        .networkResponse(stripBody(networkResponse))
        .build();

    if (cache != null) {
      if (HttpHeaders.hasBody(response) && CacheStrategy.isCacheable(response, networkRequest)) {
        // Offer this request to the cache.
        CacheRequest cacheRequest = cache.put(response);
        return cacheWritingResponse(cacheRequest, response);
      }

      if (HttpMethod.invalidatesCache(networkRequest.method())) {
        try {
          cache.remove(networkRequest);
        } catch (IOException ignored) {
          // The cache cannot be written.
        }
      }
    }

    return response;
  }

1.先從cache中以chain.request()為key擷取緩存的Response，這裡的request就是外部調用時建立的，這裡的cache是InternalCache，實作類為okhttp3.Cache,檢視get方法

3.5.2 okhttp3.Cache#get 

@Nullable
    Response get(Request request) {
        String key = key(request.url());

        Snapshot snapshot;
        try {
            snapshot = this.cache.get(key);
            if (snapshot == null) {
                return null;
            }
        } catch (IOException var7) {
            return null;
        }

        Cache.Entry entry;
        try {
            entry = new Cache.Entry(snapshot.getSource(0));
        } catch (IOException var6) {
            Util.closeQuietly(snapshot);
            return null;
        }

        Response response = entry.response(snapshot);
        if (!entry.matches(request, response)) {
            Util.closeQuietly(response.body());
            return null;
        } else {
            return response;
        }
    }

1.根據請求的url，進行計算獲得一個key

2.在内部cache中通過key看有沒儲存的快照Snapshot。這裡cache是采用了DiskLruCache的算法

3.如果Snapshot不為空，通過Snapshot建立出Cache.Entry，檢視一下Cache.Entry的組成

開源架構源碼解析系列（1）——進入OkHttp的世界1.請求部分源碼解析2 Dispatcher分析3.攔截器解析4.CallServerInterceptor 攔截器

其實就是存儲了一些請求傳回的資訊

4.通過entry.response方法擷取緩存中的Response

public Response response(Snapshot snapshot) {
            String contentType = this.responseHeaders.get("Content-Type");
            String contentLength = this.responseHeaders.get("Content-Length");
            Request cacheRequest = (new okhttp3.Request.Builder()).url(this.url).method(this.requestMethod, (RequestBody)null).headers(this.varyHeaders).build();
            return (new okhttp3.Response.Builder()).request(cacheRequest).protocol(this.protocol).code(this.code).message(this.message).headers(this.responseHeaders).body(new Cache.CacheResponseBody(snapshot, contentType, contentLength)).handshake(this.handshake).sentRequestAtMillis(this.sentRequestMillis).receivedResponseAtMillis(this.receivedResponseMillis).build();
        }

通過緩存得參數構造Request ，然後通過Request再建立出Response

5.校驗緩存中的請求和相應是否和傳入的Request所關聯的一緻

entry.matches(request, response)

6.将請求鍊chain中的request和緩存Response構造出一個CacheStrategy

CacheStrategy strategy = new CacheStrategy.Factory(now, chain.request(), cacheCandidate).get();

3.5.3 CacheStrategy#Factory方法

開源架構源碼解析系列（1）——進入OkHttp的世界1.請求部分源碼解析2 Dispatcher分析3.攔截器解析4.CallServerInterceptor 攔截器

其實就是從緩存的cacheResponse中取出一些值進行指派

3.5.4 CacheStrategy.Factory#get方法 

public CacheStrategy get() {
      CacheStrategy candidate = getCandidate();

      if (candidate.networkRequest != null && request.cacheControl().onlyIfCached()) {
        // We're forbidden from using the network and the cache is insufficient.
        return new CacheStrategy(null, null);
      }

      return candidate;
    }

主要邏輯在getCandidate方法中

  1. CacheStrategy#getCandidate 
private CacheStrategy getCandidate() {
      // No cached response.
      if (cacheResponse == null) {
        return new CacheStrategy(request, null);
      }

      // Drop the cached response if it's missing a required handshake.
      if (request.isHttps() && cacheResponse.handshake() == null) {
        return new CacheStrategy(request, null);
      }

      // If this response shouldn't have been stored, it should never be used
      // as a response source. This check should be redundant as long as the
      // persistence store is well-behaved and the rules are constant.
      if (!isCacheable(cacheResponse, request)) {
        return new CacheStrategy(request, null);
      }

      CacheControl requestCaching = request.cacheControl();
      if (requestCaching.noCache() || hasConditions(request)) {
        return new CacheStrategy(request, null);
      }

      CacheControl responseCaching = cacheResponse.cacheControl();
      if (responseCaching.immutable()) {
        return new CacheStrategy(null, cacheResponse);
      }

      long ageMillis = cacheResponseAge();
      long freshMillis = computeFreshnessLifetime();

      if (requestCaching.maxAgeSeconds() != -1) {
        freshMillis = Math.min(freshMillis, SECONDS.toMillis(requestCaching.maxAgeSeconds()));
      }

      long minFreshMillis = 0;
      if (requestCaching.minFreshSeconds() != -1) {
        minFreshMillis = SECONDS.toMillis(requestCaching.minFreshSeconds());
      }

      long maxStaleMillis = 0;
      if (!responseCaching.mustRevalidate() && requestCaching.maxStaleSeconds() != -1) {
        maxStaleMillis = SECONDS.toMillis(requestCaching.maxStaleSeconds());
      }

      if (!responseCaching.noCache() && ageMillis + minFreshMillis < freshMillis + maxStaleMillis) {
        Response.Builder builder = cacheResponse.newBuilder();
        if (ageMillis + minFreshMillis >= freshMillis) {
          builder.addHeader("Warning", "110 HttpURLConnection \"Response is stale\"");
        }
        long oneDayMillis = 24 * 60 * 60 * 1000L;
        if (ageMillis > oneDayMillis && isFreshnessLifetimeHeuristic()) {
          builder.addHeader("Warning", "113 HttpURLConnection \"Heuristic expiration\"");
        }
        return new CacheStrategy(null, builder.build());
      }

      // Find a condition to add to the request. If the condition is satisfied, the response body
      // will not be transmitted.
      String conditionName;
      String conditionValue;
      if (etag != null) {
        conditionName = "If-None-Match";
        conditionValue = etag;
      } else if (lastModified != null) {
        conditionName = "If-Modified-Since";
        conditionValue = lastModifiedString;
      } else if (servedDate != null) {
        conditionName = "If-Modified-Since";
        conditionValue = servedDateString;
      } else {
        return new CacheStrategy(request, null); // No condition! Make a regular request.
      }

      Headers.Builder conditionalRequestHeaders = request.headers().newBuilder();
      Internal.instance.addLenient(conditionalRequestHeaders, conditionName, conditionValue);

      Request conditionalRequest = request.newBuilder()
          .headers(conditionalRequestHeaders.build())
          .build();
      return new CacheStrategy(conditionalRequest, cacheResponse);
    }

getCandidate方法決定CacheStrategy的構成，一般會有如下一些情況：

noCache ：不使用緩存，全部走網絡

noStore ： 不使用緩存，也不存儲緩存

onlyIfCached ： 隻使用緩存

maxAge ：設定最大失效時間，失效則不使用

maxStale ：設定最大失效時間，失效則不使用

minFresh ：設定最小有效時間，失效則不使用

FORCE_NETWORK ： 強制走網絡

FORCE_CACHE ：強制走緩存

可以發現CacheStrategy中cacheResponse為null空有幾種情況

1）沒有緩存的response

2）如果這個請求是https的，但上次緩存的cacheResponse沒有TLS handshake

3.通過isCacheable判斷，一些請求傳回值不符合要求的不緩存，還有就是請求頭中有配置no-store參數時

4.請求頭中聲明了“no-cache”，或者“If-Modified-Since”，“If-None-Match”（伺服器緩存）

5.請求頭中沒有添加任何條件時候

繼續看 get()方法

if (candidate.networkRequest != null && request.cacheControl().onlyIfCached()) {
        // We're forbidden from using the network and the cache is insufficient.
        return new CacheStrategy(null, null);
      }

如果外部設定了onlyIfCached（隻讀緩存），但緩存又無效，那就構造的CacheStrategy中既沒有request也沒有request

3.5.5 再次回到 CacheInterceptor#intercept 

// If we're forbidden from using the network and the cache is insufficient, fail.
    if (networkRequest == null && cacheResponse == null) {
      return new Response.Builder()
          .request(chain.request())
          .protocol(Protocol.HTTP_1_1)
          .code(504)
          .message("Unsatisfiable Request (only-if-cached)")
          .body(Util.EMPTY_RESPONSE)
          .sentRequestAtMillis(-1L)
          .receivedResponseAtMillis(System.currentTimeMillis())
          .build();
    }

1.如果設定了only-if-cached，隻讀緩存，但又沒有緩存的Response，那就傳回504

// If we don't need the network, we're done.
    if (networkRequest == null) {
      return cacheResponse.newBuilder()
          .cacheResponse(stripBody(cacheResponse))
          .build();
    }

2.networkRequest == null這裡表示隻用緩存，不用網絡請求，那就将緩存傳回

Response networkResponse = null;
    try {
      networkResponse = chain.proceed(networkRequest);
    } finally {
      // If we're crashing on I/O or otherwise, don't leak the cache body.
      if (networkResponse == null && cacheCandidate != null) {
        closeQuietly(cacheCandidate.body());
      }
    }

3.networkRequest ！=null，那就調用接下來的攔截器進行請求，傳回Response

if (cacheResponse != null) {
      if (networkResponse.code() == HTTP_NOT_MODIFIED) {
        Response response = cacheResponse.newBuilder()
            .headers(combine(cacheResponse.headers(), networkResponse.headers()))
            .sentRequestAtMillis(networkResponse.sentRequestAtMillis())
            .receivedResponseAtMillis(networkResponse.receivedResponseAtMillis())
            .cacheResponse(stripBody(cacheResponse))
            .networkResponse(stripBody(networkResponse))
            .build();
        networkResponse.body().close();

        // Update the cache after combining headers but before stripping the
        // Content-Encoding header (as performed by initContentStream()).
        cache.trackConditionalCacheHit();
        cache.update(cacheResponse, response);
        return response;
      } else {
        closeQuietly(cacheResponse.body());
      }
    }

4.如果舊的cacheResponse不為null，又通過網絡請求傳回操作碼304，則将新的response更新

if (cache != null) {
      if (HttpHeaders.hasBody(response) && CacheStrategy.isCacheable(response, networkRequest)) {
        // Offer this request to the cache.
        CacheRequest cacheRequest = cache.put(response);
        return cacheWritingResponse(cacheRequest, response);
      }

      if (HttpMethod.invalidatesCache(networkRequest.method())) {
        try {
          cache.remove(networkRequest);
        } catch (IOException ignored) {
          // The cache cannot be written.
        }
      }
    }

5.新的請求進行緩存，然後過期緩存進行移除

3.6 ConnectInterceptor攔截器

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, chain, doExtensiveHealthChecks);
    RealConnection connection = streamAllocation.connection();

    return realChain.proceed(request, streamAllocation, httpCodec, connection);
  }

從鍊中擷取出StreamAllocation，通過streamAllocation.newStream方法傳回一個HttpCodec，HttpCodec的作用是對請求進行編碼，然後對響應進行解碼

3.6.1 StreamAllocation#newStream 

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

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

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

通過findHealthyConnection方法建立一個RealConnection，此類負責https連接配接的主要工作，RealConnection#newCodec方法可以建立出HttpCodec，并傳回。

3.6.2 StreamAllocation#findHealthyConnection

通過findConnection方法建立RealConnection，如果這個RealConnection是全新的連接配接，就跳過連接配接健康檢查，如果是之前已經連接配接過的RealConnection，則判斷是不是一個健康的連接配接，如果否的話就将其從連接配接池connectionPool中進行回收。接下來看findConnection方法做了什麼

3.6.3 StreamAllocation#findConnection 

......
if (result == null) {
        // Attempt to get a connection from the pool.
        Internal.instance.get(connectionPool, address, this, null);
        if (connection != null) {
          foundPooledConnection = true;
          result = connection;
        } else {
          selectedRoute = route;
        }
      }
 ......

Internal.instance.get方法從連接配接池中擷取是否有複用的連接配接，Internal是一個接口，它的實作在OkHttpClient的内部類中

1.OkHttpClient.Internal.instance#get

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

2.ConnectionPool#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;
  }

在連接配接池中找出能比對Address 的連接配接，注意這裡的route傳進來為null,connection.isEligible方法判斷連接配接池中connection是否可複用，主要通過判斷請求連接配接的host是否一緻，具體邏輯在RealConnection#isEligible中

3.RealConnection#isEligible

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

    // 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/#eliminate-domain-sharding
    // 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.
  }

通過streamAllocation.acquire(connection, true)，将connection和StreamAllocation相關聯

4.StreamAllocation#acquire

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，然後connection有一個集合存放與其關聯的StreamAllocation，這裡StreamAllocationReference是一個弱引用。

3.6.4 再次回到ConnectInterceptor#intercept 

RealConnection connection = streamAllocation.connection();
 return realChain.proceed(request, streamAllocation, httpCodec, connection);

通過streamAllocation擷取出connection，繼續傳遞給下一個攔截器

4.CallServerInterceptor 攔截器

這個是Okhttp中自帶的最後一個攔截器

4.1CallServerInterceptor#intercept 

@Override public Response intercept(Chain chain) throws IOException {
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    HttpCodec httpCodec = realChain.httpStream();
    StreamAllocation streamAllocation = realChain.streamAllocation();
    RealConnection connection = (RealConnection) realChain.connection();
    Request request = realChain.request();

    long sentRequestMillis = System.currentTimeMillis();

    realChain.eventListener().requestHeadersStart(realChain.call());
    httpCodec.writeRequestHeaders(request);
    realChain.eventListener().requestHeadersEnd(realChain.call(), request);

    Response.Builder responseBuilder = null;
    if (HttpMethod.permitsRequestBody(request.method()) && request.body() != null) {
      // If there's a "Expect: 100-continue" header on the request, wait for a "HTTP/1.1 100
      // Continue" response before transmitting the request body. If we don't get that, return
      // what we did get (such as a 4xx response) without ever transmitting the request body.
      if ("100-continue".equalsIgnoreCase(request.header("Expect"))) {
        httpCodec.flushRequest();
        realChain.eventListener().responseHeadersStart(realChain.call());
        responseBuilder = httpCodec.readResponseHeaders(true);
      }

      if (responseBuilder == null) {
        // Write the request body if the "Expect: 100-continue" expectation was met.
        realChain.eventListener().requestBodyStart(realChain.call());
        long contentLength = request.body().contentLength();
        CountingSink requestBodyOut =
            new CountingSink(httpCodec.createRequestBody(request, contentLength));
        BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut);

        request.body().writeTo(bufferedRequestBody);
        bufferedRequestBody.close();
        realChain.eventListener()
            .requestBodyEnd(realChain.call(), requestBodyOut.successfulCount);
      } else if (!connection.isMultiplexed()) {
        // If the "Expect: 100-continue" expectation wasn't met, prevent the HTTP/1 connection
        // from being reused. Otherwise we're still obligated to transmit the request body to
        // leave the connection in a consistent state.
        streamAllocation.noNewStreams();
      }
    }

    httpCodec.finishRequest();

    if (responseBuilder == null) {
      realChain.eventListener().responseHeadersStart(realChain.call());
      responseBuilder = httpCodec.readResponseHeaders(false);
    }

    Response response = responseBuilder
        .request(request)
        .handshake(streamAllocation.connection().handshake())
        .sentRequestAtMillis(sentRequestMillis)
        .receivedResponseAtMillis(System.currentTimeMillis())
        .build();

    int code = response.code();
    if (code == 100) {
      // server sent a 100-continue even though we did not request one.
      // try again to read the actual response
      responseBuilder = httpCodec.readResponseHeaders(false);

      response = responseBuilder
              .request(request)
              .handshake(streamAllocation.connection().handshake())
              .sentRequestAtMillis(sentRequestMillis)
              .receivedResponseAtMillis(System.currentTimeMillis())
              .build();

      code = response.code();
    }

    realChain.eventListener()
            .responseHeadersEnd(realChain.call(), response);

    if (forWebSocket && code == 101) {
      // Connection is upgrading, but we need to ensure interceptors see a non-null response body.
      response = response.newBuilder()
          .body(Util.EMPTY_RESPONSE)
          .build();
    } else {
      response = response.newBuilder()
          .body(httpCodec.openResponseBody(response))
          .build();
    }

    if ("close".equalsIgnoreCase(response.request().header("Connection"))
        || "close".equalsIgnoreCase(response.header("Connection"))) {
      streamAllocation.noNewStreams();
    }

    if ((code == 204 || code == 205) && response.body().contentLength() > 0) {
      throw new ProtocolException(
          "HTTP " + code + " had non-zero Content-Length: " + response.body().contentLength());
    }

    return response;
  }

這個攔截器主要的任務如下：

1.寫入請求頭

2.寫入請求體

3.讀取響應頭

4.讀取響應體

這樣所有的Okhttp流程基本分析完畢，再往深入網絡連接配接流這一塊能力有限無法進行分析

開源架構解析 okhttp OKHttp源碼解析
上一篇: OkHttp3找不到FormEncodingBuilder
下一篇: AI又來放大招！3D動畫自動生成！會拍視訊就能做CG動畫

繼續閱讀