OkHttp3 任務隊列

OkHttp3 有兩種運行方式：

1.同步阻塞調用而且直接返回；

2.經過內部線程池分發調度實現非阻塞的異步回調;

 

下面講的是非阻塞異步回調，OkHttp在多併發網絡下的分發調度過程，主要是Dispatcher對象：

 

 多線程：多線程技術主要解決處理器單元內多個線程執行的問題，它能夠顯著減小處理器單元的閒置時間，增長處理器單元的吞吐能力。但若是對多線程應用不當，會增長對單個任務的處理時間

 ThreadPool線程池：線程池的關鍵在於線程複用以減小非核心任務的損耗。

好比：

T = T1+T2+T3其中T1和T3是多線程自己的帶來的開銷（在Java中，經過映射pThead，並進一步經過SystemCall實現native線程），咱們渴望減小T1,T3所用的時間，從而減小T的時間。但一些線程的使用者並無注意到這一點，因此在程序中頻繁的建立或銷燬線程，這致使T1和T3在T中佔有至關比例。顯然這是突出了線程的弱點（T1，T3），而不是優勢（併發性）。

線程池技術正是關注如何縮短或調整T1，T3時間的技術，從而提升服務器程序性能的。

 

1.經過對線程進行緩存，減小了建立銷燬的時間損失;

2.經過控制線程數量閥值，減小了當線程過少時帶來的CPU閒置（好比說長時間卡在I/O上了）與線程過多時對JVM的內存與線程切換時系統調用的壓力.

 

構造單例線程池：

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; 
}
參數說明：

   
   
   

  
 
 
   
 
    
int corePoolSize: 最小併發線程數，這裏併發同時包括空閒與活動的線程，若是是0的話，空閒一段時間後全部線程將所有被銷燬。
 
    
int maximumPoolSize: 最大線程數，當任務進來時能夠擴充的線程最大值，當大於了這個值就會根據丟棄處理機制來處理
 
    
long keepAliveTime: 當線程數大於corePoolSize時，多餘的空閒線程的最大存活時間，相似於HTTP中的Keep-alive
 
    
TimeUnit unit: 時間單位，通常用秒
 
    
BlockingQueue<Runnable> workQueue: 工做隊列，先進先出，能夠看出並不像Picasso那樣設置優先隊列。（阻塞隊列）
 
    
ThreadFactory threadFactory: 單個線程的工廠，能夠打Log，設置Daemon(即當JVM退出時，線程自動結束)等。
 
   
 
 
   

     能夠看出，在Okhttp中，構建了一個閥值爲[0, Integer.MAX_VALUE]的線程池，它不保留任何最小線程數，隨時建立更多的線程數，當線程空閒時只能活60秒，它使用了一個不存儲元素的阻塞工做隊列，一個叫作"OkHttp Dispatcher"的線程工廠。 
   
 
 
   

     也就是說，在實際運行中，當收到10個併發請求時，線程池會建立十個線程，當工做完成後，線程池會在60s後相繼關閉全部線程。 
   
  
 
    
 
   反向代理模型： 
 
   
 
 
   
  在OkHttp中，使用了與Nginx相似的反向代理與分發技術，這是典型的單生產者多消費者問題。咱們知道在Nginx中，用戶經過HTTP(Socket)訪問前置的服務器，服務器會添加Header並自動轉發請求給後端集羣，接着返回數據結果給用戶(好比簡書上次掛了也顯示了Nginx報錯)。經過將工做分配給多個後臺服務器並共享Session，能夠提升服務的負載均衡能力，實現非阻塞、高可用、高併發鏈接，避免資源所有放到一臺服務器而帶來的負載，速度，在線率等影響。
 

  
   
   
   

  
 
 
   
 
     
    

      而在OkHttp中，很是相似於上述場景，它使用 
     Dispatcher做爲任務的派發器，線程池對應多臺後置服務器，用 
     AsyncCall對應Socket請求，用 
     Deque<readyAsyncCalls>對應Nginx的內部緩存 
    
 
   
 
 
   
 
    
具體成員以下 
 
    
 
     
maxRequests = 64: 最大併發請求數爲64
 
     
maxRequestsPerHost = 5: 每一個主機最大請求數爲5
 
     
Dispatcher: 分發者，也就是生產者（默認在主線程）
 
     
AsyncCall: 隊列中須要處理的Runnable（包裝了異步回調接口）
 
     
ExecutorService：消費者池（也就是線程池）
 
     
Deque<readyAsyncCalls>：緩存（用數組實現，可自動擴容，無大小限制）
 
     
Deque<runningAsyncCalls>：正在運行的任務，僅僅是用來引用正在運行的任務以判斷併發量，注意它並非消費者緩存
 
    
 
    
經過將請求任務分發給多個線程，能夠顯著的減小I/O等待時間
 
   
 
 
   
OkHttp的任務調度
當咱們但願使用OkHttp的異步請求時，通常進行以下構造: 

  
   
   
   

  
 
 
    

  
   
   
   

  

   
  當HttpClient的請求 
 
   入隊時，根據代碼，咱們能夠發現其實是Dispatcher進行了 
 
   入隊操做 

  
   
   
   

  
 
 
   
synchronized void enqueue(AsyncCall call) {
 if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) { 
      //添加正在運行的請求
        runningAsyncCalls.add(call);
    //線程池執行請求
        executorService().execute(call);  
 } else { 
      //添加到緩存隊列排隊等待    
       readyAsyncCalls.add(call);  
         } 
}
 

  
   
   
   

  

   
  能夠發現請求是否進入緩存的條件以下： 
 
   (runningRequests<64 && runningRequestsPerHost<5)若是知足條件，那麼就直接把 
 
   AsyncCall直接加到 
 
   runningCalls的隊列中，並在線程池中執行（線程池會根據當前負載自動建立，銷燬，緩存相應的線程）。反之就放入 
 
   readyAsyncCalls進行緩存等待。 

  
   
   
   

  
 
 
   
咱們再分析請求元素AsyncCall（它實現了Runnable接口），它內部實現的execute方法以下:
 

  
   
   
   

  
 
 
    

  
當任務執行完成後，不管是否有異常，finally代碼段總會被執行，也就是會調用Dispatcher的finished函數，打開源碼，發現它將正在運行的任務Call從隊列runningAsyncCalls中移除後，接着執行promoteCalls()函數
   
   
   

  
 
 
    

  
   
   
   

  

   
  這樣，就主動的把緩存隊列向前走了一步，而沒有使用互斥鎖等複雜編碼. 

  
   
   
   

  
 
 
   
經過上述的分析，咱們知道了：
 
 
   
 
    
OkHttp採用Dispatcher技術，相似於Nginx，與線程池配合實現了高併發，低阻塞的運行
 
    
Okhttp採用Deque做爲緩存，按照入隊的順序先進先出
 
    
OkHttp最出彩的地方就是在try/finally中調用了finished函數，能夠主動控制等待隊列的移動，而不是採用鎖或者wait/notify，極大減小了編碼複雜性
 
   
 

  
地址：corePoolSizeDaemonAsyncCallDeque<readyAsyncCalls>synchronized void enqueue(AsyncCall call) {
 if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) { 
      //添加正在運行的請求
        runningAsyncCalls.add(call);
    //線程池執行請求
        executorService().execute(call);  
 } else { 
      //添加到緩存隊列排隊等待    
       readyAsyncCalls.add(call);  
         } 
}(runningRequests<64 && runningRequestsPerHost<5)AsyncCallrunningCallsreadyAsyncCallsfinallypromoteCalls()finished
  

   
  http://www.jianshu.com/p/aad5aacd79bf