Java进阶——— 线程池的原理分析

谁借莪１个温暖的怀抱￠ 2023-10-10 18:37 65阅读 0赞

### 前言 ###

在了解线程池之前，其实首先出现的疑问是：为什么要使用线程池，其次是了解什么是线程池，最后是如何使用线程池，带着疑问去学习。

### 为什么要使用 ###

前面多线程文章中，需要使用线程就开启一个新线程，简单方便，但是这样在大量线程被开启时：如果并发的线程数量很多，并且每个线程都是执行一个时间很短的任务就结束了，这样频繁创建线程就会大大降低系统的效率，因为频繁创建线程和销毁线程需要时间。

那么我们可不可以开启适量的线程，执行完任务不被销毁，继续执行新的任务呢？

Java中，为我们提供了线程池来实现这个目标，所以先来了解，什么是线程池，线程池的实现原理是什么？

### 线程池 ###

线程池为线程生命周期开销问题和资源不足问题提供了解决方案。通过对多个任务重用线程，线程创建的开销被分摊到了多个任务上。其好处是，因为在请求到达时线程已经存在，所以无意中也消除了线程创建所带来的延迟。这样，就可以立即为请求服务，使应用程序响应更快。而且，通过适当地调整线程池中的线程数目，也就是当请求的数目超过某个阈值时，就强制其它任何新到的请求一直等待，直到获得一个线程来处理为止，从而可以防止资源不足

#### 一. Java中的ThreadPoolExecutor类 ####

了解线程池，从线程池最核心的类ThreadPoolExecutor类开始了解，尽管一万个不愿意，还是需要看源码

在ThreadPoolExecutor类中提供了四个构造方法：

java

复制代码

`public class ThreadPoolExecutor extends AbstractExecutorService { ..... public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); } public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler); } public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler); } public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { ...... } ... }`

可以看出，前三个构造函数最终调用的是第四个构造函数进行初始化操作。

逐个解释构造器中各个参数的含义：

*  `corePoolSize`：核心池的大小，这个参数跟后面讲述的线程池的实现原理有非常大的关系。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务。除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这2个方法的名字就可以看出，是预创建线程的意思，即在没有任务到来之前就创建corePoolSize个线程或者一个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中；
 *  `maximumPoolSize`：线程池最大线程数，这个参数也是一个非常重要的参数，它表示在线程池中最多能创建多少个线程；
 *  `keepAliveTime`：表示线程没有任务执行时最多保持多久时间会终止。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0；
 *  `unit`：参数keepAliveTime的时间单位，有7种取值，在TimeUnit类中有7种静态属性：
    
     *  TimeUnit.DAYS; //天
     *  TimeUnit.HOURS; //小时
     *  TimeUnit.MINUTES; //分钟
     *  TimeUnit.SECONDS; //秒
     *  TimeUnit.MILLISECONDS; //毫秒
     *  TimeUnit.MICROSECONDS; //微妙
     *  TimeUnit.NANOSECONDS; //纳秒
 *  `workQueue`：一个阻塞队列，用来存储等待执行的任务，这个参数的选择也很重要，会对线程池的运行过程产生重大影响，一般来说，这里的阻塞队列有以下几种选择
    
     *  ArrayBlockingQueue：基于数组的先进先出队列，此队列创建时必须指定大小；
     *  LinkedBlockingQueue：基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX\_VALUE；
     *  synchronousQueue：这个队列比较特殊，它不会保存提交的任务，而是直接新建一个线程来执行新来的任务。
     *  DelayQueue：DelayQueue中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。DelayQueue是一个没有大小限制的队列，因此往队列中插入数据的操作（生产者）永远不会被阻塞，而只有获取数据的操作（消费者）才会被阻塞。
     *  PriorityBlockingQueue：基于优先级的阻塞队列（优先级的判断通过构造函数传入的Compator对象来决定），但需要注意的是PriorityBlockingQueue并不会阻塞数据生产者，而只会在没有可消费的数据时，阻塞数据的消费者。因此使用的时候要特别注意，生产者生产数据的速度绝对不能快于消费者消费数据的速度，否则时间一长，会最终耗尽所有的可用堆内存空间。
 *  `threadFactory`：线程工厂，主要用来创建线程；一般不用自己实现，使用Executors.defaultThreadFactory()
 *  `handler`：表示当拒绝处理任务时的策略，有以下四种取值
    
     *  ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
     *  ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
     *  ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
     *  ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

根据上面ThreadPoolExecutor类的代码可以知道，ThreadPoolExecutor继承了AbstractExecutorService，简单看一下AbstractExecutorService的实现：

swift

复制代码

`public abstract class AbstractExecutorService implements ExecutorService { protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {} protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {} public Future<?> submit(Runnable task) {} public <T> Future<T> submit(Runnable task, T result) {} public <T> Future<T> submit(Callable<T> task) {} private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,boolean timed, long nanos) throws InterruptedException, ExecutionException, TimeoutException {} public <T> T invokeAny(Collection<? extends Callable<T>> tasks)throws InterruptedException, ExecutionException {} public <T> T invokeAny(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException {} public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)throws InterruptedException {} public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)throws InterruptedException {} private static <T> void cancelAll(ArrayList<Future<T>> futures) {} private static <T> void cancelAll(ArrayList<Future<T>> futures, int j) {} }`

AbstractExecutorService是一个抽象类，它实现了ExecutorService接口

java

复制代码

`public interface ExecutorService extends Executor { void shutdown(); List<Runnable> shutdownNow(); boolean isShutdown(); boolean isTerminated(); boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException; <T> Future<T> submit(Runnable task, T result); Future<?> submit(Runnable task); <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException; <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit) throws InterruptedException; <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException; <T> T invokeAny(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }`

而ExecutorService又是继承了Executor接口，我们看一下Executor接口的实现：

csharp

复制代码

`public interface Executor { void execute(Runnable command); }`

通过上面的继承与实现，能很明了的掌握ThreadPoolExecutor与AbstractExecutorService、ExecutorService、Executor之间的关系：

1.  Executor是一个顶层接口，在它里面只声明了一个方法execute(Runnable)，返回值为void，参数为Runnable类型，从字面意思可以理解，就是用来执行传进去的任务的；
2.  ExecutorService接口继承了Executor接口，并声明了一些方法：submit、invokeAll、invokeAny以及shutDown等
3.  抽象类AbstractExecutorService实现了ExecutorService接口，基本实现了ExecutorService中声明的所有方法
4.  最后ThreadPoolExecutor继承了类AbstractExecutorService

在ThreadPoolExecutor类中有几个非常重要的方法：

scss

复制代码

`execute() submit() shutdown() shutdownNow()`

`execute()`方法实际上是Executor中声明的方法，在ThreadPoolExecutor进行了具体的实现，这个方法是ThreadPoolExecutor的核心方法，通过这个方法可以向线程池提交一个任务，交由线程池去执行。

`submit()`方法是在ExecutorService中声明的方法，在AbstractExecutorService就已经有了具体的实现，在ThreadPoolExecutor中并没有对其进行重写，这个方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，去看submit()方法的实现，会发现它实际上还是调用的execute()方法，只不过它利用了Future来获取任务执行结果（Future相关内容将在下一篇讲述）。

`shutdown()`和`shutdownNow()`是用来关闭线程池的。

#### 二.深入剖析线程池实现原理 ####

第一节，简单了解ThreadPoolExecutor，接下来深入分析线程池的具体实现原理，将从下面几个方面讲解：  
[1.线程池状态][1.]  
[2.任务的执行][2.]  
[3.线程池中的线程初始化][3.]  
[4.任务缓存队列及排队策略][4.]  
[5.任务拒绝策略][5.]  
[6.线程池的关闭][6.]  
[7.线程池容量的动态调整][7.]

1.线程池状态

在ThreadPoolExecutor中定义了几个static final变量表示线程池的各个状态：

arduino

复制代码

`// runState is stored in the high-order bits private static final int RUNNING = -1 << COUNT_BITS; private static final int SHUTDOWN = 0 << COUNT_BITS; private static final int STOP = 1 << COUNT_BITS; private static final int TIDYING = 2 << COUNT_BITS; private static final int TERMINATED = 3 << COUNT_BITS;`

*  RUNNING  
    (1) 状态说明：线程池处在RUNNING状态时，能够接收新任务，以及对已添加的任务进行处理。  
    (2) 状态切换：线程池的初始化状态是RUNNING。换句话说，线程池被一旦被创建，就处于RUNNING状态，并且线程池中的任务数为0！
 *  SHUTDOWN  
    (1) 状态说明：线程池处在SHUTDOWN状态时，不接收新任务，但能处理已添加的任务。  
    (2) 状态切换：调用线程池的shutdown()接口时，线程池由RUNNING -> SHUTDOWN。
 *  STOP  
    (1) 状态说明：线程池处在STOP状态时，不接收新任务，不处理已添加的任务，并且会中断正在处理的任务。  
    (2) 状态切换：调用线程池的shutdownNow()接口时，线程池由(RUNNING or SHUTDOWN ) -> STOP。
 *  TIDYING  
    (1) 状态说明：当所有的任务已终止，ctl记录的”任务数量”为0，线程池会变为TIDYING状态。当线程池变为TIDYING状态时，会执行钩子函数terminated()。terminated()在ThreadPoolExecutor类中是空的，若用户想在线程池变为TIDYING时，进行相应的处理；可以通过重载terminated()函数来实现。  
    (2) 状态切换：当线程池在SHUTDOWN状态下，阻塞队列为空并且线程池中执行的任务也为空时，就会由 SHUTDOWN -> TIDYING。  
    当线程池在STOP状态下，线程池中执行的任务为空时，就会由STOP -> TIDYING。
 *  TERMINATED  
    (1) 状态说明：线程池彻底终止，就变成TERMINATED状态。  
    (2) 状态切换：线程池处在TIDYING状态时，执行完terminated()之后，就会由 TIDYING -> TERMINATED。

2.任务的执行

在了解将任务提交给线程池到任务执行完毕整个过程之前，我们先来看一下ThreadPoolExecutor类中其他的一些比较重要成员变量：

java

复制代码

`private final BlockingQueue<Runnable> workQueue; //任务缓存队列，用来存放等待执行的任务 private final ReentrantLock mainLock = new ReentrantLock(); //线程池的主要状态锁，对线程池状态（比如线程池大小 //、runState等）的改变都要使用这个锁 private final HashSet<Worker> workers = new HashSet<Worker>(); //用来存放工作集 private volatile long keepAliveTime; //线程存货时间 private volatile boolean allowCoreThreadTimeOut; //是否允许为核心线程设置存活时间 private volatile int corePoolSize; //核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列） private volatile int maximumPoolSize; //线程池最大能容忍的线程数 private volatile int poolSize; //线程池中当前的线程数 private volatile RejectedExecutionHandler handler; //拒绝策略的处理句柄。所谓拒绝策略，是指将任务添加到线程池中时，线程池拒绝该任务所采用的相应策略。 private volatile ThreadFactory threadFactory; //线程工厂，用来创建线程 private int largestPoolSize; //用来记录线程池中曾经出现过的最大线程数 private long completedTaskCount; //用来记录已经执行完毕的任务个数`

每个变量的作用添加了解释，这里要对corePoolSize、maximumPoolSize、largestPoolSize三个变量重点解释

corePoolSize一般翻译为核心池的大小，也可以理解就是线程池的大小  
，举个例子来说明：  
假如有一个工厂，工厂里面有10个工人，每个工人同时只能做一件任务。  
因此只要当10个工人中有工人是空闲的，来了任务就分配给空闲的工人做；  
当10个工人都有任务在做时，如果还来了任务，就把任务进行排队等待；  
如果说新任务数目增长的速度远远大于工人做任务的速度，那么此时工厂主管可能会想补救措施，比如重新招4个临时工人进来；  
然后就将任务也分配给这4个临时工人做；  
如果说着14个工人做任务的速度还是不够，此时工厂主管可能就要考虑不再接收新的任务或者抛弃前面的一些任务了。  
当这14个工人当中有人空闲时，而新任务增长的速度又比较缓慢，工厂主管可能就考虑辞掉4个临时工了，只保持原来的10个工人，毕竟请额外的工人是要花钱的。

这里corePoolSize就是10，而maximumPoolSize就是14（10+4），也就是说corePoolSize就是线程池大小，maximumPoolSize在我看来是线程池的一种补救措施，即任务量突然过大时的一种补救措施。

largestPoolSize只是一个用来起记录作用的变量，用来记录线程池中曾经有过的最大线程数目，跟线程池的容量没有任何关系

ThreadPoolExecutor类中最核心的方法是任务提交execute()，虽然通过submit也可以提交任务，但是实际上submit方法里面最终调用的还是execute()方法，所以我们只需要研究execute()方法的实现原理即可

execute()

scss

复制代码

`public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false)) reject(command); }`

下面一行行对代码进行解释:  
首先判断任务command是否为null，若是null，则抛出空指针异常；  
接着获取ctl对应的int值。该int值保存了"线程池状态"和"线程池中任务的数量"信息  
`private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0))`

ini

复制代码

`int c = ctl.get();`

接下来判断 线程池中线程数量是否 < “核心线程池数量”,也就是线程池中少于corePoolSize个任务

less

复制代码

`if (workerCountOf(c) < corePoolSize)`

状态1：小于  
尝试通过`addWorker(command, true)`将创建新线程将任务添加到该线程，然后添加到工作集`workers`。如果添加成功则启动线程开始执行任务。

*  1.如果任务添加成功，则整个方法结束。
 *  2.如果失败，重新获取ctl的int值，然后继续进入状态2

状态2：大于

less

复制代码

`if (isRunning(c) && workQueue.offer(command))`

如果线程池处于`RUNNING`状态，尝试将任务添加到任务缓存队列  
如果添加成功,进入状态3，否则状态4

状态3：再次确认“线程池状态”，若线程池异常终止了，则删除任务；然后通过reject()执行相应的拒绝策略的内容。

scss

复制代码

`int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command);`

否则，如果"线程池中任务数量"为0，则通过addWorker(null, false)尝试新建一个线程，新建线程对应的任务为null。

csharp

复制代码

`else if (workerCountOf(recheck) == 0) addWorker(null, false);`

状态4：  
最后再尝试通过addWorker(command, false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。如果还是失败，则通过reject(command)执行相应的拒绝策略的内容。

bash

复制代码

`else if (!addWorker(command, false)) reject(command);`

**梳理**  
至此，提交新任务就执行完毕，再来梳理一遍

*  1.如果 当前线程池线程数量 < corePoolSize，则尝试开启新线程添加任务并执行线程。如果成功，则整个方法结束
 *  2.如果 当前线程池线程数量 > corePoolSize，并且允许将当前任务添加任务缓存队列。
    
     *  2.1这时候来重新检查线程池运行状态,如果不是`RUNNING`状态，则尝试将任务从缓存队列删除，如果成功然后通过`reject`执行相应拒绝策略
     *  2.2否则检查线程池中任务数量是不是0，如果为0，则通过addWorker(null, false)尝试新建一个线程，新建线程对应的任务为null
 *  3.上面两种都不符合，再尝试通过addWorker(command, false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。如果还是失败，则通过reject(command)执行相应的拒绝策略的内容

继续查看addWorker()源码

addWorker()

java

复制代码

`private boolean addWorker(Runnable firstTask, boolean core) { // 更新"线程池状态和计数"标记，即更新ctl。 retry: for (;;) { //获取ctl对应的int值。该int值保存了"线程池状态"和"线程池中任务的数量"信息 int c = ctl.get(); //获取线程池运行状态 int rs = runStateOf(c); // 检查运行状态是不是可运行状态并且者缓存队列是否为空，如果不合法，则返回false if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { //获取线程池中任务数量 int wc = workerCountOf(c); //如果数量超过默认容量或者最大线程池数量，返回false if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; // 通过CAS函数将c的值+1。操作失败的话，则退出循环。 if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl // 检查"线程池状态"，如果与之前的状态不同，则从retry重新开始。 if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; // 开启新线程添加任务，然后加入线程池，并启动任务所在的线程。 try { // 新建Worker，并且指定firstTask为Worker的第一个任务。 w = new Worker(firstTask); // 获取Worker对应的线程。 final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; //获取锁 mainLock.lock(); try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. //获取线程池运行状态 int rs = runStateOf(ctl.get()); //再次确认运行状态是不是"RUNNING"状态 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { //检查线程是不是可以启动 if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); //将任务添加到缓存队列 workers.add(w); //获取缓存队列数量 int s = workers.size(); //如果缓存队列大于线程池出现过的最大线程数量 if (s > largestPoolSize) //修改largestPoolSize的值 largestPoolSize = s; //添加任务成功 workerAdded = true; } } finally { //不要忘记释放锁 mainLock.unlock(); } //如果任务添加成功 if (workerAdded) { //运行线程 t.start(); //线程启动成功 workerStarted = true; } } } finally { //如果线程未成功启动 if (! workerStarted) //任务启动失败，将任务从缓存队列删除 addWorkerFailed(w); } //返回任务启动的状态 return workerStarted; }`

**梳理**  
addWorker(Runnable firstTask,boolean core)的作用是将任务（firstTask）添加到线程池中，并启动该任务。

*  core为true的话，则以corePoolSize为界限，若”线程池中已有任务数量>=corePoolSize”，则返回false；
 *  core为false的话，则以maximumPoolSize为界限，若”线程池中已有任务数量>=maximumPoolSize”，则返回false。
 *  addWorker()会先通过for循环不断尝试更新ctl状态，ctl记录了”线程池中任务数量和线程池状态”。
 *  更新成功之后，再通过try模块来将任务添加到线程池中，并启动任务所在的线程。
 *  可以得到：线程池在添加任务时，会创建任务对应的Worker对象；而一个Worker对象包含一个Thread对象。
    
     *  通过将worker对象添加到“线程的workers集合”中，从而实现将任务添加到线程池中。
     *  通过启动Worker对应的Thread线程，则执行该任务。

查看关闭线程池shutdown()

shutdown()

scss

复制代码

`public void shutdown() { //获取锁 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //检查是否有权限终止线程池中线程 checkShutdownAccess(); //将线程池运行状态设为关闭 advanceRunState(SHUTDOWN); //中断线程池中空闲的线程 interruptIdleWorkers(); // 钩子函数，在ThreadPoolExecutor中没有任何动作。 onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { //释放锁 mainLock.unlock(); } //尝试终止线程 tryTerminate(); }`

3.线程池中的线程初始化

默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务之后才会创建线程。  
如果需要线程池创建之后立即创建线程，可以通过以下两个方法：

*  prestartCoreThread()：初始化一个核心线程；
 *  prestartAllCoreThreads()：初始化所有核心线程

csharp

复制代码

`public boolean prestartCoreThread() { return addIfUnderCorePoolSize(null); //注意传进去的参数是null } public int prestartAllCoreThreads() { int n = 0; while (addIfUnderCorePoolSize(null))//注意传进去的参数是null ++n; return n; }`

4.任务缓存队列及排队策略

前面我们提到了任务缓存队列，即workQueue，它用来存放等待执行的任务

*  ArrayBlockingQueue：基于数组的先进先出队列，此队列创建时必须指定大小；
 *  LinkedBlockingQueue：基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX\_VALUE；
 *  synchronousQueue：这个队列比较特殊，它不会保存提交的任务，而是直接新建一个线程来执行新来的任务。
 *  DelayQueue：DelayQueue中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。DelayQueue是一个没有大小限制的队列，因此往队列中插入数据的操作（生产者）永远不会被阻塞，而只有获取数据的操作（消费者）才会被阻塞。
 *  PriorityBlockingQueue：基于优先级的阻塞队列（优先级的判断通过构造函数传入的Compator对象来决定），但需要注意的是PriorityBlockingQueue并不会阻塞数据生产者，而只会在没有可消费的数据时，阻塞数据的消费者。因此使用的时候要特别注意，生产者生产数据的速度绝对不能快于消费者消费数据的速度，否则时间一长，会最终耗尽所有的可用堆内存空间。

5.任务拒绝策略

当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略，通常有以下四种策略：

*  ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
 *  ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
 *  ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
 *  ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

6.线程池的关闭

ThreadPoolExecutor提供了两个方法，用于线程池的关闭，分别是shutdown()和shutdownNow()，其中：

*  shutdown()：不会立即终止线程池，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务
 *  shutdownNow()：立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务

7.线程池容量的动态调整

ThreadPoolExecutor提供了动态调整线程池容量大小的方法：setCorePoolSize()和setMaximumPoolSize()，

*  setCorePoolSize：设置核心池大小
 *  setMaximumPoolSize：设置线程池最大能创建的线程数目大小

当上述参数从小变大时，ThreadPoolExecutor进行线程赋值，还可能立即创建新的线程来执行任务。

#### 怎么使用线程池 ####

下面来看看线程池的具体使用

kotlin

复制代码

`class ThreadExecutor { fun test(){ val executor = ThreadPoolExecutor(5,10,300,TimeUnit.MILLISECONDS, ArrayBlockingQueue<Runnable>(5),ThreadPoolExecutor.DiscardOldestPolicy()) for (i in 0 until 20){ try { val task = MyRunnable(i) executor.execute(task) System.out.println("线程池中线程数目："+executor.poolSize +"，队列中等待执行的任务数目："+ executor.queue.size+"，已执行完别的任务数目："+executor.completedTaskCount ) } catch (e: Exception) { e.printStackTrace() } } executor.shutdown() } class MyRunnable constructor(private val number:Int) : Runnable { override fun run() { System.out.println("正在执行的task $number") try { Thread.sleep(2000) } catch (e: Exception) { e.printStackTrace() }finally { } System.out.println("task $number 执行完毕"); } } }`

查看输出结果

bash

复制代码

`01-21 19:50:14.756 24533-24533/com.t9.news I/System.out: 线程池中线程数目：1，队列中等待执行的任务数目：0，已执行完别的任务数目：0 01-21 19:50:14.757 24533-24605/com.t9.news I/System.out: 正在执行的task 0 01-21 19:50:14.757 24533-24533/com.t9.news I/System.out: 线程池中线程数目：2，队列中等待执行的任务数目：0，已执行完别的任务数目：0 01-21 19:50:14.758 24533-24533/com.t9.news I/System.out: 线程池中线程数目：3，队列中等待执行的任务数目：0，已执行完别的任务数目：0 01-21 19:50:14.758 24533-24607/com.t9.news I/System.out: 正在执行的task 2 01-21 19:50:14.759 24533-24533/com.t9.news I/System.out: 线程池中线程数目：4，队列中等待执行的任务数目：0，已执行完别的任务数目：0 01-21 19:50:14.759 24533-24533/com.t9.news I/System.out: 线程池中线程数目：5，队列中等待执行的任务数目：0，已执行完别的任务数目：0 01-21 19:50:14.759 24533-24608/com.t9.news I/System.out: 正在执行的task 3 01-21 19:50:14.759 24533-24533/com.t9.news I/System.out: 线程池中线程数目：5，队列中等待执行的任务数目：1，已执行完别的任务数目：0 01-21 19:50:14.759 24533-24606/com.t9.news I/System.out: 正在执行的task 1 01-21 19:50:14.759 24533-24533/com.t9.news I/System.out: 线程池中线程数目：5，队列中等待执行的任务数目：2，已执行完别的任务数目：0 01-21 19:50:14.760 24533-24609/com.t9.news I/System.out: 正在执行的task 4 01-21 19:50:14.760 24533-24533/com.t9.news I/System.out: 线程池中线程数目：5，队列中等待执行的任务数目：3，已执行完别的任务数目：0 01-21 19:50:14.760 24533-24533/com.t9.news I/System.out: 线程池中线程数目：5，队列中等待执行的任务数目：4，已执行完别的任务数目：0 01-21 19:50:14.760 24533-24533/com.t9.news I/System.out: 线程池中线程数目：5，队列中等待执行的任务数目：5，已执行完别的任务数目：0 01-21 19:50:14.760 24533-24533/com.t9.news I/System.out: 线程池中线程数目：6，队列中等待执行的任务数目：5，已执行完别的任务数目：0 01-21 19:50:14.764 24533-24610/com.t9.news I/System.out: 正在执行的task 10 01-21 19:50:14.764 24533-24533/com.t9.news I/System.out: 线程池中线程数目：7，队列中等待执行的任务数目：5，已执行完别的任务数目：0 01-21 19:50:14.765 24533-24533/com.t9.news I/System.out: 线程池中线程数目：8，队列中等待执行的任务数目：5，已执行完别的任务数目：0 01-21 19:50:14.766 24533-24533/com.t9.news I/System.out: 线程池中线程数目：9，队列中等待执行的任务数目：5，已执行完别的任务数目：0 01-21 19:50:14.767 24533-24614/com.t9.news I/System.out: 正在执行的task 13 01-21 19:50:14.767 24533-24533/com.t9.news I/System.out: 线程池中线程数目：10，队列中等待执行的任务数目：5，已执行完别的任务数目：0 01-21 19:50:14.767 24533-24615/com.t9.news I/System.out: 正在执行的task 14 01-21 19:50:14.767 24533-24533/com.t9.news I/System.out: 线程池中线程数目：10，队列中等待执行的任务数目：5，已执行完别的任务数目：0 01-21 19:50:14.767 24533-24613/com.t9.news I/System.out: 正在执行的task 12 01-21 19:50:14.767 24533-24533/com.t9.news I/System.out: 线程池中线程数目：10，队列中等待执行的任务数目：5，已执行完别的任务数目：0 01-21 19:50:14.768 24533-24533/com.t9.news I/System.out: 线程池中线程数目：10，队列中等待执行的任务数目：5，已执行完别的任务数目：0 01-21 19:50:14.771 24533-24612/com.t9.news I/System.out: 正在执行的task 11 01-21 19:50:16.759 24533-24605/com.t9.news I/System.out: task 0 执行完毕 01-21 19:50:16.759 24533-24607/com.t9.news I/System.out: task 2 执行完毕 01-21 19:50:16.760 24533-24605/com.t9.news I/System.out: 正在执行的task 15 01-21 19:50:16.760 24533-24607/com.t9.news I/System.out: 正在执行的task 16 01-21 19:50:16.760 24533-24608/com.t9.news I/System.out: task 3 执行完毕 01-21 19:50:16.760 24533-24609/com.t9.news I/System.out: task 4 执行完毕 01-21 19:50:16.760 24533-24606/com.t9.news I/System.out: task 1 执行完毕 01-21 19:50:16.761 24533-24608/com.t9.news I/System.out: 正在执行的task 17 01-21 19:50:16.761 24533-24609/com.t9.news I/System.out: 正在执行的task 18 01-21 19:50:16.763 24533-24606/com.t9.news I/System.out: 正在执行的task 19 01-21 19:50:16.765 24533-24610/com.t9.news I/System.out: task 10 执行完毕 01-21 19:50:16.767 24533-24614/com.t9.news I/System.out: task 13 执行完毕 01-21 19:50:16.767 24533-24613/com.t9.news I/System.out: task 12 执行完毕 01-21 19:50:16.767 24533-24615/com.t9.news I/System.out: task 14 执行完毕 01-21 19:50:16.771 24533-24612/com.t9.news I/System.out: task 11 执行完毕 01-21 19:50:18.761 24533-24605/com.t9.news I/System.out: task 15 执行完毕 01-21 19:50:18.761 24533-24608/com.t9.news I/System.out: task 17 执行完毕 01-21 19:50:18.762 24533-24607/com.t9.news I/System.out: task 16 执行完毕 01-21 19:50:18.762 24533-24609/com.t9.news I/System.out: task 18 执行完毕 01-21 19:50:18.763 24533-24606/com.t9.news I/System.out: task 19 执行完毕`

从执行结果可以看出，当线程池中线程的数目大于5时，便将任务放入任务缓存队列里面，当任务缓存队列满了之后，变丢弃最前面的任务，即5-9任务被丢弃。

不过在java doc中，并不提倡我们直接使用ThreadPoolExecutor，而是使用Executors类中提供的几个静态方法来创建线程池：

scss

复制代码

`Executors.newCachedThreadPool(); //创建一个缓冲池，缓冲池容量大小为Integer.MAX_VALUE Executors.newSingleThreadExecutor(); //创建容量为1的缓冲池 Executors.newFixedThreadPool(int); //创建固定容量大小的缓冲池`

下面是这三个静态方法的具体实现:

csharp

复制代码

`//corePoolSize=maximumPoolSize，使用LinkedBlockingQueue public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); } //corePoolSize和maximumPoolSize都设置为1，也使用的LinkedBlockingQueue public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); } corePoolSize为0，将maximumPoolSize为Integer.MAX_VALUE，时间60s,使用的SynchronousQueue，也就是说来了任务就创建线程运行，当线程空闲超过60s，就销毁线程 public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }`

一般开发中，尽量使用Executors提供的静态方式生成线程池，如果不满足需求，再自己去实现。

### 总结 ###

以上就是线程池的主要原理，核心是提交execute()任务、添加任务addWorker()、关闭线程池shutdown()以及线程池的使用方式

[1.]: https://juejin.cn/post/7234341843579502649#1111
[2.]: https://juejin.cn/post/7234341843579502649#1112
[3.]: https://juejin.cn/post/7234341843579502649#1113
[4.]: https://juejin.cn/post/7234341843579502649#1114
[5.]: https://juejin.cn/post/7234341843579502649#1115
[6.]: https://juejin.cn/post/7234341843579502649#1116
[7.]: https://juejin.cn/post/7234341843579502649#1117