Okhttp的线程池

2021-01-04

2021年第一篇文章~~

线程池策略

当一个任务通过execute(Runnable)方法添加到线程池时：

线程数量小于corePoolSize，新建线程数来处理被添加的任务。
线程数量大于等于corePoolSize，存在空闲线程，使用空闲线程执行新任务。
线程数量大于等于corePoolSize，不存在空闲线程，新任务被添加到等待队列，添加成功则等待空闲线程；
添加失败：线程数量小于最大线程数，新建线程执行新任务；线程数量等于最大线程数，则拒绝此任务。

Okhttp线程池配置

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

可以看出，创建的线程池核心线程数为0，最大线程数MAX_VALUE，闲置时间60s，队列SynchronousQueue。这样的参数配置出来的线程池高并发，最大吞吐量。

那么为什么Okhttp的线程池队列用的SynchronousQueue而不是其他的？我们先看个栗子：

public void testQueue(){
    //SynchronousQueue<Runnable> queue=new SynchronousQueue<>();
    //LinkedBlockingDeque<Runnable> queue=new LinkedBlockingDeque<>();
    ArrayBlockingQueue<Runnable> queue=new ArrayBlockingQueue<Runnable>(1);


    ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
            queue);

    poolExecutor.execute(new Runnable() {
        @Override
        public void run() {
            System.out.print("任务1");
            while (true){

            }
        }
    });

    poolExecutor.execute(new Runnable() {
        @Override
        public void run() {
            System.out.print("任务2");
        }
    });

    while (true){

    }
}

这里用的ArrayBlockingQueue，传的capacity为1。如果任务1是个耗时的操作，那么任务2得不到执行。我们可以看一下ThreadPoolExecutor的源码：

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    
    int c = ctl.get();
    //corePoolSize为0，这里不用看了，进不来
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    
    //当任务1来了，添加到队列
    //当任务2来了，任务1已经从队列出来了，所以可以添加到队列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
            
        //如果当前线程池线程空，则添加一个线程，所以任务1执行
        //但是任务2来了，不满足这个判断条件，所以任务2得不到执行
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}

那么我再加一个任务3呢？

public void testQueue(){
    //SynchronousQueue<Runnable> queue=new SynchronousQueue<>();
    //LinkedBlockingDeque<Runnable> queue=new LinkedBlockingDeque<>();
    ArrayBlockingQueue<Runnable> queue=new ArrayBlockingQueue<Runnable>(1);


    ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
            queue);

    poolExecutor.execute(new Runnable() {
        @Override
        public void run() {
            System.out.print("任务1");
            while (true){

            }
        }
    });
    
    try {
        Thread.sleep(100);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    poolExecutor.execute(new Runnable() {
        @Override
        public void run() {
            System.out.print("任务2");
        }
    });
    
    poolExecutor.execute(new Runnable() {
        @Override
        public void run() {
            System.out.print("任务3");
        }
    });

    while (true){

    }
}

这次可以执行了，不过得到的执行顺序为：任务1、任务3、任务2。为什么呢？因为当任务3进来了后添加不到队列了，而此时线程数量小于最大线程数，所以新建线程执行了任务3，当任务3执行完后会从队列里取任务2执行。

所以用ArrayBlockingQueue存在问题，哪怕把capacity改为别的数(除了0)，也存在这样的问题。

那么LinkedBlockingDeque呢？LinkedBlockingDeque内部用的是链表结构，ArrayBlockingQueue用的是数组。LinkedBlockingDeque可以传capacity，也可以不传，不传的话有默认的为Integer.MAX_VALUE。它的效果其实是跟ArrayBlockingQueue一样的，也是有上述的问题。

而SynchronousQueue是一个没有容量的队列，往里面添加永远失败，会开一个线程去执行。(当然了是在线程数量小于最大线程数的情况下，不过这个应该可以保证啦)不会因为谁把谁堵住而得不到执行。