Okhttp3源码分析(2) Dispatcher分发器

灰太狼 2022-05-24 08:23 274阅读 0赞

> okhttp3源码分析基于okhttp3.10.0。

在上一章节中提到在RealCall请求方法中，不管是同步请求方法execute()还是异步请求方法enqueue(CallBack)最终都会调用Dispatcher分发器的相关方法，所以这里对Dispatcher类做具体分析。

# Dispatcher #

Dispatcher，顾名思义就是分发器，主要的作用就是用于处理具体的网络请求，这个请求包括同步请求以及异步请求。

## Dispatcher创建 ##

Dispatcher是在OkHttpClient实例化时一同被实例化的，这个可以在OkHttpClient的内部类Builder中看到，这里不做过多的介绍。

## 相关属性介绍 ##

在说明Dispatcher的工作原理前，首先对Dispatcher类中声明的几个属性坐下说明，如下

/**
     * 设置默认的最大并发请求数量，这个在异步请求时会使用到
     * 当然，通过setMaxRequests可以自定义设置最大并发请求数量
     */
    private int maxRequests = 64;
    /**
     * 同时请求相同主机的请求数量最大值
     * 通过setMaxRequestsPerHost方法可以修改
     */
    private int maxRequestsPerHost = 5;
    
    /**
     * 所有请求结束后的回调方法，默认为null，所以不设置的话不会被执行
     */
    private @Nullable Runnable idleCallback;
    
    // 线程池,用于执行异步请求
    private @Nullable ExecutorService executorService;
    
    
    // 异步请求队列（等待被线程池处理的队列）
    private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();
    
    // 异步请求队列 (正在被处理的请求队列)
    private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();
    
    // 同步请求队列
    private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

## Dispatcher处理同步请求 ##

根据上文的分析可以知道，当发起一个同步请求时，在RealCall.execute()方法中，最后会调用Dispatchr.execute()方法，这里可以看下用Dispatchr.execute()的实现细节。

synchronized void executed(RealCall call) {
        runningSyncCalls.add(call);
      }

Dispatchr.execute()的代码很简单，就是将该请求Call添加到了同步请求队列中，至于同步请求最终的实现就是通过上文中提到的拦截器完成的，这里先不做过多介绍。

在同步请求完成后，将会通过调用Dispatcher.finished()方法，将该请求从请求队列中移除，表示该请求已经处理完成。  
finished()方法将在下文异步请求中分析。

## Dispatcher处理异步请求 ##

同同步请求，当开始一个异步请求时，Dispatcher会调用enqueue(Call)方法。

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

在enqueue(Call)方法中，首先会判断当前并发请求以及对同一个主机的请求是否达到阈值。

*  如果没有，直接将该请求添加到正在运行的异步队列中，并通过线程池执行该请求。
 *  如果达到阈值，那么该请求会被添加到异步请求等待队列中，等待线程池的执行

### ExecutorService线程池 ###

在enqueue()方法中会通过executorService()方法获取一个ExecutorService实例，这里看下这个方法。

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;
    }

从executorService()方法可以看出，被创建的线程池的是一个无限大并发数量的线程池，但是线程池中的线程的keepAliveTime被设置成了60秒，也就是当线程池的空闲线程超过60秒没有执行新的任务时就会被回收。

**这里需要注意的是，虽然这个线程池被设置成无限大并发数量的线程池，但是在该线程池的最大并发数量并不是无限大，而是通过上文中提到的maxRequests属性决定的，这个在finished()方法中可以看到。**

### finished方法 ###

在executorService线程池执行异步请求后，都会通过调用finished方法来从请求队列中移除该请求，表示该请求已经处理完成。

这个在上文中有提到，这里就来看看finished方法的具体实现了。

/**
         * 结束一个异步请求
         * @param call
         */
        void finished(AsyncCall call) {
            finished(runningAsyncCalls, call, true);
        }
    
        /**
         * 结束一个同步请求
         * @param call
         */
        void finished(RealCall call) {
            finished(runningSyncCalls, call, false);
        }
    
        /**
         * 结束一个请求
         * @param calls
         * @param call
         * @param promoteCalls
         * @param <T>
         */
        private <T> void finished(Deque<T> calls, T call, boolean promoteCalls) {
            int runningCallsCount;
            Runnable idleCallback;
            synchronized (this) {
                // 1、首先从请求队列中，移除该请求。
                if (!calls.remove(call)) throw new AssertionError("Call wasn't in-flight!");
    
                // 2、当结束一个异步请求时，还需要通过promoteCalls()方法让线程池继续处理异步请求等待队列中的请求
                if (promoteCalls) promoteCalls();
                runningCallsCount = runningCallsCount();
                idleCallback = this.idleCallback;
            }
    
            // 3、调用空闲回到，一般为null
            if (runningCallsCount == 0 && idleCallback != null) {
                idleCallback.run();
            }
        }

### promoteCalls方法 ###

在上面代码中可知，当通过finished()处理一个异步请求时，会通过调用promoteCalls()方法来让线程池继续处理其他等待队列的请求，这里需要看下promoteCalls()方法的具体实现。

private void promoteCalls() {
            // 1、首先判断并发请求是否超过阈值，这个也就解释了无线大线程池需要通过maxRequests决定最大并发量
            if (runningAsyncCalls.size() >= maxRequests) return; // Already running max capacity.
    
            // 2、如果异步请求等待队列为空，也就是没有其他的请求需要处理，
            if (readyAsyncCalls.isEmpty()) return; // No ready calls to promote.
    
            // 3、执行异步请求等待队列中的请求，通过for循环可知，这里可以同时处理多个等待的请求，只要并发数量不超过maxRequests即可
            for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
                AsyncCall call = i.next();
    
                if (runningCallsForHost(call) < maxRequestsPerHost) {
                    i.remove();
    
                    // 4、将异步请求等待队列中获取一个请求，并通过线程池执行该请求
                    runningAsyncCalls.add(call);
                    executorService().execute(call);
                }
                // 5、如果并发请求数量达到阈值，那么请求仍需要继续等待
                if (runningAsyncCalls.size() >= maxRequests) return; // Reached max capacity.
            }
        }

这样，关于okhttp的Dispatcher请求分发器就讲解到这里了。