在上一节中，我对Dispatcher进行了概述。本节主要内容就是带大家了解Dispatcher如何完成任务调度，并进行管理同步／异步的请求状态，如何维护一个线程池，来执行请求。
成员变量
为了下文描述更直观，这里我们先熟悉一下Dispathcer的几个重要的成员变量。我们还是结合源码来看：
public final class Dispatcher {
private int maxRequests = 64;
private int maxRequestsPerHost = 5;
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<>();
...
}
以上是Dispatcher类的所有成员变量，从备注的翻译我们也能理解。当然为了照顾所有小伙伴，我这里还是来解释一下。
- ExecutorService
这个就是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;
}
- readyAsyncCalls
就绪的异步请求队列，保存所有就绪的异步请求，等待线程池来执行。 - runningAsyncCalls
正在执行的异步请求队列，保存所有正在执行的异步请求，包含那些已被取消的但是还没完成的请求Call。 - runningSyncCalls
同步请求队列，用于保存所有同步请求Call，包含那些已被取消的但是还没完成的请求Call。
如何维护同步请求状态
对于同步请求，Dispatcher就是通过runningSyncCalls管理的，下面从源码中分析一下如何做到的。
我们先回到RealCall的execute方法中
@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 {
client.dispatcher().finished(this);
}
}
我们注意到这里通过调用client.dispatcher().executed(this);方法add这个call
/** Used by {@code Call#execute} to signal it is in-flight. */
synchronized void executed(RealCall call) {
runningSyncCalls.add(call);
}
并在finally代码块中执行client.dispatcher().finished(this);来remove这个call
/** Used by {@code Call#execute} to signal completion. */
void finished(RealCall call) {
finished(runningSyncCalls, call, false);
}
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();
}
}
如何维护异步请求状态
这里讲述的将是Dispatcher中的核心思想。我们先回到第一部分成员变量，我们发现异步请求使用了两个队列来存放，Why?。且执行异步请求的线程池初始化时第二个参数maximumPoolSize使用的是Integer.MAX_VALUE，Why?
其实，这里采用到了生产者消费者模式来实现管理，接下来我将带大家了解Dispatcher是怎么做的。
为了大家更好的理解这个问题，我们把Dispatcher理解成生产者，ExecutorService理解成消费者。有了生产者和消费者，就需要有两个队列来存放正在执行异步请求和等待的异步请求，也就是Deque<AsyncCall> readyAsyncCalls 、Deque<AsyncCall> runningAsyncCalls这两个队列，这就回答了第一个问题。
我们注意到我们的线程池maximumPoolSize使用的是Integer.MAX_VALUE居然没有对线程数量做限制。那么是不是这样呢？我们来看一下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、maxRequestsPerHost即最大请求数、最大同主机同时请求数。当请求超过这两个最大值，将会存放到等待队列中，因此，线程池ExecutorService线程池的大小将在这里得到限制，并不是初始化传入的Integer.MAX_VALUE，这就回答了第二个问题。
那么有同学肯定在疑问，就绪队列中的请求又是什么时候执行的呢？why。这里将涉及的一个很核心的方法promoteCalls()。
别急，我们接着往下看，异步请求在子线程中执行即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 {
client.dispatcher().finished(this);
}
}
注意到在finally中同样执行了client.dispatcher().finished(this);
/** Used by {@code AsyncCall#run} to signal completion. */
void finished(AsyncCall call) {
finished(runningAsyncCalls, call, true);
}
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();
}
}
异步请求执行的finished方法参数AsyncCall ，此时注意到调用最终实现到finished方法时传入的promoteCalls参数为true，因此对于异步请求结束时，除了从异步请求队列runningAsyncCalls中remove结束的异步请求，同时调用了promoteCalls()方法。注意到这个过程是线程不安全的，所有放到同步代码块中去执行。这个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.
}
}
其实这个方法的作用就是管理任务队列的，我们注意到中间有一段for循环，在if中判断正在执行的call请求是否小于maxRequestsPerHost，小于的话就将就绪队列readyAsyncCalls中的请求添加到正在执行的队列runningAsyncCalls中，同时移除就绪队列中请求，再调用线程池来执行这个被移除的就绪队列请求call。
小结
写到这里，关于Dispatcher的源码解析已经告一段落，想必大家对该类也有了一点的认识。回过头来看看Dispatcher的定义是不是更清晰了呢。
上一节 okhttp 源码学习（二）基本流程
下一节 okhttp 源码学习（四）RetryAndFollowUpInterceptor 深入解析
