OKhttp源碼學習（九）—— 任務管理（Dispatcher）

源碼位址：https://github.com/square/okhttp

針對具體一個請求的流程，前面已經做了學習分析，現在對OkHttp的請求任務管理進行分析學習。

使用過OkHttp的都知道，調用分為同步阻塞式的請求execute()，以及異步調用 enqueue(Callback responseCallback)

，同步請求沒有什麼好分析的，基本就是直接發起了請求。這裡主要分析異步請求，是如何進行請求的管理和配置設定的。

主要的類：Dispatcher.

主要内容：

1. 線程池
2. 任務分發模型

1. 線程池

線程池，為解決的問題，很多資料都有具體的闡述，這裡就引用一些專業的解釋多線程：

多線程技術主要解決處理器單元内多個線程執行的問題，它可以顯著減少處理器單元的閑置時間，增加處理器單元的吞吐能力。但如果對多線程應用不當，會增加對單個任務的處理時間。可以舉一個簡單的例子：

假設在一台伺服器完成一項任務的時間為T

T1 建立線程的時間

T2 線上程中執行任務的時間，包括線程間同步所需時間

T3 線程銷毀的時間

顯然T ＝ T1＋T2＋T3。注意這是一個極度簡化的假設。

可以看出T1,T3是多線程本身的帶來的開銷（在Java中，通過映射pThead，并進一步通過SystemCall實作native線程），我們渴望減少T1,T3所用的時間，進而減少T的時間。但一些線程的使用者并沒有注意到這一點，是以在程式中頻繁的建立或銷毀線程，這導緻T1和T3在T中占有相當比例。顯然這是突出了線程的弱點（T1，T3），而不是優點（并發性）。

線程池，就是針對解決減少T1 和 T3的時間，提高服務的性能。

1.1 OkHttp的線程池

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

Dispatcher 通過單例建立了一個線程池，針對幾個參數，可以發現，OkHttp的線程池具備特點：

1. 線程數區間[0，Integer.MAX_VALUE]，不保留最少線程數，随時建立更多線程 ；
2. 當線程空閑的時候，最多保活時間為60s；
3. 使用一個同步隊列作為工作隊列，先進先出；
4. 建立一個名為“OkHttp Dispatcher” 的線程工廠 ThreadFactory 。

線程池的處理，就到這裡，下面就開始，分析OkHttp是如何使用這個線程池來進行請求任務的排程和配置設定的。

2. 任務分發模型

在我們發起一個異步請求的時候，其實是交給了Dispatcher來處理的

RealCall.java

@Override public void enqueue(Callback responseCallback) {
    synchronized (this) {
      if (executed) throw new IllegalStateException("Already Executed");
      executed = true;
    }
    captureCallStackTrace();
    //處理再這裡
    client.dispatcher().enqueue(new AsyncCall(responseCallback));
  }
           

在Dispatcher中：

synchronized void enqueue(AsyncCall call) {
    if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
      // 添加到runningAsyncCalls隊列中
      runningAsyncCalls.add(call);
      //線程池的調用
      executorService().execute(call);
    } else {
      //添加到準備中的隊列中
      readyAsyncCalls.add(call);
    }
  }
           

上面的代碼中幾個重要的變量：

1. runningAsyncCalls 儲存運作中的異步請求隊列
2. readyAsyncCalls 儲存準備中的異步請求隊列
3. maxRequests 最大請求數量（64個）
4. maxRequestsPerHost 相同Host最大請求數量（5）

當如果運作中的請求少用64個以及相同 Host的請求小于5個的時候，直接添加到runningAsyncCalls并且調用線程池來執行這個AsyncCall,交給線程池去排程。 如果已經超出了這個限制，就把這個請求添加到 readyAsyncCalls，等待調用。但是這個準備中的隊列是什麼時候被調用的呢？

撸一下代碼，發現在AsyncCall的execute方法裡面。也就是這個異步線程的執行方法裡面：

@Override protected void execute() {
      boolean signalledCallback = false;
      try {
        //執行真正的請求
        Response response = getResponseWithInterceptorChain();
         
        //通過CallBack 回調給使用者
        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 {
          responseCallback.onFailure(RealCall.this, e);
        }
      } finally {
        //重點在這裡
        client.dispatcher().finished(this);
      }
    }
           

最後的時候調用了Dispatcher的finished，繼續向下撸：

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

這裡會調用一個 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.
    }
  }
           

終于找到了，在請求完成的時候，調用Dispatcher的finished同時，會檢查這個時候準備中的請求，是否有可以添加到運作請求中線程池中去的。

3. 總結

整個任務管理的流程，其實也不複雜：

1. 通過兩個請求隊列，來管理請求數量以及準備中的請求的數量；
2. 請求交一個線程池來完成。

最後用一個圖的形式總結一下，OkHttp的請求任務管理的實作：

系列：

OKhttp源碼學習（一）—— 基本請求流程

OKhttp源碼學習（二）—— OkHttpClient

OKhttp源碼學習（三）—— Request, RealCall

OKhttp源碼學習（四）—— RetryAndFollowUpInterceptor攔截器

OKhttp源碼學習（五）—— BridgeInterceptor攔截器

OKhttp源碼學習（六）—— CacheInterceptor攔截器

OKhttp源碼學習（七）—— ConnectInterceptor攔截器

OKhttp源碼學習（八）—— CallServerInterceptor攔截器