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