Okhttp源碼解析(二)——任務調度

Okhttp源碼版本：3.4.2

參考：https://www.jianshu.com/p/074dff0f4ecb 

 

一.來源

在對Okhttp的使用中

執行的操做就是在RealCall類進行

同步執行：

　　

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

　

異步執行：

　　　

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

　　

final class AsyncCall extends NamedRunnable {
    private final Callback responseCallback;

    private AsyncCall(Callback responseCallback) {
      super("OkHttp %s", redactedUrl().toString());
      this.responseCallback = responseCallback;
    }

    String host() {
      return originalRequest.url().host();
    }

    Request request() {
      return originalRequest;
    }

    RealCall get() {
      return RealCall.this;
    }

    @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 {
          responseCallback.onFailure(RealCall.this, e);
        }
      } finally {
        client.dispatcher().finished(this);
      }
    }
  }

　　具體的執行都是在RealCall中進行，同步的在RealCall的execute()執行，異步的在RealCall.AsyncCall.execute()執行。二者都是經過getResponseWithInterceptorChain()責任連模式調用。

 

 

二.Dispatcher類

1. 線程池

　　　　內部維護着一個線程池，線程池相關能夠了解下另一片文章

　　　　

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

　　　　該線程維護的阻塞隊列爲同步隊列SychronousQueue，該隊列的特色就是生產者往隊列放入數據的前提要有消費者準備消費，一樣消費者要從隊列中獲取數據前提是要有生產者準備往數據放入數據。這在多任務處理的隊列中屬於最快的任務處理方式，網絡請求屬於高頻狀況，最佳。

 

　　2.請求

　　　　異步：

synchronized void enqueue(AsyncCall call) {
    if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
      runningAsyncCalls.add(call);
      executorService().execute(call);
    } else {
      readyAsyncCalls.add(call);
    }
  }

　　

　　　　內部維護者一個待運行異步任務的集合Deque<AsyncCall> readyAsyncCalls，和一個正在運行異步任務的集合Deque<AsyncCall> runningAsyncCalls，（Deque就是一個繼承Queue的接口，雙端操做的隊列，而AsyncCall就是一個通過封裝的Runnable）,每次執行異步任務就是：1.判斷runningAsyncCalls長度是否到達了最大數64，2.判斷runningAsyncCall中訪問同一個服務器不一樣端口的數量是否達到5，若是都沒有則往runningAsyncCall增長該任務，而且讓線程池執行該任務。若是1或者2知足了就只往readAsyncCalls中添加該任務。

 

　　　　同步：

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

　　

　　　　內部還維護着一個待運行同步任務的集合Deque<RealCall> readySyncCalls。執行同步任務就是隻往同步隊列中添加該realCall

　　3.完成

　　　　

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

 

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

　　

　

不管同步的任務仍是異步的任務，完成時都執行finished方法，calls就可放入同步的running隊列或者異步的running隊列，總的來講就是往隊列中remove掉執行玩的call，同時因爲Dispacher類有一個idleCallback，判斷當前running的同步隊列+running的異步隊列，若是任務數==0，就執行idleCallback。

promoteCalls方法就是當一個call結束時，是否執行異步任務隊列中下一個任務