Java线程池总结

2020-04-21

线程池对于开发者来说应该是一个很熟悉的东西了，在处理多并发的任务时其能提供便利的线程复用和任务调度能力，Java上的线程池封装得也很简单。只是没想到在多年后的今天居然栽在线程池的配置上，还差点导致问题，这我不得不让我反思自己的基础技术的掌握能力是否及格。

0x0 一般使用

在Java中我们要使用线程池很简单，直接调用Execotors的静态方法就可以获取一个线程池实例了，如创建一个无上限线程数的线程池只需要调用

1	Executors.newCachedThreadPool()

Executors内置了好几种不同模式的线程池形式，以用于不同的使用场合，如此简单。

1
2
3

Executors.newFixedThreadPool(3);//定义一个有三个线程的线程池
Executors.newSingleThreadExecutor();//定义有一个线程池的线程池
Execurots.newCachedThreadPool();//定义无上限的线程池

0x1 默认模式

拆开Executors的常用方法来看，这几种模式其实无非是对ThreadPoolExecutor配置一些默认参数，如newCachedThreadPool的实现为

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

选取ThradPoolExecutor参数最完整的构造方法如下

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {

其中包括了当前核心线程数，最大线程数，任务队列和线程的生命周期等,ThreadFactory可以自定义生成Thread时的一些过程，如是否是守护线程，线程的名称自定义等等。

对于一般使用者而言，更多关心的是核心与最大线程数以及任务队列之间的关系，它们之间的组合直接会影响到线程池最终的表现。

Execotor其实是一个Manager的角色，用于调度和控制线程生成回收，任务派发等；
corePoolSize: 核心线程数一般定义为这个线程池里常驻的线程数，也就是不会因为被停止的线程，用于一些任务数比较固定的场合会比较合适；
maximumPoolSize: 最大的线程数一般定义为这个线程池里允许最多能同时存活多少个线程，一般这个数会比核心的多，当任务超过核心线程数但未走过最大线程数后，就会在线程池内动态创建线程用于新任务执行。
workQueue：任务队列，会决定任务调试策略与任务缓存的
RejectedExecutionHandler：线程池中资源全部占用，提交任务失败后的回调

理论上应该是这样子的，但！是！这不是绝对的，中间一个比较关键的变量 —— 任务队列, 可能会导致代码运行情况跟预期不一样。

0x2 任务队列&&排队策略

如上所描述，一般我所理解的线程池调度都是核心线程先用，不够了就放队列里，没超过最大线程数就开新线程执行，但这个是不一定的，任务队列的差异会导致任务的调度表现有所不同。

结合例子先来看有哪些队列，以及它们的表现(先以BlockingQueue占坑）。


public class Main {
    private static Executor executor;

    public static void main(String[] args) {
        executor = new ThreadPoolExecutor(2, 10, 30L, TimeUnit.SECONDS, new BlockingQueue<Runnable>());

        for (int i = 0; i < 10; i++) {
            executor.execute(new Task(i));
        }

        System.out.println("start");
    }

    public static class Task implements Runnable {

        private int index;

        public Task(int i) {
            super();
            index = i;
        }

        @Override
        public void run() {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println("" + index+", thread="+Thread.currentThread()+", time="+System.currentTimeMillis());

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

        }
    }
}

0x21 LinkedBlockingQueue

LikedBlockingQueue是一个无界的有序缓存等待队列，可以将任务丢到队列中等待，当线程池已经满的时候新增加的任务就会被放入等待队列中等待调度，先来先处理。

在上述例子中，按照以往的理解，因为核心线程数是2，最大线程数是10，那同时开启的话应该会有10个线程，等待队列应该是为空的，且先看一次执行的输出

1, thread=Thread[pool-1-thread-2,5,main], time=1595145659355
0, thread=Thread[pool-1-thread-1,5,main], time=1595145659355
2, thread=Thread[pool-1-thread-2,5,main], time=1595145663361
3, thread=Thread[pool-1-thread-1,5,main], time=1595145663361
4, thread=Thread[pool-1-thread-2,5,main], time=1595145667371
5, thread=Thread[pool-1-thread-1,5,main], time=1595145667371
6, thread=Thread[pool-1-thread-2,5,main], time=1595145671378
7, thread=Thread[pool-1-thread-1,5,main], time=1595145671378
9, thread=Thread[pool-1-thread-1,5,main], time=1595145675384
8, thread=Thread[pool-1-thread-2,5,main], time=1595145675384

很明显不是，线程池最终只开启了两个核心线程，剩下的新提交的任务都被放到了等待队列中等待了。也就是说，maximumPoolSize这个参数在这种场景下没发挥作用。

再看maximumPoolSize=2时的情况

size=0
size=0
size=1
size=2
size=3
size=4
size=5
size=6
size=7
size=8
start
0, thread=Thread[pool-1-thread-1,5,main], time=1595156453471
1, thread=Thread[pool-1-thread-2,5,main], time=1595156453475
2, thread=Thread[pool-1-thread-1,5,main], time=1595156457477
3, thread=Thread[pool-1-thread-2,5,main], time=1595156457482
4, thread=Thread[pool-1-thread-1,5,main], time=1595156461485
5, thread=Thread[pool-1-thread-2,5,main], time=1595156461490
6, thread=Thread[pool-1-thread-1,5,main], time=1595156465493
7, thread=Thread[pool-1-thread-2,5,main], time=1595156465499
8, thread=Thread[pool-1-thread-1,5,main], time=1595156469495
9, thread=Thread[pool-1-thread-2,5,main], time=1595156469507

所以最终除了核心线程数，也只是会直接添加到队列中去。

需要注意的是，LinkedBlockingQueue虽然说是无界的，但它的无界只是说在默认情况下容量为Integer.MAX_VALUE而已，当然也是可以自己指定它的容量大小的，如果添加队列失败，核心线程又已经满了，那最后就会导致添加任务失败，被reject了。

1	LinkedBlockingQueue queue=new LinkedBlockingQueue(2);

限制queue大小为2时的结果

4, thread=Thread[pool-1-thread-3,5,main], time=1595154384425
8, thread=Thread[pool-1-thread-7,5,main], time=1595154384425
5, thread=Thread[pool-1-thread-4,5,main], time=1595154384425
0, thread=Thread[pool-1-thread-1,5,main], time=1595154384425
7, thread=Thread[pool-1-thread-6,5,main], time=1595154384425
1, thread=Thread[pool-1-thread-2,5,main], time=1595154384425
9, thread=Thread[pool-1-thread-8,5,main], time=1595154384425
6, thread=Thread[pool-1-thread-5,5,main], time=1595154384425
2, thread=Thread[pool-1-thread-3,5,main], time=1595154388433
3, thread=Thread[pool-1-thread-5,5,main], time=1595154388433

可见同时开了8个线程，workQueue.offer里添加失败了，只放进了两了，减去2个核心的，剩下6个全部都是走了最后的一个addWorker的判断，因为任务最终还是少于maximumPoolSize,所以另外开了6个线程，总共8个线程在并行。

当maximumPoolSize小于总任务数时，就reject了。

0x22 SynchronousQueue

SynchronousQueue其实并不会缓存任何元素，调用size方法永远返回0，如果总任务数满足最大任务数，直接就提交给线程池去创建新线程了。

1, thread=Thread[pool-1-thread-2,5,main], time=1595155069288
4, thread=Thread[pool-1-thread-5,5,main], time=1595155069288
8, thread=Thread[pool-1-thread-9,5,main], time=1595155069288
7, thread=Thread[pool-1-thread-8,5,main], time=1595155069288
5, thread=Thread[pool-1-thread-6,5,main], time=1595155069288
0, thread=Thread[pool-1-thread-1,5,main], time=1595155069288
2, thread=Thread[pool-1-thread-3,5,main], time=1595155069288
3, thread=Thread[pool-1-thread-4,5,main], time=1595155069288
6, thread=Thread[pool-1-thread-7,5,main], time=1595155069288
9, thread=Thread[pool-1-thread-10,5,main], time=1595155069288

线程都是同一时间启动的，且创建了10个线程。

当maximumPoolSize少于任务数时，就reject了。

0x23 ArrayBlockingQueue

ArrayBlockingQueue必须要指定大小capacity，意为有界。

它的表现需要分情况来讨论

当capacity+核心线程数>=总任务数时，除了核心线程后的其他任务都会被放到队列中缓存，但最终只会开启核心线程，其他线程任务排队执行

size=0
size=0
size=1
size=2
size=3
size=4
size=5
size=6
size=7
size=8
start
1, thread=Thread[pool-1-thread-2,5,main], time=1595155410120
0, thread=Thread[pool-1-thread-1,5,main], time=1595155410120
2, thread=Thread[pool-1-thread-1,5,main], time=1595155414128
3, thread=Thread[pool-1-thread-2,5,main], time=1595155414128
4, thread=Thread[pool-1-thread-2,5,main], time=1595155418136
5, thread=Thread[pool-1-thread-1,5,main], time=1595155418136
7, thread=Thread[pool-1-thread-1,5,main], time=1595155422145
6, thread=Thread[pool-1-thread-2,5,main], time=1595155422145
8, thread=Thread[pool-1-thread-1,5,main], time=1595155426153
9, thread=Thread[pool-1-thread-2,5,main], time=1595155426153

当capacity+核心线程数<总任务数时，核心线程优化会满足，另外的capacity个线程会进入到队列，剩余的会直接新建线程直接执行。

以下为capacity=2时的运行情况

size=0
size=0
size=1
size=2
size=2
size=2
size=2
size=2
size=2
size=2
start
0, thread=Thread[pool-1-thread-1,5,main], time=1595155863188
1, thread=Thread[pool-1-thread-2,5,main], time=1595155863189
9, thread=Thread[pool-1-thread-8,5,main], time=1595155863189
7, thread=Thread[pool-1-thread-6,5,main], time=1595155863189
5, thread=Thread[pool-1-thread-4,5,main], time=1595155863189
8, thread=Thread[pool-1-thread-7,5,main], time=1595155863189
4, thread=Thread[pool-1-thread-3,5,main], time=1595155863189
6, thread=Thread[pool-1-thread-5,5,main], time=1595155863189
3, thread=Thread[pool-1-thread-1,5,main], time=1595155867196
2, thread=Thread[pool-1-thread-2,5,main], time=1595155867196

可见开启了8个线程，剩下的两个任务用复用原来的线程，但会后置执行。

0x3 总结

综上可知，使用哪一种BlockedQueue都好，ThreadPoolExecutor都会遵循以下逻辑执行

优先使用核心线程去执行
核心线程跑满，优先添加到队列中去
如果队列容量不够，则直接新创建线程执行
如果目前线程池总数>maximumPoolSize，执行reject

其实也就是对应源码中的这一段

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {//workerCountOf为目前活动的线程数
        if (addWorker(command, true))//添加任务到核心线程
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {//尝试添加到队列中，如果能加就加
        int recheck = ctl.get();
        //double-check 一下当前线程池的情况，正常情况下下面两个判断都不会命中
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}