Okhttp任务队列工作原理

Posted 2023-02-14 tangjiean

1 概述

1.1 引言

android完成非阻塞式的异步请求的时候都是通过启动子线程的方式来解决，子线程执行完任务的之后通过handler的方式来和主线程来完成通信。无限制的创建线程，会给系统带来大量的开销。如果在高并发的任务下，启用个线程池，可以不断的复用里面不再使用和有效的管理线程的调度和数量的管理。就可以节省系统的成本，有效的提高执行效率。

1.2 线程池ThreadPoolExecutor

ThreadPoolExecutor是java线程创建工具。存在于java.util.concurrent 包中。
看构造方法：

   public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) 
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    

• corePoolSize : 最小并发线程数。
• maximumPoolSize ：线程池中最大的线程池并发数。
• keepAliveTime : 当线程的数目大于corePoolSize时，线程的最大存活时间。
• unit : 时间单位
• BlockingQueue 工作队列

okhttp的线程池对象存在于Dispatcher类中。实例过程如下

  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;
  

1.2 Call对象

了解源码或使用过okhttp的都知道。 okttp的操作元是Call对象。异步的实现是RealCall.AsyncCall。而 AsyncCall是实现的一个Runnable接口。

final class AsyncCall extends NamedRunnable 

所以Call本质就是一个Runable线程操作元肯定是放进excutorService中直接启动的。

那么okttp是如何完成线程复用和线程管理的呢。

2 线程池的复用和管理

2.1 图解

为了完成调度和复用，定义了两个队列分别用作等待队列和执行任务的队列。这两个队列都是Dispatcher 成员变量。Dispatcher是一个控制执行，控制所有Call的分发和任务的调度、通信、清理等操作。这里只介绍异步调度任务。

  /** 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<>();

在 《okhttp连接池复用机制》 文章中我们在缓存Connection连接的时候也是使用的Deque双端队列。这里同样的方式，可以方便在队列头添加元素，移除尾部的元素。

2.2 过程分析

Call代用equeue方法的时候

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

方法中满足执行队列里面不足最大线程数maxRequests并且Call对应的host数目不超过maxRequestsPerHost 的时候直接把call对象直接推入到执行队列里，并启动线程任务（Call本质是一个Runnable）。否则，当前线程数过多，就把他推入到等待队列中。Call执行完肯定需要在runningAsyncCalls 队列中移除这个线程。那么readyAsyncCalls队列中的线程在什么时候才会被执行呢。

追溯下AsyncCall 线程的执行方法

 @Override protected void execute() 
      boolean signalledCallback = false;
      try 
        Response response = getResponseWithInterceptorChain(forWebSocket);
        if (canceled) 
          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);
      
    
  

这里做了核心request的动作，并把失败和回复数据的结果通过responseCallback 回调到Dispatcher。执行操作完毕了之后不管有无异常都会进入到dispactcher的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();
    
  

在这里call在runningAsyncCalls队列中被移除了，重新计算了目前正在执行的线程数量。并且调用了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.
    
  

原来实在这里对readyAsyncCalls 进行调度的。最终会在readyAsyncCalls 中通过remove操作把元素迭代取出并移除之后加入到runningAsyncCalls的执行队列中执行操作。ArrayDeque 是非线程安全的所以finished在调用promoteCalls 的时候都在synchronized块中执行的。执行等待队列线程当然的前提是runningAsyncCalls 线程数没有超上线，而且等待队列里面有等待的任务。

以上完成了线程线程池的复用和线程的管理工作。

小结，Call在执行任务通过Dispatcher把单元任务优先推到执行队列里进行操作，如果操作完成再执行等待队列的任务。