OkHttp源码中Dispatcher和connectionPool线程池分析

Posted hymKing

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了OkHttp源码中Dispatcher和connectionPool线程池分析相关的知识,希望对你有一定的参考价值。

在前面《OkHttp原理流程源码分析》中梳理分析了OkHttp中同步发起网络请求和异步发起网络请求,最终的请求分发是有Dispatcher分发器来完成的,Dispatcher分发器的实现中涉及到的线程池的应用代码,并未做分析,今天我们先看第一个线程池应用,Dispatcher中的线程池实现。

一、OkHttp中Dispatcher的线程池实现

异步网络请求最终会被Dispatcher统一管理调度,client.dispatcher().enqueue(new AsyncCall(responseCallback));,enquene()实现了入队列的逻辑。

public final class Dispatcher 
  synchronized void enqueue(AsyncCall call) 
    //正在运行请求中的call数<64&&同一主机地址运行请求中的call数<5
    if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) 
      //添加到运行中的管理队列中
      runningAsyncCalls.add(call);
      //直接调用executeService.execute方法,执行调用,后续就到拦截器流程了
      executorService().execute(call);
     else 
      readyAsyncCalls.add(call);
    
  

这里我们主要分析一下这里的线程池,看一下executorService的实现,同样是在Dispatcher类的源码中:

public final class 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;
  

上述代码中,通过new ThreadPoolExecutor创建,我们看一下这里使用的构造函数中的每个参数:

/**
 * 通过初始化参数和默认拒绝处理器创建一个线程池
 * @param corePoolSize 核心线程数,即使她们是idle状态,也会保留。除非通allCoreThreadTimeOut设置了,才可能销毁。

 * @param maximumPoolSize 在池子中允许的最大的线程数
 * @param keepAliveTime 当线程池中空闲线程数量超过corePoolSize时,多余的线程会在多长时间内
 * 被销毁
 * @param keepAliveTime的时间单位参数
 * @param workQueue 工作队列,在任务执行之前,持有任务集合。只有runable 任务通过excute方法被
 * 提交的才会被持有。
 * @param threadFactory 创建线程池的工厂
 * @throws IllegalArgumentException 非法参数异常         
 * @throws NullPointerException 空指针异常
 */
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory) 
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         threadFactory, defaultHandler);

在以上注释中,我们对创建线程池的构造方法,进行了参数解释。Dispatcher类中调用此构造函数传入的参数分别如下:

  • @param corePoolSize 核心线程数:0;
  • @param maximumPoolSize 在池子中允许的最大的线程数:Integer.MAX_VALUE
  • @param keepAliveTime 非核心线程存活时间:60 秒
  • @param keepAliveTime的时间单位参数 seconds
  • @param workQueue 工作队列:SynchronousQueue
  • @param threadFactory 创建线程池的工厂:Util.threadFactory工厂类

核心线程数为0,线程池中可以创建的线程数为整形的最大值。

工作队列为SynchronousQueue,这个队列的容量不是0就是1。

这样的线程池设计,是不是有问题呢?

理论上来讲,性能极好的情况下,这样的线程池,拥有最高的并发吞吐能力,为什么,你提交多少runable,线程池就立即执行,发现没有空闲的线程,就创建线程执行。

但实际上怎么可能有这么好的性能呢?当然OkHttp的设计者,也并没有不考虑这个问题,在为了让线程池本身具有很好的吞吐能力的同事,OkHttp对框架内部提交的任务,进行了自维护。也就是runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) ;使得最大并发线程就最多不超过64个。

其实这块的代码设计,是比较简单的。为什么要单独拿出来说呢,我觉得这种设计考虑挺好,既然是OkHttp框架自己处理任务的事情,就交给框架自己,另外框架内部还要处理同域名的域名请求的任务的提交,那就有框架自己同一来管理就好了。

这样的设计我觉得职责还是挺清晰的,线程池你就啥呀别管了,我给你你就执行就好了呗。

二、 OkHttp中ConnectionPool连接池的实现

《OkHttp原理流程源码分析》这篇文章中,关于ConnectionInterceptor,我们简单的分析了其主要流程如下:

/** 打开一个到目标服务器的连接以及执行下一个拦截器 */
public final class ConnectInterceptor implements Interceptor 
  public final OkHttpClient client;

  public ConnectInterceptor(OkHttpClient client) 
    this.client = client;
  

  @Override public Response intercept(Chain chain) throws IOException 
    RealInterceptorChain realChain = (RealInterceptorChain) chain;
    Request request = realChain.request();
    StreamAllocation streamAllocation = realChain.streamAllocation();

    // We need the network to satisfy this request. Possibly for validating a conditional GET.
    boolean doExtensiveHealthChecks = !request.method().equals("GET");
    
    //关键代码
    HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks);
    //创建一个链接
    RealConnection connection = streamAllocation.connection();
    return realChain.proceed(request, streamAllocation, httpCodec, connection);
  

HttpCodec是一个编解码请求和请求和响应的接口,HttpCodec接口的实现分别对应Http1Codec、Http2Codec。httpCodec实例对象是通过streamAllocation.newStream(client, chain, doExtensiveHealthChecks);来获得的。

同时通过streamAllocation获得的是一个RealConnection的对象。

这个拦截器的作用就是打开一个和Server的连接,并HttpCodec、RealConnection实例传递给下一个拦截器CallServerInterceptor,发起执行网络请求,可以理解成ConnectionInterceptor的核心作用就是为真实的网络请求做最直接的准备工作。

今天主要分析两个问题:

1、socket的连接池是如何创建的?

2、socket的连接池的复用原理?

2.1 OkHttp中连接复用

创建一个链接池的核心代码:

HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks);
RealConnection connection = streamAllocation.connection();

streamAllocation的实例对象的创建,是在RetryAndFollowUpInterceptor拦截器中创建的,并传递到了ConnectionInterceptor中。接着我们继续看newStream(创建一个新流对象):

public final class StreamAllocation 
  public HttpCodec newStream(OkHttpClient client, Chain chain, boolean doExtensiveHealthChecks) 
       ....
        try 
            //这里创建了真实的连接
            RealConnection resultConnection = this.findHealthyConnection(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis, connectionRetryEnabled, doExtensiveHealthChecks);
            HttpCodec resultCodec = resultConnection.newCodec(client, chain, this);
            synchronized(this.connectionPool) 
                this.codec = resultCodec;
              //这是连接已HttpCodec的实例返回
                return resultCodec;
            
         ...
    

还没有看到connectionPool的创建,继续看this.findHealthyConnection()方法的源码实现:

private RealConnection findHealthyConnection(int connectTimeout, int readTimeout, int writeTimeout, int pingIntervalMillis, boolean connectionRetryEnabled, boolean doExtensiveHealthChecks) throws IOException 
    while(true) //死循环,完全阻塞式查找
      //候选连接查找
        RealConnection candidate = this.findConnection(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis, connectionRetryEnabled);
        synchronized(this.connectionPool) 
            if (candidate.successCount == 0) 
              //如果是全新的连接,直接返回,可以查看successCount理解这段代码的含义
                return candidate;
            
        
        //非全新的,要看当前候选连接是否可用了
        if (candidate.isHealthy(doExtensiveHealthChecks)) 
            return candidate;
        
        //如果不可用,继续查找
        this.noNewStreams();
    

候选连接查找的实现,在findConnection()方法的实现中,继续看源码:

private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout, int pingIntervalMillis, boolean connectionRetryEnabled) throws IOException 
   ....
    //第一步:查找当前连接是否可用,如果可用直接返回。
    synchronized(this.connectionPool) 
        ...
        //检查当前有没有可用连接,如果有就用
        releasedConnection = this.connection;
        toClose = this.releaseIfNoNewStreams();
        if (this.connection != null) 
            result = this.connection;
            releasedConnection = null;
        

        //当前可用连接为空
        if (result == null) 
            //从连接池中查找可用连接
            Internal.instance.get(this.connectionPool, this.address, this, (Route)null);
            if (this.connection != null) 
              //非空,表示找到可用连接,赋值给查找结果result
                foundPooledConnection = true;
                result = this.connection;
             else 
                // 空,备选路由使用当前路由
                selectedRoute = this.route;
            
        
    
    //第二步:选择新的路由,继续查找可用连接(从连接池中的查找,实际上可能经历两次,进行了路由切换查找)
    ...
    //无论通过上述哪种方式,如果找到了就返回。
    if (result != null) 
        return result;
     else 
        //否则
        // 新路由选择先置为false
        boolean newRouteSelection = false;
        // 如果条件满足,选择一条新路由
        if (selectedRoute == null && (this.routeSelection == null || !this.routeSelection.hasNext())) 
            newRouteSelection = true;
            this.routeSelection = this.routeSelector.next();
        
        //锁住连接池(连接池的创建还没看到)
        synchronized(this.connectionPool) 
          ...
            if (newRouteSelection) 
                List<Route> routes = this.routeSelection.getAll();
                int i = 0;

                for(int size = routes.size(); i < size; ++i) 
                  // 在此路由下根据ip集合去匹配连接池中的连接
                  // Route的构造函数:Route(Address address, Proxy proxy, InetSocketAddress inetSocketAddress) 
                    Route route = (Route)routes.get(i);
                    Internal.instance.get(this.connectionPool, this.address, this, route);
                    if (this.connection != null) 
                        foundPooledConnection = true;
                        //找到赋值给结果result,并终止遍历查找
                        result = this.connection;
                        this.route = route;
                        break;
                    
                
            
            // 还没找到
            if (!foundPooledConnection) 
                if (selectedRoute == null) 
                    selectedRoute = this.routeSelection.next();
                

                this.route = selectedRoute;
                this.refusedStreamCount = 0;
                //只能创建连接
                result = new RealConnection(this.connectionPool, selectedRoute);
                this.acquire(result, false);
            
        

        if (foundPooledConnection) 
            this.eventListener.connectionAcquired(this.call, result);
            return result;
         else 
          
          //执行新创建连接连接
            result.connect(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis, connectionRetryEnabled, this.call, this.eventListener);
            this.routeDatabase().connected(result.route());
            //创建socket
            Socket socket = null;
            synchronized(this.connectionPool) 
                this.reportedAcquired = true;
                //新创建的连接要记录到连接池
                Internal.instance.put(this.connectionPool, result);
                if (result.isMultiplexed()) 
                    //内部重复的socket连接去重,就是释放旧的连接,保留新的连接
                    socket = Internal.instance.deduplicate(this.connectionPool, this.address, this);
                    result = this.connection;
                
            

            Util.closeQuietly(socket);
            this.eventListener.connectionAcquired(this.call, result);
            return result;
        
    

这部分的源码实现有点多,在源码中已经添加了部分注释,接下来,我们梳理一下连接复用和创建的简单流程图。

查找可复用的连接的流程梳理就如上图,在从连接池中的查找细节中,从连接池查找连接时会查询两次,第一次只是根据当前目标服务器地址去查,如果没有查到,则第二次会重新选择路由表,然后用该地址去匹配。

连接的复用流程是完事了,在连接复用的源码流程,我们看到了多处 this.connectionPool,这个理论上就应该是我们的连接池。那这个连接池是在哪儿创建的呢。

2.2 OkHttp中的连接池的创建

this.connectionPool,这个显然是一个连接池,接下来我们就分析OkHttp网络请求框架中的连接池的本身到底是啥,查看connectionPool()的源码实现。

2.2.1 连接池实现

连接池的本质就是通过Deque connections 这样一个双端队列维护的connecitons的容器。

public final class ConnectionPool 
  //最大的空闲连接数--每个地址的最大空闲连接数
  private final int maxIdleConnections;
  //连接持续时间
  private final long keepAliveDurationNs;
  ...
  /**维护线程连接数的双端队列*/
  private final Deque<RealConnection> connections = new ArrayDeque<>();
  /**
   * 根据参数创建一个连接池来满足一个单一的用户进程。
   * 持有5个连接,每个连接的存活时间是5分钟
   */
  public ConnectionPool() 
    // 默认实现
    this(5, 5, TimeUnit.MINUTES);
  
  // 有参数的构造,说明也可以做定制实现
  public ConnectionPool(int maxIdleConnections, long keepAliveDuration, TimeUnit timeUnit) 
    this.maxIdleConnections = maxIdleConnections;
    this.keepAliveDurationNs = timeUnit.toNanos(keepAliveDuration);

    // Put a floor on the keep alive duration, otherwise cleanup will spin loop.
    if (keepAliveDuration <= 0) 
      throw new IllegalArgumentException("keepAliveDuration <= 0: " + keepAliveDuration);
    
  
  ......
  //添加一个真实的连接
  void put(RealConnection connection) 
    assert (Thread.holdsLock(this));
    //启动清理runable,也交给当前线程池进行处理
    if (!cleanupRunning) 
      cleanupRunning = true;
      //清理线程池
      executor.execute(cleanupRunnable);
    
    connections.add(connection);
  
  为了方便查看,以下部分代码调整了这段代码在实际源码的位置
  private final Runnable cleanupRunnable = new Runnable() 
    @Override public void run() 
      //清理线程,通过死循环进行连接的清理
      while (true) 
        //执行清理并返回下次需要清理的时间
        long waitNanos = cleanup(System.nanoTime());
        if (waitNanos == -1) return;
        if (waitNanos > 0) 
          long waitMillis = waitNanos / 1000000L;
          waitNanos -= (waitMillis * 1000000L);
          synchronized (ConnectionPool.this) 
            try 
              ConnectionPool.this.wait(waitMillis, (int) waitNanos);
             catch (InterruptedException ignored) 
            
          
        
      
    
  ;

在构建连接池的时候,限制了最大的空闲连接数只能是5个,每个空闲连接数的存活时间是5分钟,在每次添加一个RealConnection实例对象到连接池中的时候,会立即检查清理线程是否在运行,如果不在运行,则启动清理检查的runable的循环执行,根据清理策略进行清理。

2.2.2 清理线程池实现
public final class ConnectionPool 
  /**
   * 线程池实现
   */
  private static final Executor executor = new ThreadPoolExecutor(0 /* corePoolSize */,
      Integer.MAX_VALUE /* maximumPoolSize */, 60L /* keepAliveTime */, TimeUnit.SECONDS,
      new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp ConnectionPool", true));

这个线程池的构造参数如下:

corePoolSize:0;maximumPoolSize:Integer.MAX_VALUE;非核心线程的存活时间:60s;

workQueue:SynchronousQueue(),无容量的工作队列。

线程创建工厂:Util.threadFactory(“OkHttp ConnectionPool”, true)) ,是否守护为true;

这个线程池的实现和Dispatcher线程池的实现基本一致,上面的这种线程池的设计,是让线程池本身具备最大的吞吐能力。

2.2.3 再看连接(RealConnection的复用)

获取一个连接(RealConnection)的源码实现如下:

@Nullable RealConnection get(Address address, StreamAllocation streamAllocation, Route route) 
  assert (Thread.holdsLock(this));
  for (RealConnection connection : connections) 
    // 取可用连接的核心判断逻辑
    if (connection.isEligible(address, route)) 
      streamAllocation.acquire(connection, true);
      return connection;
    
  
  /**是否合格的可用连接的判断*/
  public boolean isEligible(Address address, @Nullable Route route) 
        if (this.allocations.以上是关于OkHttp源码中Dispatcher和connectionPool线程池分析的主要内容,如果未能解决你的问题,请参考以下文章

OkHttp源码中Dispatcher和connectionPool线程池分析

OkHttp源码中Dispatcher和connectionPool线程池分析

OkHttp源码中Dispatcher和connectionPool线程池分析

深入理解OkHttp源码——提交请求

okhttp异步请求流程和源码分析

Android:OkHttp请求分发器 Dispatcher的理解和使用