OKHttp3--Dispatcher分发器实现同步异步请求源码解析【三】

爱被打了一巴掌 2022-03-22 14:17 276阅读 0赞

### Dispatcher ###

*  系列
 *  回顾
 *  Dispatcher
 *   *  数据结构
     *  线程池
     *  请求管理
 *  源码

# 系列 #

[OKHttp3–详细使用及源码分析系列之初步介绍【一】][OKHttp3]  
[OKHttp3–流程分析 核心类介绍 同步异步请求源码分析【二】][OKHttp3_ _]  
[OKHttp3–Dispatcher分发器源码解析【三】][OKHttp3_Dispatcher]  
[OKHttp3–调用对象RealCall源码解析【四】][OKHttp3_RealCall]  
[OKHttp3–拦截器链RealInterceptorChain源码解析【五】][OKHttp3_RealInterceptorChain]  
[OKHttp3–重试及重定向拦截器RetryAndFollowUpInterceptor源码解析【六】][OKHttp3_RetryAndFollowUpInterceptor]  
[OKHttp3–桥接拦截器BridgeInterceptor源码解析及相关http请求头字段解析【七】][OKHttp3_BridgeInterceptor_http]  
[OKHttp3–缓存拦截器CacheInterceptor源码解析【八】][OKHttp3_CacheInterceptor]  
[OKHttp3-- HTTP缓存机制解析 缓存处理类Cache和缓存策略类CacheStrategy源码分析 【九】][OKHttp3-- HTTP_ _Cache_CacheStrategy_]

# 回顾 #

在上一篇解析OKHttp同步异步请求文章中，已经大概介绍了Dispatcher这个类，它在OKHttp中的地位非常重要，一定要对它的作用理解透；作为整个框架的任务调度员，我们会有一些疑问，比如

*  Dispatcher到底是什么，干啥用的？
    
        在OKHttp中Dispatcher就是用来保存并管理用户产生的同步请求（RealCall）和异步请求（AsyncCall），
        并负责执行AsyncCall
 *  OKHttp是如何实现同步以及异步请求的？
    
        没有别的，就是靠Dispatcher将我们发送的请求进行管理
        
        在同步请求时，将请求添加到runningSyncCalls中（正在执行的同步请求队列）；
        在请求结束时从队列移除该请求
        
        在异步请求时，将封装成AsyncCall的异步请求线程根据条件添加到runningAsyncCalls队列（正在执行的异步请求队列）
        和readyAsyncCalls队列（等待执行的异步请求队列），然后执行AsyncCall；在请求结束时，从队列移除，
        并调整两个队列的元素

# Dispatcher #

这里我们先不摆源码，先说它有什么特点，让大家有个总体上的认识

## 数据结构 ##

在Dispatcher中维护了三个队列，队列类型是ArrayDeque，这是一个双端队列，即可以从队列头部和尾部进行数据操作，能够实现FIFO原则，即先进去的可以先执行，不过不是线程安全的实现，多线程环境中需要加锁；看过AsyncTask源码的肯定知道它，具体细节可参考[从源码解析-Android数据结构之双端队列ArrayDeque 实现FIFO和LIFO队列][-Android_ArrayDeque _FIFO_LIFO]

*  第一个队列是runningSyncCalls，是一个正在执行的同步请求队列，所有我们添加的同步请求都会被添加到这里面，包括已被取消但没执行完的请求，队列泛型是RealCall对象
 *  第二个队列是runningAsyncCalls，是一个正在执行的异步请求队列，所有我们添加的异步请求都会被添加到这里，包括已被取消但没执行完的请求，队列泛型是AsyncCall对象，实现了Runnable接口
 *  第三个队列是readyAsyncCalls，是一个等待执行的异步请求队列，什么意思呢？因为OKHttp允许同时执行的异步请求数量必须在64个以内，且单个host同时执行的最大请求数量在5个以内，所以当我们添加的异步请求数超过它，该请求就会被添加到该队列，等runningAsyncCalls队列有空闲位置后添加到里面
    
    对于单个host同时执行的最大请求数量不理解的童靴，这里说明下，host在这里指的是hostname，即主机名（代表一个主机，每个主机都有一个唯一标识，即ip地址，但是每个主机的主机名并不一定是唯一的），你可以理解为同时往同一个服务器上发送的请求数量不能超过5个；不过OKHttp是通过URL的主机名判断的，所以对同一个ip地址的并发请求仍然可能会超过5个，因为多个主机名可能共享同一个ip地址或者路由（相同的HTTP代理）

有的人可能说了，用这些队列有什么用呢？好处在哪呢？

要知道通过这些队列，OKHttp可以轻松的实现并发请求，更方便的维护请求数以及后续对这些请求的操作（比如取消请求），大大提高网络请求效率；同时可以更好的管理请求数，防止同时运行的线程过多，导致OOM，同时限制了同一hostname下的请求数，防止一个应用占用的网络资源过多，优化用户体验

## 线程池 ##

OKHttp在其内部维护了一个线程池，用于执行异步请求AsyncCall，其构造如下

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

我们需要关注的就是ThreadPoolExecutor前三个参数

*  第一个是0，说明该线程池没有核心线程，所有线程都是工作线程，即所有线程超过一定空闲时间会被回收
 *  第二个参数是Integer.MAX\_VALUE，即最大线程数，虽然设置值这么大，但是无须担心性能消耗过大问题，因为有队列去维护请求数
 *  第三个参数是60，即工作线程空闲60s后就会被回收

关于线程池更多信息可参考[Android开发-通过ExecutorService构建一个APP使用的全局线程池][Android_-_ExecutorService_APP]

## 请求管理 ##

private int maxRequests = 64;
    private int maxRequestsPerHost = 5;

在类里定义了这两个变量，其含义是默认支持的最大并发请求数量是64个，单个host并发请求的最大数量是5个；这两个值是可以通过后续设置进行更改的，并且这个要求只针对异步请求，对于同步请求数量不做限制

当异步请求进入Dispatcher中，如果满足上面两个数量要求，该请求会被添加到runningAsyncCalls中，然后执行它；如果不满足就将其添加到readyAsyncCalls中；当一个异步请求结束时，会遍历readyAsyncCalls队列，再进行条件判断，符合条件就将请求从该队列移到runningAsyncCalls队列中并执行它

不管是同步请求还是异步请求最终执行完后都会从队列中移除

除了添加和移除，OKHttp还支持用户取消添加过的请求，可以全部取消，即将队列清空；也可以取消某一个请求

# 源码 #

最后我们通过源码来了解Dispatcher

public final class Dispatcher {
      //最大并发请求数
      private int maxRequests = 64;
      //单个主机最大并发请求数
      private int maxRequestsPerHost = 5;
      
      private Runnable idleCallback;
    
      /** 执行AsyncCall的线程池 */
      private ExecutorService executorService;
    
      /** 等待执行的异步请求队列. */
      private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();
    
      /** 正在执行的异步请求队列，包含已取消但为执行完的请求 */
      private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();
    
      /** 正在执行的同步请求队列，包含已取消但为执行完的请求 */
      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(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
              new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
        }
        return executorService;
      }
    
      /**
       * 设置并发执行的最大请求数
       */
      public synchronized void setMaxRequests(int maxRequests) {
        if (maxRequests < 1) {
          throw new IllegalArgumentException("max < 1: " + maxRequests);
        }
        this.maxRequests = maxRequests;
        promoteCalls();
      }
    
      public synchronized int getMaxRequests() {
        return maxRequests;
      }
    
      /**
       * 设置每个主机同时执行的最大请求数
       */
      public synchronized void setMaxRequestsPerHost(int maxRequestsPerHost) {
        if (maxRequestsPerHost < 1) {
          throw new IllegalArgumentException("max < 1: " + maxRequestsPerHost);
        }
        this.maxRequestsPerHost = maxRequestsPerHost;
        promoteCalls();
      }
    
      public synchronized int getMaxRequestsPerHost() {
        return maxRequestsPerHost;
      }
    
      /**
       * 当分发器处于空闲状态下，即没有正在运行的请求，设置回调
       */
      public synchronized void setIdleCallback(Runnable idleCallback) {
        this.idleCallback = idleCallback;
      }
    
      /**
       * 执行异步请求
       * 当正在执行的异步请求数量小于64且单个host正在执行的请求数量小于5的时候，就执行该请求，并添加到队列
       * 否则添加到等待队列中
       */
      synchronized void enqueue(AsyncCall call) {
        if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) {
          runningAsyncCalls.add(call);
          executorService().execute(call);
        } else {
          readyAsyncCalls.add(call);
        }
      }
    
      /**
       * 取消所有请求
       */
      public synchronized void cancelAll() {
        for (AsyncCall call : readyAsyncCalls) {
          call.get().cancel();
        }
    
        for (AsyncCall call : runningAsyncCalls) {
          call.get().cancel();
        }
    
        for (RealCall call : runningSyncCalls) {
          call.cancel();
        }
      }
    
      //调整请求队列，将等待队列中的请求放入正在请求的队列
      private void promoteCalls() {
        if (runningAsyncCalls.size() >= maxRequests) return; // 如果正在执行请求的队列已经满了，那就不用调整了.
        if (readyAsyncCalls.isEmpty()) return; // 如果等待队列是空的，也不需要调整
    
        //遍历等待队列
        for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
          AsyncCall call = i.next();
          //单个host正在执行的请求数量小于5的时候，将该请求添加到runningAsyncCalls中并执行它
          //同时从等待队列中删除它
          if (runningCallsForHost(call) < maxRequestsPerHost) {
            i.remove();
            runningAsyncCalls.add(call);
            executorService().execute(call);
          }
    
          if (runningAsyncCalls.size() >= maxRequests) return; // 如果正在执行请求的队列已经满了，就退出循环
        }
      }
    
      /** 返回单个host的请求数 */
      private int runningCallsForHost(AsyncCall call) {
        int result = 0;
        for (AsyncCall c : runningAsyncCalls) {
          if (c.host().equals(call.host())) result++;
        }
        return result;
      }
    
      /** 执行同步请求，只是将其添加到队列中 */
      synchronized void executed(RealCall call) {
        runningSyncCalls.add(call);
      }
    
      /** 异步请求执行完成调用. */
      void finished(AsyncCall call) {
        finished(runningAsyncCalls, call, true);
      }
    
      /** 同步请求执行完成调用. */
      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();
        }
      }
    
      /** 返回当前正在等待执行的异步请求的快照. */
      public synchronized List<Call> queuedCalls() {
        List<Call> result = new ArrayList<>();
        for (AsyncCall asyncCall : readyAsyncCalls) {
          result.add(asyncCall.get());
        }
        return Collections.unmodifiableList(result);
      }
    
      /** 返回当前正在执行的异步请求的快照. */
      public synchronized List<Call> runningCalls() {
        List<Call> result = new ArrayList<>();
        result.addAll(runningSyncCalls);
        for (AsyncCall asyncCall : runningAsyncCalls) {
          result.add(asyncCall.get());
        }
        return Collections.unmodifiableList(result);
      }
    
      //返回等待执行的异步请求数量
      public synchronized int queuedCallsCount() {
        return readyAsyncCalls.size();
      }
    
      //计算正在执行的请求数量
      public synchronized int runningCallsCount() {
        return runningAsyncCalls.size() + runningSyncCalls.size();
      }
    }

可以说Dispatcher的源码比较简单，不难理解，需要把握住关键，它的作用就是负责对用户的请求进行管理，保存，备份，重点在于使用线程池执行异步请求，对于同步请求基本不做多少处理

[OKHttp3]: https://blog.csdn.net/qq_30993595/article/details/82079586
[OKHttp3_ _]: https://blog.csdn.net/qq_30993595/article/details/86438864
[OKHttp3_Dispatcher]: https://blog.csdn.net/qq_30993595/article/details/86681210
[OKHttp3_RealCall]: https://blog.csdn.net/qq_30993595/article/details/86686105
[OKHttp3_RealInterceptorChain]: https://blog.csdn.net/qq_30993595/article/details/87475028
[OKHttp3_RetryAndFollowUpInterceptor]: https://blog.csdn.net/qq_30993595/article/details/87523713
[OKHttp3_BridgeInterceptor_http]: https://blog.csdn.net/qq_30993595/article/details/87738481
[OKHttp3_CacheInterceptor]: https://blog.csdn.net/qq_30993595/article/details/87886518
[OKHttp3-- HTTP_ _Cache_CacheStrategy_]: https://blog.csdn.net/qq_30993595/article/details/87941893
[-Android_ArrayDeque _FIFO_LIFO]: https://blog.csdn.net/qq_30993595/article/details/80799444
[Android_-_ExecutorService_APP]: https://blog.csdn.net/qq_30993595/article/details/84324681