天天看点
返回

OkHttp3源码分析之任务队列Dispatcher前言分析总结

前言

在前边，我们知道了整个OkHttp3发起一次网络请求的整个流程，还有分析了它的一个很棒的设计——拦截器/链。本篇文章主要了解在发起异步请求时，OkHttp3是怎么调度任务的，使其能够高效地执行异步任务。

分析

源码基于最新的版本：3.10.0。

我们先回到第一篇文章，在分析异步请求的回调方法执行的线程位置时：

首先

RealCall#enqueue()

：

@Override 
public void enqueue(Callback responseCallback) {
    ...
    client.dispatcher().enqueue(new AsyncCall(responseCallback));
  }

client.dispatcher()

会在

OkHttpClient

初始化的时候

new

一个

Dispatcher

对象。

Dispatcher

就是其任务队列机制的核心，任务分发器，负责为任务找到合适的执行线程。

那么我们先来看一下

Dispatcher

的几个重要的成员变量和方法：

public final class Dispatcher {
  private int maxRequests = ;
  private int maxRequestsPerHost = ;
  private @Nullable Runnable idleCallback;

  /** Executes calls. Created lazily. */
  private @Nullable ExecutorService executorService;

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

  /** Running synchronous calls. Includes canceled calls that haven't finished yet. */
  private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

  public Dispatcher(ExecutorService executorService) {
    this.executorService = executorService;
  }

  public Dispatcher() {
  }

  public synchronized ExecutorService executorService() {
    if (executorService == null) {
      executorService = new ThreadPoolExecutor(, Integer.MAX_VALUE, , TimeUnit.SECONDS,
          new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
    }
    return executorService;
  }

其中：

  • readyAsyncCalls

     ：待执行异步任务队列

  • runningAsyncCalls

     ：运行中异步任务队列，包括未完成已经取消的请求

  • runningSyncCalls

     ：运行中同步任务队列，包括未完成已经取消的请求

我们注意到上边三个队列都是用

ArrayDeque

来管理的，

ArrayDeque

是在Java容器

Deque

接口循环队列(双端队列)的实现，其内部维护了一个可变的数组。

关于更多关于

ArrayDeque

的分析，可以参考下面这篇文章：

Java ArrayDeque源码剖析

然后在

Dispatcher#enqueue()

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

  • maxRequests

     ：当前最大请求数（64）

  • maxRequestsPerHost

     ：对单一host的最大请求数（5）

可以看到由

HttpClient

将

Request

转为

Call

后，进行请求时，实际上就是交由

Dispatcher

进行了入队操作。

如果当前请求数小于最大请求数且当前共用一个主机的请求数小于单个主机最大请求数，则将该任务插入正在执行任务队列，并执行对应任务。如果不满足则将其放入待执行队列。

接着到执行

Dispatcher#executorService()

 
public synchronized ExecutorService executorService() {
    if (executorService == null) {
      executorService = new ThreadPoolExecutor(, Integer.MAX_VALUE, , TimeUnit.SECONDS,
          new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
    }
    return executorService;
  }

  • executorService

     ：任务队列线程池

首先多线程技术主要解决处理器单元内多个线程执行的问题，它可以显著减少处理器单元的闲置时间，增加处理器单元的吞吐能力。但是频繁的创建和销毁线程会有很大的开销，造成性能的降低，而线程池的关键就在于线程复用以减少非核心任务的损耗，而且便于对线程统一进行维护和管理。

源码中用的是下边这个构造方法：

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }

下边简单说一下，构造方法中的几个参数：

  • corePoolSize

     核心线程数，默认情况下核心线程会一直存活，即使处于闲置状态也不会受存

    keepAliveTime

     限制。除非将

    allowCoreThreadTimeOut

     设置为true。

  • maximumPoolSize

     线程池所能容纳的最大线程数。超过这个数的线程将被阻塞。当任务队列为没有设置大小的

    LinkedBlockingDeque

     时，这个值无效。

  • keepAliveTime

     非核心线程的闲置超时时间，超过这个时间就会被回收。

  • unit

     指定

    keepAliveTime

     的单位，如

    TimeUnit.SECONDS

     。当将

    allowCoreThreadTimeOut

     设置为true时对

    corePoolSize

     生效。

  • workQueue

    线程池中的任务队列.

    常用的有三种队列，

    SynchronousQueue

     ,

    LinkedBlockingDeque

     ,

    ArrayBlockingQueue

     。

  • threadFactory

     线程工厂，提供创建新线程的功能。它是一个接口，内部只有一个方法，就是创建一个新的线程。

  • RejectedExecutionHandler

    RejectedExecutionHandler

     也是一个接口，只有一个方法

    rejectedExecution()

     ，当线程池中的资源已经全部使用，添加新线程被拒绝时，会调用

    RejectedExecutionHandler

     的

    rejectedExecution()

     方法。

所以OkHttp没有设置核心线程数，且最大线程数算是不设限了，空闲线程存活时间为60s，设置了

SynchronousQueue

这一任务队列（关于

SynchronousQueue

的分析，可以查看这里），一个名为”OkHttp Dispatcher”的线程工厂。

关于线程池的内容也就先说到这里，更加深入的可以参考这篇文章：

深入理解java线程池—ThreadPoolExecutor

好，我们接着往下，

RealCall.AsyncCall#execute()

 
@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 {
          eventListener.callFailed(RealCall.this, e);
          responseCallback.onFailure(RealCall.this, e);
        }
      } finally {
        //通知dispatcher对应任务已完成
        client.dispatcher().finished(this);
      }
  }

我们注意到最后无论任务执行是否成功，我们都要执行

client.dispatcher().finished(this);

，这一句。其实在同步任务中，同样的：

@Override 
 public Response execute() throws IOException {
    synchronized (this) {
      if (executed) throw new IllegalStateException("Already Executed");
      executed = true;
    }
    captureCallStackTrace();
    eventListener.callStart(this);
    try {
      client.dispatcher().executed(this);
      Response result = getResponseWithInterceptorChain();
      if (result == null) throw new IOException("Canceled");
      return result;
    } catch (IOException e) {
      eventListener.callFailed(this, e);
      throw e;
    } finally {
      //通知dispatcher对应任务已完成
      client.dispatcher().finished(this);
    }
  }

最后也是要调用

client.dispatcher().finished(this);

方法，只不过两者在调用真正执行的方法时，传入的参数不同。

那么我们来看一下

Dispatcher#finished()

方法：

void finished(AsyncCall call) {
    //异步任务调用
    finished(runningAsyncCalls, call, true);
  }

void finished(RealCall call) {
    //同步任务调用
    finished(runningSyncCalls, call, false);
  }

//真正执行的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 ==  && idleCallback != null) {
      idleCallback.run();
    }
  }

首先不管是异步还是同步，都要执行出队操作，如果是异步任务呢，多了一步：

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

首先两个判断，当前运行中异步任务队列长度大于最大请求数（64），直接return，因为不能超过负载。如果准备队列为空，也return，因为无准备任务要添加。

接着往下，也就是准备队列有值，那么就使用迭代器遍历，首先判断当前任务的请求主机的请求数是否小于最大单主机请求数，它的判断呢，也非常简单：

Dispatcher#runningCallsForHost()

 
private int runningCallsForHost(AsyncCall call) {
    int result = ;
    for (AsyncCall c : runningAsyncCalls) {
      if (c.get().forWebSocket) continue;
      if (c.host().equals(call.host())) result++;
    }
    return result;
  }

遍历整个运行中的异步任务队列，如果是

forWebSocket

，则跳过。如果队列中的异步任务的请求主机和该任务的请求主机相同，

result

就加1，最后和

maxRequestsPerHost

（5）做比较。

这里插一句，OkHttp从3.4.1版本加入了对WebSocket协议的支持，这里不做深入了解，需要的可以查找相关资料。

我们接着往下，如果满足条件，就将该任务从准备队列中移除，添加到运行队列中，然后执行该任务。当然，我们不能一直添加，是有负载限制的，所以在每次添加完一个任务之后，都要再次判断当前运行任务队列长度是否超过最大负载，如果没有则继续添加，反之则return。

可能这时候给一个图的话会更清晰一点，我就偷个懒，在网上找到一个图这里：

OkHttp3源码分析之任务队列Dispatcher前言分析总结

上边是异步任务多出来的部分，我们继续往下看，

runningCallsCount = runningCallsCount();

统计当前运行任务队列的长度，很简单就是异步任务和同步任务队列的长度相加：

public synchronized int runningCallsCount() {
    return runningAsyncCalls.size() + runningSyncCalls.size();
}

然后如果当前运行任务队列长度总数量为0，也就是说分发器处于空闲状态，就执行空闲回调。

  • idleCallback

     这个成员变量就是一个

    Runnable

     对象，它的唯一一次赋值就是在set方法中，也就是说暴露了一个接口，使得我们可以自定义空闲状态时的回调。当分发器空闲时，就会执行这个回调方法。

总结

好了，上边就是任务队列的基本内容了，我们总结一下，在执行一个异步或者同步请求时，

Dispatcher

将其进行入队操作，然后不管改任务运行成功与否，都会调用

finished()

方法，首先做出队操作，如果是异步任务，会从准备队列中取出任务，放入正在运行队列中，并执行该任务。最后如果分发器空闲且我们设置了空闲回调，当分发器处于空闲状态时，就会调用我们设置的空闲回调。

Android学习笔记 OkHttp3源码分析 okhttp3 dispatcher
上一篇: 警告: No mapping found for HTTP request with URI [/Springmvc/test] in DispatcherServlet with dispatche
下一篇: The Struts dispatcher cannot be found. This is usually caused by using Struts tags without the asso

继续阅读