從原始碼角度深入理解OKHttp3

摘要： 在日常開發中網路請求是很常見的功能。OkHttp作為Android開發中最常用的網路請求框架，在Android開發中我們經常結合retrofit一起使用，俗話說得好：“知其然知其所以然”，所以這篇文章我們通過原始碼來深入理解OKHttp3（基於3.12版本） 常規使用 ...

在日常開發中網路請求是很常見的功能。OkHttp作為Android開發中最常用的網路請求框架，在Android開發中我們經常結合retrofit一起使用，俗話說得好：“知其然知其所以然”，所以這篇文章我們通過原始碼來深入理解OKHttp3（基於3.12版本）

• 常規使用

  • 在瞭解原始碼前，我們先了解如何使用OKHttp這個框架（ ofollow,noindex">框架地址 ）

  //框架引入專案
  implementation("com.squareup.okhttp3:okhttp:3.12.0")
  
  //引用官方Demo的例子
  @Override
  protected void onCreate(Bundle savedInstanceState) {
  super.onCreate(savedInstanceState);
  setContentView(R.layout.activity_main);
  //主執行緒不能進行耗時操作
  new Thread(){
  @Override
  public void run() {
  super.run();
  /**
  * 同步請求
  */
  GetExample getexample = new GetExample();
  String syncresponse = null;
  try {
  syncresponse = getexample.run("https://raw.github.com/square/okhttp/master/README.md");
  Log.i("maoqitian","非同步請求返回引數"+syncresponse);
  } catch (IOException e) {
  e.printStackTrace();
  }
  }
  }.start();
  /**
  * 非同步請求
  */
  PostExample postexample = new PostExample();
  String json = postexample.bowlingJson("Jesse", "Jake");
  try {
  postexample.post("http://www.roundsapp.com/post", json);
  } catch (IOException e) {
  e.printStackTrace();
  }
  }
  /**
  * 非同步請求
  */
  class PostExample {
  final MediaType JSON = MediaType.get("application/json; charset=utf-8");
  //獲取 OkHttpClient 物件
  OkHttpClient client = new OkHttpClient();
  
  void post(String url, String json) throws IOException {
  RequestBody body = RequestBody.create(JSON, json);
  Request request = new Request.Builder()
  .url(url)
  .post(body)
  .build();
  client.newCall(request).enqueue(new Callback() {
  @Override
  public void onFailure(Call call, IOException e) {
  Log.i("maoqitian","非同步請求返回引數"+e.toString());
  }
  
  @Override
  public void onResponse(Call call, Response response) throws IOException {
  String asynresponse= response.body().string();
  Log.i("maoqitian","非同步請求返回引數"+asynresponse);
  }
  });
  }
  
  String bowlingJson(String player1, String player2) {
  return "{'winCondition':'HIGH_SCORE',"
  + "'name':'Bowling',"
  + "'round':4,"
  + "'lastSaved':1367702411696,"
  + "'dateStarted':1367702378785,"
  + "'players':["
  + "{'name':'" + player1 + "','history':[10,8,6,7,8],'color':-13388315,'total':39},"
  + "{'name':'" + player2 + "','history':[6,10,5,10,10],'color':-48060,'total':41}"
  + "]}";
  }
  }
  /**
  * 同步請求
  */
  class GetExample {
  OkHttpClient client = new OkHttpClient();
  String run(String url) throws IOException {
  Request request = new Request.Builder()
  .url(url)
  .build();
  try (Response response = client.newCall(request).execute()) {
  return response.body().string();
  }
  }
  }
  複製程式碼

  • 由例子我們可以看到，client.newCall(request).execute()執行的是非同步請求，我們可以加入Callback來非同步獲取返回值，Response response = client.newCall(request).execute()執行的是同步請求，更多post請求方式例子可以檢視官方 sample專案

• 原始碼分析

  • 首先看一個流程圖，對於接下來的原始碼分析有個大體印象

    

  • 通過上面的例子可以看到，不管是同步請求還是非同步請求，首先呼叫的OkHttpClient的newCall(request)方法，先來看看這個方法

    /**
    * Prepares the {@code request} to be executed at some point in the future.
    */
    @Override public Call newCall(Request request) {
    return RealCall.newRealCall(this, request, false /* for web socket */);
     }
    複製程式碼

  • 通過newCall方法的原始碼可以看到該方法返回值是Call，Call是一個介面，他的實現類是RealCall，所以我們執行的同步execute()方法或者非同步enqueue()方法都是RealCall的方法。newCall方法接收了的網路請求引數，接下來我們看看execute()和enqueue()方法

  /**
   * 同步請求
   */
   @Override public Response execute() throws IOException {
  synchronized (this) {
  if (executed) throw new IllegalStateException("Already Executed");
  executed = true;
  }
  captureCallStackTrace();
  timeout.enter();
  eventListener.callStart(this);
  try {
  client.dispatcher().executed(this);
  Response result = getResponseWithInterceptorChain();
  if (result == null) throw new IOException("Canceled");
  return result;
  } catch (IOException e) {
  e = timeoutExit(e);
  eventListener.callFailed(this, e);
  throw e;
  } finally {
  client.dispatcher().finished(this);
  }
  }
   /**
  * 非同步請求
  */
   @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));
   }
  複製程式碼

  • 這裡先看非同步的enqueue方法，很直觀可以看到真正執行網路請求的是最後一句程式碼，而它是怎麼做的呢，我們還得先弄明白dispatcher， Dispatcher的本質是非同步請求的排程器 ，它內部持有一個執行緒池，結合線程池調配併發請求。官方文件描述也說了這一點。

    

  /**最大併發請求數*/
   private int maxRequests = 64;
   /**每個主機最大請求數*/
   private int maxRequestsPerHost = 5;
   
   /** 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<>();
   
   /** Dispatcher 構造方法 */
   public Dispatcher(ExecutorService executorService) {
   this.executorService = executorService;
  }
  
  public Dispatcher() {
  }
  
   public synchronized ExecutorServiceexecutorService() {
  if (executorService == null) {
  executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
  new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
  }
  return executorService;
   }
  複製程式碼

  • 通過Dispatcher的構造方法我們知道我們可以使用自己的執行緒池，也可以使用Dispatcher預設的執行緒池，預設的執行緒池相當於CachedThreadPool執行緒池，這個執行緒池比較適合執行大量的耗時較少的任務（執行緒池介紹）。
  • 瞭解了Dispatcher之後，我們繼續探究Dispatcher的enqueue方法

  void enqueue(AsyncCall call) {
  synchronized (this) {
  readyAsyncCalls.add(call);
   }
   promoteAndExecute();
   }
   /**
  * Promotes eligible calls from {@link #readyAsyncCalls} to {@link #runningAsyncCalls} and runs
  * them on the executor service. Must not be called with synchronization because executing calls
  * can call into user code.
  *
  * @return true if the dispatcher is currently running calls.
  */
  private boolean promoteAndExecute() {
  assert (!Thread.holdsLock(this));
  
  List<AsyncCall> executableCalls = new ArrayList<>();
  boolean isRunning;
  synchronized (this) {
  for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
  AsyncCall asyncCall = i.next();
  
  if (runningAsyncCalls.size() >= maxRequests) break; // Max capacity.
  if (runningCallsForHost(asyncCall) >= maxRequestsPerHost) continue; // Host max capacity.
  
  i.remove();
  executableCalls.add(asyncCall);
  runningAsyncCalls.add(asyncCall);
  }
  isRunning = runningCallsCount() > 0;
  }
  
  for (int i = 0, size = executableCalls.size(); i < size; i++) {
  AsyncCall asyncCall = executableCalls.get(i);
  asyncCall.executeOn(executorService());
  }
  return isRunning;
   }
  複製程式碼

  • 這裡分三步走， 首先 將傳入的AsyncCall（這其實是一個Runnable物件）加入準備要執行的非同步請求佇列， 其次 呼叫promoteAndExecute()方法，變數準備要執行的非同步請求佇列，如果佇列任務數超過最大併發請求數，則直接退出遍歷，則不會進行下面的操作；如果超過每個主機最大請求數，則跳過這次迴圈，繼續下一次遍歷，否則將非同步任務加入到正在執行的非同步請求佇列， 最後 遍歷儲存非同步任務的佇列，執行AsyncCall.executeOn(executorService())方法，並且傳入了Dispatcher的預設執行緒池。

  /**
   * Attempt to enqueue this async call on {@code executorService}. This will attempt to clean up
   * if the executor has been shut down by reporting the call as failed.
   */
  void executeOn(ExecutorService executorService) {
  assert (!Thread.holdsLock(client.dispatcher()));
  boolean success = false;
  try {
  executorService.execute(this);
  success = true;
  } catch (RejectedExecutionException e) {
  InterruptedIOException ioException = new InterruptedIOException("executor rejected");
  ioException.initCause(e);
  eventListener.callFailed(RealCall.this, ioException);
  responseCallback.onFailure(RealCall.this, ioException);
  } finally {
  if (!success) {
  client.dispatcher().finished(this); // This call is no longer running!
  }
  }
  }
  複製程式碼

  • 通過執行AsyncCall.executeOn()方法的原始碼，我們看到Dispatcher的執行緒池執行了execute(this)方法， 執行非同步任務 ，並且指向的是this，也就是當前的 AsyncCall 物件（RealCall的內部類），而AsyncCall實現了抽象類 NamedRunnable

  /**
   * Runnable implementation which always sets its thread name.
   */
  public abstract class NamedRunnable implements Runnable {
  protected final String name;
  
  public NamedRunnable(String format, Object... args) {
  this.name = Util.format(format, args);
  }
  
   @Override public final void run() {
  String oldName = Thread.currentThread().getName();
  Thread.currentThread().setName(name);
   try {
  execute();
   } finally {
  Thread.currentThread().setName(oldName);
  }
   }
   protected abstract void execute();
   }
  複製程式碼

  • 可以看到NamedRunnable中run()方法呼叫了抽象方法execute()，也就說明execute()的實現在AsyncCall物件中，而上面執行緒池執行的非同步任務也是呼叫這個execute()方法，我們看看AsyncCall物件中execute()方法的實現

  @Override protected void execute() {
  boolean signalledCallback = false;
  timeout.enter();
  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) {
  e = timeoutExit(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);
  }
  }
  
  /**Dispatcher的finished方法*/
  /** Used by {@code AsyncCall#run} to signal completion. */
  void finished(AsyncCall call) {
  finished(runningAsyncCalls, call);
  }
  
  /** Used by {@code Call#execute} to signal completion. */
  void finished(RealCall call) {
  finished(runningSyncCalls, call);
  }
  
  private <T> void finished(Deque<T> calls, T call) {
  Runnable idleCallback;
  synchronized (this) {
  if (!calls.remove(call)) throw new AssertionError("Call wasn't in-flight!");
  idleCallback = this.idleCallback;
  }
  
  boolean isRunning = promoteAndExecute();
  
  if (!isRunning && idleCallback != null) {
  idleCallback.run();
  }
  }
  複製程式碼

  • 我們可以先關注最後一行，不管前面請求如何，最後finally程式碼塊中都執行了Dispatcher的finished方法，要是要將當前任務從runningAsyncCalls或runningSyncCalls 中移除， 同時把readyAsyncCalls的任務排程到runningAsyncCalls中並執行而finished方法中執行了promoteAndExecute()方法，經過前面對該方法分析，說明不管當前執行的任務如何，都會OkHttp都會去readyAsyncCalls（準備要執行的非同步請求佇列）取出下一個請求繼續執行。接下來我們繼續回到AsyncCall物件中的execute()方法，可以發現getResponseWithInterceptorChain()的方法返回了Response，說明在該方法中執行了我們的網路請求。而不管同步還是非同步請求，都是通過getResponseWithInterceptorChain()完成網路請求。

  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 (!forSocket/">WebSocket) {
  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);
  }
  複製程式碼

  • 由getResponseWithInterceptorChain()方法我們看到添加了很多Interceptor（攔截器），首先要了解每個Interceptor的作用，也能大致瞭解OKHttp完成網路請求的過程。

    • 1.首先加入我們自定義的interceptors
    • 2.通過RetryAndFollowUpInterceptor處理網路請求重試
    • 3.通過BridgeInterceptor處理請求物件轉換，應用層到網路層
    • 4.通過CacheInterceptor處理快取
    • 5.通過ConnectInterceptor處理網路請求連結
    • 6.通過CallServerInterceptor處理讀寫，和伺服器通訊，進行真正的網路請求

  • 責任鏈模式

    在閱讀接下來原始碼之前，我們先要了解責任鏈模式。通俗化的講 在責任鏈模式中有很多物件，這些物件可以理解為上面列出的攔截器，而每個物件之間都通過一條鏈子連線，網路請求在這條鏈子上傳遞，直到某一個物件處理了這個網路請求，也就是完成了網路請求 。使用這個模式的好處就是不管你用多少攔截器處理什麼操作，最終都不會影響我們的發出請求的目的，就是完成網路請求，攔截過程你可以任意新增分配責任。

  • 接著繼續看Interceptor.Chain，他是Interceptor的內部介面，前面新增的每一個攔截器都實現了Interceptor介面，而RealInterceptorChain是Interceptor.Chain介面的實現類。先看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);
    
    // Confirm that the next interceptor made its required call to chain.proceed().
    if (httpCodec != null && index + 1 < interceptors.size() && next.calls != 1) {
    throw new IllegalStateException("network interceptor " + interceptor
    + " must call proceed() exactly once");
    }
    .....
    return response;
     }
    複製程式碼

  • 通過原始碼可以注意到interceptor.intercept(next)，RetryAndFollowUpInterceptor作為預設攔截器的第一個攔截器，也就是執行了它的intercept方法

  • RetryAndFollowUpInterceptor

  • 前面說過RetryAndFollowUpInterceptor攔截器執行OKHttp網路重試，先看它的intercept方法

  /**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");
   }
  //將請求通過鏈子chain傳遞到下一個攔截器
   Response response;
   boolean releaseConnection = true;
   try {
  response = realChain.proceed(request, streamAllocation, null, null);
  releaseConnection = false;
   } catch (RouteException e) {
  // 線路異常，連線失敗，檢查是否可以重新連線
  if (!recover(e.getLastConnectException(), streamAllocation, false, request)) {
  throw e.getFirstConnectException();
  }
  releaseConnection = false;
  continue;
   } catch (IOException e) {
  // An attempt to communicate with a server failed. The request may have been sent.
  // IO異常，連線失敗，檢查是否可以重新連線
  boolean requestSendStarted = !(e instanceof ConnectionShutdownException);
  if (!recover(e, streamAllocation, requestSendStarted, request)) throw e;
  releaseConnection = false;
  continue;
   } 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;
   try {
  //效驗狀態碼、身份驗證頭資訊、跟蹤重定向或處理客戶端請求超時
  followUp = followUpRequest(response, streamAllocation.route());
   } catch (IOException e) {
  streamAllocation.release();
  throw e;
   }
  
   if (followUp == null) {
  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;
   }
  }
  複製程式碼

  • 通過RetryAndFollowUpInterceptor攔截器intercept方法原始碼，能夠理解到OKHttp的重試機制

    • 1.首先建立StreamAllocation物件（稍後分析），在一個死迴圈中將通過鏈子chain傳遞到下一個攔截器，如果捕獲異常，則判斷異常是否恢復連線，不能連線則丟擲異常，跳出迴圈並是否建立的連線池資源

    • 2.第一步沒有異常，還要返回值效驗狀態碼、頭部資訊、是否需要重定向、連線超時等資訊，捕獲異常則丟擲並退出迴圈

    • 3.如果如果重定向，迴圈超出RetryAndFollowUpInterceptor攔截器的最大重試次數，也丟擲異常，退出迴圈

      

  • 通過攔截器RetryAndFollowUpInterceptor呼叫(RealInterceptorChain) chain.proceed()方法，又再次回到了我們剛剛分析proceed()方法，而該方法繼續呼叫下一個攔截器的intercept()方法，這個攔截器就是預設的第二個攔截器BridgeInterceptor

  /**
   * Bridges from application code to network code. First it builds a network request from a user
   * request. Then it proceeds to call the network. Finally it builds a user response from the network
   * response.
   * BridgeInterceptor的intercept方法
   */
  @Override public Response intercept(Chain chain) throws IOException {
   Request userRequest = chain.request();
   Request.Builder requestBuilder = userRequest.newBuilder();
  
   RequestBody body = userRequest.body();
   ......
   Response networkResponse = chain.proceed(requestBuilder.build());
   ......
  }
  複製程式碼

  • 該攔截器主要實現了網路請求中應用層到網路層之間的資料編碼橋樑。根據使用者請求建立網路連線，根據網路響應建立網路響應，也可以看到該方法 繼續呼叫了chain.proceed()方法，同理，根據前面分析會呼叫第三個預設攔截器CacheInterceptor的intercept方法。

  • CacheInterceptor

  • 前面我們說過這個攔截器是處理快取的，接下來看看原始碼是如何實現的

  /**
   * 攔截器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) {
  //狀態碼 304 
  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;
  }
  複製程式碼

  • 先看看intercept方法的大致邏輯

    • 1.首先通過CacheStrategy.Factory().get()獲取快取策略
    • 2.如果網路被禁止使用並且沒有快取，則請求失敗，返回504
    • 3.如果有響應快取，則返回響應快取，請求完成
    • 4.沒有快取，則進行網路請求，執行下一個攔截器
    • 5.進行網路請求，如果響應狀態碼為304，說明客戶端快取了目標資源但不確定該快取資源是否是最新版本，服務端資料沒變化，繼續使用快取
    • 6.最後儲存快取

  • 快取的場景也符合設計模式中的 策略模式 ，需要CacheStrategy提供策略在不同場景下讀快取還是請求網路。

  • 瞭解了快取邏輯，繼續深入瞭解OKHttp的快取是如何做的。首先我們應該回到最初的快取攔截器設定程式碼

    /**RealCall 設定快取攔截器*/
    interceptors.add(new CacheInterceptor(client.internalCache()));
    
    /**OkHttpClient 設定快取*/
    Cache cache;
    @Override public void setCache(OkHttpClient.Builder builder, InternalCache internalCache) {
    builder.setInternalCache(internalCache);
    }
    void setInternalCache(@Nullable InternalCache internalCache) {
    this.internalCache = internalCache;
    this.cache = null;
    }
    InternalCache internalCache() {
    return cache != null ? cache.internalCache : internalCache;
    }
    
    /**Cache類中 內部持有 InternalCache */
     final DiskLruCache cache;
     final InternalCache internalCache = new InternalCache() {
     @Override public Response get(Request request) throws IOException {
     return Cache.this.get(request);
     }
    
     @Override public CacheRequest put(Response response) throws IOException {
     return Cache.this.put(response);
     }
    
     @Override public void remove(Request request) throws IOException {
     Cache.this.remove(request);
     }
    
     @Override public void update(Response cached, Response network) {
     Cache.this.update(cached, network);
     }
    
     @Override public void trackConditionalCacheHit() {
     Cache.this.trackConditionalCacheHit();
     }
    
     @Override public void trackResponse(CacheStrategy cacheStrategy) {
     Cache.this.trackResponse(cacheStrategy);
     }
    };
    複製程式碼

  • 上面我們分別截取了 RealCall類、OkHttpClient類和Cache類的原始碼，可以瞭解到攔截器使用的快取類是DiskLruCache，設定快取快取只能通過OkHttpClient的builder來設定，快取操作實現是在Cache類中，但是Cache沒有實現InternalCache介面，而是持有InternalCache介面的內部類物件來實現快取的操作方法，這樣就使得快取的操作實現只在Cache內部，外部使用者是無法實現快取操作的，方便框架內部使用，介面擴充套件也不影響外部。

    

  • ConnectInterceptor

  • 根據前面的分析，快取攔截器中也會呼叫chain.proceed方法，所以這時候執行到了第四個預設攔截器ConnectInterceptor，接著看它的intercept方法

  /**
  * 攔截器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);
  }
  複製程式碼

  • 我們看到intercept原始碼非常簡單，通過StreamAllocation開啟連線，然後就交由下一個攔截器處理請求。如何連線呢？我們需要搞懂StreamAllocation。
  • StreamAllocation物件負責協調請求和連線池之間的聯絡。每一個OKHttpClient有它對應的一個連線池，經過前面的分析我們知道StreamAllocation物件的建立在RetryAndFollowUpInterceptor攔截器的intercept方法中建立，而StreamAllocation打開了連線，則連線池在哪建立呢，答案就在OKHttpClient的Builder類構造方法中

  public Builder() {
  .......
  connectionPool = new ConnectionPool();
  .......
  }
  複製程式碼

  • 瞭解了StreamAllocation物件和ConnectionPool物件的建立，下面來分析StreamAllocation是如何開啟連線的。首先是streamAllocation.newStream()方法

  public HttpCodec newStream(
  OkHttpClient client, Interceptor.Chain chain, boolean doExtensiveHealthChecks) {
  ........
  try {
  RealConnection resultConnection = findHealthyConnection(connectTimeout, readTimeout,
  writeTimeout, pingIntervalMillis, connectionRetryEnabled, doExtensiveHealthChecks);
  .......
  }
  } catch (IOException e) {
  throw new RouteException(e);
  }
   }
  
  /**
  * Finds a connection and returns it if it is healthy. If it is unhealthy the process is repeated
  * until a healthy connection is found.
  */
  private RealConnection findHealthyConnection(int connectTimeout, int readTimeout,
  int writeTimeout, int pingIntervalMillis, boolean connectionRetryEnabled,
  boolean doExtensiveHealthChecks) throws IOException {
  while (true) {
  RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout,
  pingIntervalMillis, connectionRetryEnabled);
  ........
  return candidate;
  }
  } 
  /**
  * Returns a connection to host a new stream. This prefers the existing connection if it exists,
  * then the pool, finally building a new connection.
  */
  private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout,
  int pingIntervalMillis, boolean connectionRetryEnabled) throws IOException {
  ............
  //
  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;
  }
  }
   
   ..........
  
  if (!foundPooledConnection) {
  ........
  result = new RealConnection(connectionPool, selectedRoute);
  //記錄每個連線的引用，每個呼叫必須與同一連線上的呼叫配對。
  acquire(result, false);
  }
  }
  .........
  synchronized (connectionPool) {
  .......
  // Pool the connection. 將連線放入連線池
  Internal.instance.put(connectionPool, result);
  ......
  }
   }
  .......
  return result;
  }
  複製程式碼

  • 根據上面的原始碼，我們可以知道findHealthyConnection在迴圈找健康的連線，直到找到連線，說明findConnection方法是尋找連線的核心方法，該方法中存在可以複用的連線則複用，否則建立新的連線，並且記錄連線引用，我們可以明白 StreamAllocation主要是為攔截器提供一個連線， 如果連線池中有複用的連線則複用連線， 如果沒有則建立新的連線 。

    

  • ConnectionPool連線池實現

  • 明白StreamAllocation是如何建立和複用連線池，我們還要明白連線池（ConnectionPool）的是如何實現的。

  • 理解ConnectionPool之前，我們需要明白TCP連線的知識，Tcp建立連線三次握手和斷開連線四次握手過程是需要消耗時間的，在http/1.0每一次請求只能開啟一次連線，而在http/1.1是支援 持續連線（persistent connection） ，使得一次連線開啟之後會保持一段時間，如果還是同一個請求並且使同一個伺服器則在這段時間內繼續請求連線是可以複用的。而ConnectionPool也實現了這個機制，在它內部持有一個執行緒池和一個快取連線的雙向列表，連線中最多隻能存在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 longkeepAliveDurationNs;
  private final Runnable cleanupRunnable = new Runnable() {
  @Override public void run() {
  // 後臺定期清理連線的執行緒
  while (true) {
  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<>();
  ............
  /**
  * Create a new connection pool with tuning parameters appropriate for a single-userapplication.
  * The tuning parameters in this pool are subject to change in future OkHttp releases. Currently
  * this pool holds up to 5 idle connections which will be evicted after 5 minutes of inactivity.
  */
  public ConnectionPool() {
  this(5, 5, TimeUnit.MINUTES);
  }
  ............
   }
  複製程式碼

  

  • 這裡還要說的一點是streamAllocation.newStream()返回的HttpCodec物件就是我們編碼HTTP請求並解碼HTTP響應的介面，他的實現類Http2Codec和Http1Codec對應https和http的解析request與響應response對socket讀寫過程實現，並最終放到RealConnection物件newCodec類中建立。

  /** 
  RealConnection類newCodec方法
  */
   public HttpCodec newCodec(OkHttpClient client, Interceptor.Chain chain,
  StreamAllocation streamAllocation) throws SocketException {
   if (http2Connection != null) {
  return new Http2Codec(client, chain, streamAllocation, http2Connection);
   } else {
  socket.setSoTimeout(chain.readTimeoutMillis());
  source.timeout().timeout(chain.readTimeoutMillis(), MILLISECONDS);
  sink.timeout().timeout(chain.writeTimeoutMillis(), MILLISECONDS);
  return new Http1Codec(client, streamAllocation, source, sink);
  }
  }
  複製程式碼

  • streamAllocation得到連線物件，也就是RealConnection物件，它封裝了套接字socket連線，也就是該類的connectSocket方法。並且使用OKio來對資料讀寫。OKio封裝了Java的I/O操作，這裡就不細說了。最後返回的ConnectInterceptor攔截器的intercept方法同樣呼叫了Chain.proceed，將拿到的連線交由CallServerInterceptor做處理。

  /** Does all the work necessary to build a full HTTP or HTTPS connection on a raw socket. 
  RealConnection類connectSocket方法
  */
  private void connectSocket(int connectTimeout, int readTimeout, Call call,
  EventListener eventListener) 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);
  
  eventListener.connectStart(call, route.socketAddress(), 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;
  }
  
  // The following try/catch block is a pseudo hacky way to get around a crash on Android 7.0
  // More details:
  // https://github.com/square/okhttp/issues/3245
  // https://android-review.googlesource.com/#/c/271775/
  try {
  //使用OKio來對資料讀寫
  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);
  }
   }
  }
  複製程式碼

  • 最後返回的ConnectInterceptor攔截器的intercept方法同樣呼叫了Chain.proceed，將拿到的連線交由CallServerInterceptor做處理。

  • CallServerInterceptor

  • 在經過前面一系列攔截器之後，OKHttp最終把拿到網路請求連線給到CallServerInterceptor攔截器進行網路請求和伺服器通訊。

  /**CallServerInterceptor的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());
  //按照HTTP協議，依次寫入請求體
  httpCodec.writeRequestHeaders(request);
  .................
   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();
  ...............
  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 {
  //響應資料OKio寫入
  response = response.newBuilder()
  .body(httpCodec.openResponseBody(response))
  .build();
  }
  }
  return response;
  }
  /**Http1Codec方法**/
  //OKio 讀寫物件
  final BufferedSource source;
  final BufferedSink sink;
  @Override public void writeRequestHeaders(Request request) throws IOException {
  //構造好請求頭
  String requestLine = RequestLine.get(
  request, streamAllocation.connection().route().proxy().type());
   writeRequest(request.headers(), requestLine);
  }
   /** Returns bytes of a request header for sending on an HTTP transport.
   將請求資訊寫入sink
   */
  public void writeRequest(Headers headers, String requestLine) throws IOException {
  if (state != STATE_IDLE) throw new IllegalStateException("state: " + state);
  sink.writeUtf8(requestLine).writeUtf8("\r\n");
  for (int i = 0, size = headers.size(); i < size; i++) {
  sink.writeUtf8(headers.name(i))
  .writeUtf8(": ")
  .writeUtf8(headers.value(i))
  .writeUtf8("\r\n");
   }
  sink.writeUtf8("\r\n");
  state = STATE_OPEN_REQUEST_BODY;
  }
  複製程式碼

  • 可以看到在CallServerInterceptor攔截器的方法中首先通過HttpCodec（上面貼的是Http1Codec的方法）writeRequestHeaders和writeRequest方法寫入請求體，並將請求體寫入OKio的寫入物件sink中

  /**Http1Codec方法**/
  /**
   讀取響應頭資訊
  */ 
  @Override public Response.Builder readResponseHeaders(boolean expectContinue) throws IOException {
  if (state != STATE_OPEN_REQUEST_BODY && state != STATE_READ_RESPONSE_HEADERS) {
  throw new IllegalStateException("state: " + state);
  }
  
   try {
  StatusLine statusLine = StatusLine.parse(readHeaderLine());
  
  Response.Builder responseBuilder = new Response.Builder()
  .protocol(statusLine.protocol)
  .code(statusLine.code)
  .message(statusLine.message)
  .headers(readHeaders());
  
  if (expectContinue && statusLine.code == HTTP_CONTINUE) {
  return null;
  } else if (statusLine.code == HTTP_CONTINUE) {
  state = STATE_READ_RESPONSE_HEADERS;
  return responseBuilder;
  }
  
  state = STATE_OPEN_RESPONSE_BODY;
  return responseBuilder;
  } catch (EOFException e) {
  // Provide more context if the server ends the stream before sending a response.
  IOException exception = new IOException("unexpected end of stream on " + streamAllocation);
  exception.initCause(e);
  throw exception;
  }
  }
   /**
   寫入響應輸入到ResponseBody
   */
   @Override public ResponseBody openResponseBody(Response response) throws IOException {
   streamAllocation.eventListener.responseBodyStart(streamAllocation.call);
   String contentType = response.header("Content-Type");
  
   if (!HttpHeaders.hasBody(response)) {
  Source source = newFixedLengthSource(0);
  return new RealResponseBody(contentType, 0, Okio.buffer(source));
   }
  
   if ("chunked".equalsIgnoreCase(response.header("Transfer-Encoding"))) {
  Source source = newChunkedSource(response.request().url());
  return new RealResponseBody(contentType, -1L, Okio.buffer(source));
   }
  
   long contentLength = HttpHeaders.contentLength(response);
   if (contentLength != -1) {
  Source source = newFixedLengthSource(contentLength);
  return new RealResponseBody(contentType, contentLength, Okio.buffer(source));
   }
  
  return new RealResponseBody(contentType, -1L, Okio.buffer(newUnknownLengthSource()));
  }
  複製程式碼

  • 通過readResponseHeaders方法讀取響應頭資訊，openResponseBody得到響應體資訊。最終將網路請求的響應資訊通過Callback()回撥方法非同步傳遞出去，同步請求則直接返回。到此OKHttp原始碼理解到此為止。

最後，通過OKHttp這個框架原始碼閱讀，也是對自己的一個提升，不僅瞭解了框架原理，設計模式在適宜場景的運用，同時也是對自己耐心的一次考驗，原始碼的閱讀是枯燥的，但是隻要靜下心來，也能發現閱讀原始碼的樂趣。由於本人水平有限，文章中如果有錯誤，請大家給我提出來，大家一起學習進步，如果覺得我的文章給予你幫助，也請給我一個喜歡和關注。

• 參考連結：

  • 深入解析OkHttp3
  • OkHttp3原始碼和設計模式-1

• 參考書籍：

  • 《計算機網路》第六版

• 簡書同步地址