【java笔记】线程池ThreadPoolExecutor
作者：互联网
===============================================
2022/6/23_第2次修改 ccb_warlock
更新说明：
2022/6/23:
1.补充了“为什么不允许通过Executors创建线程”的内容；
===============================================
在java中使用线程池，是在一个项目接口的业务后新加一个接口转发，但是该转发不能因为转发业务的响应慢而阻塞该项目接口。
之前看《阿里巴巴Java开发手册》在“并发控制”中写明了“线程池不允许使用Executors创建，而是通过ThreadPoolExecutor的方式创建，这样的处理方式能让编写代码的工程师更加明确线程池的运行规则，规避资源耗尽的风险”，于是在项目中开始使用ThreadPoolExecutor创建线程。
应用了一段时间后，发现ThreadPoolExecutor在不同的应用场景需要做不同的配置，于是我针对这块做了试验和整理。
一、为什么不允许通过Executors创建线程
根据《阿里巴巴Java开发手册》提到的Executors，学习后发现其实java的工具类提供了2套线程池的方案，我理解是Executors对ThreadPoolExecutor进行一层封装，可以减少学习上手就用。
但是由于JDK 1.8之前的Executors创建出来的线程池队列长度都是Integer的最大值，大量创建任务时会导致队列中塞入过多的任务，从而造成内存溢出情况，这获取就是手册标记“强制”不允许使用Executors创建线程池的原因吧。
不过JDK 1.8开始有了一种新的线程池newWorkStealingPool是基于ForkJoinPool创建的，以后有应用后再来补充。
Executors有newFixedThreadPool、newWorkStealingPool、newSingleThreadExecutor、newCachedThreadPool、newScheduledThreadPool5种方式创建线程池。
1.1 newFixedThreadPool
基于ThreadPoolExecutor。
创建一个多并发的线程池，任务量达到最大并发数时，超出的线程进入队列等待执行。
弊端：队列长度为Integer.MAX_VALUE，当大量任务堆积在队列中，从而导致OOM。
1.2 newSingleThreadExecutor
基于ThreadPoolExecutor。
创建一个单线程的线程池，其中只有一个工作线程执行任务。
弊端：队列长度为Integer.MAX_VALUE，当大量任务堆积在队列中，从而导致OOM。
1.3 newCachedThreadPool
基于ThreadPoolExecutor。
创建一个无工作线程的线程池，来一个任务就执行一个任务。
弊端：最大线程数是Integer.MAX_VALUE，当大量任务时会创建大量线程，从而导致OOM。
1.4 newScheduledThreadPool
基于ThreadPoolExecutor。
创建一个可定时执行的线程池。
弊端：最大线程数是Integer.MAX_VALUE，当大量任务时会创建大量线程，从而导致OOM。
1.5 newWorkStealingPool
JDK1.8引入，基于ForkJoinPool。
创建一个抢占式操作的线程池，每个并行线程都有各自的队列，每个线程执行各自队列的任务。
二、ThreadPoolExecutor构造参数
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
ThreadPoolExecutor的构造函数由corePoolSize、maximumPoolSize、keepAliveTime、unit、workQueue、threadFactory、handler7个参数组成。
2.1 corePoolSize
核心线程数。
线程池中初始的线程数，即使执行任务数小于核心线程数，线程池也不会回收这些线程。
2.2 maximumPoolSize
最大线程数。
该线程池最多创建多少个线程同时执行任务。
当任务数超过核心线程数（核心线程数+队列最大长度）时，线程池开始创建新线程来执行任务。
2.3 keepAliveTime
空闲线程保留时间。
工作线程数超过核心线程数时，多余的空闲线程会在保留时间后回收线程进行销毁。
2.4 unit
空闲线程保留时间的单位。
2.5 workQueue
任务队列。包括SynchronousQueue、ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue。
1）ArrayBlockingQueue
有界队列。该队列是基于数组的有界队列，所以在初始化的时候需要指定队列的长度。
当任务添加到线程池后，线程池会分配线程执行。当任务数超过corePoolSize后，多出来的任务添加到队列中。当多出来的任务已经塞满了队列后，将根据线程池的拒绝策略进行处理（拒绝策略后面会写）。
2）SynchronousQueue
同步队列。该队列我理解是长度为1的队列，put（生产）与take（消费）都是阻塞操作，当该任务生产后没有被消费将发生阻塞，当队列中没有可消费的任务时阻塞，这种机制也叫作配对通信机制。
当任务添加到使用同步队列的线程池后，任务通过阻塞的方式进入队列，然后该任务将选择空闲线程执行，当然如果没有空闲线程则创建新线程执行。
3）LinkedBlockingQueue
无界队列。该队列是基于链表的无界队列。
所以使用这种队列的时候maximumPoolSize实际是无效的，因为当线程池中的工作线程达到corePoolSize后，所有新任务将会持续增加到队列中。
4）PriorityBlockingQueue
优先队列。该队列是基于数组的无界队列。
当任务添加到线程池后，线程池会分配线程执行。当任务数超过corePoolSize后，多出来的任务在添加到队列后会根据任务的优先级重新排列，优先级高的排在队列前面。换句话说由于优先级不同，会出现“后进先出”的情况。
由于优先队列也属于无界队列，所以使用这种队列的时候maximumPoolSize实际是无效的，道理和LinkedBlockingQueue是一样的。
5）LinkedBlockingDeque
双端无界队列。该队列是基于双向链表的无界队列，由于基于双向队列所以可以从队列的两端插入和移除元素，支持FIFO和FILO。
6）DelayQueue
延迟队列。该队列是基于数组的无界队列，由于该队列在PriorityQueue基础上实现的，先按延迟优先级排序。
当任务添加到线程池后，线程池会分配线程执行。当任务数超过corePoolSize后，多出来的任务在添加到队列后会根据任务的延迟优先级重新排列。延迟时间短的排在队列前面。换句话说由于优先级不同，会出现“后进先出”的情况。
7）LinkedTransferQueue
//todo: 待学习补充
2.6 threadFactory
线程工厂。
用于创建线程，一般使用Executors.defaultThreadFactory()创建。
2.7 handler
拒绝策略。包括AbortPolicy、CallerRunsPolicy、DiscardPolicy、DiscardOldestPolicy。
当任务数超过（最大线程数+队列最大长度）时，如何处理多出来的任务。
1）AbortPolicy
该策略拒绝多出来的任务添加到线程池中，直接抛异常（RejectedExecutionException）。
2）CallerRunsPolicy
该策略将多出来的任务放到调用添加线程池的线程中执行。
3）DiscardPolicy
该策略将丢弃多出来的任务。
4）DiscardOldestPolicy
该策略将丢弃线程池中最早的任务（处于队头的任务），然后将多出来的任务添加到线程池中
三、创建ThreadPoolExecutor线程池
线程池的各项参数要根据实际的业务进行配置，下面是一个例子。
public class Test{
private ThreadPoolExecutor executor;
@PostConstruct
private void initExecutor(){
if(null == executor){
executor = new ThreadPoolExecutor(
1,
10,
15,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(15),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy()
);
}
}
}
四、通过ThreadPoolExecutor创建线程
4.1 创建单个不等待执行完成的异步线程
由于示例的线程池使用了AbortPolicy策略，所以runTask方法不会被阻塞。
public class Test{
private ThreadPoolExecutor executor;
@PostConstruct
private void initExecutor(){
if(null == executor){
executor = new ThreadPoolExecutor(
1,
10,
15,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(15),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy()
);
}
}
public void runTask(){
CompletableFuture.runAsync(() ->{
// todo: 业务代码
}, executor);
}
}
4.2 创建单个等待执行完成的异步线程
在CompletableFuture.runAsync后加上join，只要没有触发AbortPolicy策略，runTask方法在执行后会被线程阻塞，直到线程结束。
public class Test{
private ThreadPoolExecutor executor;
@PostConstruct
private void initExecutor(){
if(null == executor){
executor = new ThreadPoolExecutor(
1,
10,
15,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(15),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy()
);
}
}
public void runTask(){
CompletableFuture.runAsync(() ->{
// todo: 业务代码
}, executor).join();
}
}
4.3 创建多个不等待执行完成的异步线程
在项目中，经常需要创建多个线程业务，且希望这些业务不会阻塞外部方法，通过CompletableFuture.allOf可以创建多个线程。
例如针对每个order，我都需要执行各自的业务，但是希望线程任务不阻塞runTask方法。
public class Test{
private ThreadPoolExecutor executor;
@PostConstruct
private void initExecutor(){
if(null == executor){
executor = new ThreadPoolExecutor(
1,
10,
15,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(15),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy()
);
}
}
public void runTask(){
List<Order> orders = new ArrayList<>();
CompletableFuture.allOf(orders.stream().map(t ->
CompletableFuture.runAsync(() ->{
// todo：业务代码
}, executor)
).toArray(CompletableFuture[]::new));
}
}
4.4 创建多个等待执行完成的异步线程
在项目中，经常需要先完成多个异步线程后，再进行后续的操作，只需要在CompletableFuture.allOf后执行join就可以实现。
例如针对每个order，我都需要执行各自的业务，但是希望线程任务都完成后再继续执行runTask方法。
public class Test{
private ThreadPoolExecutor executor;
@PostConstruct
private void initExecutor(){
if(null == executor){
executor = new ThreadPoolExecutor(
1,
10,
15,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(15),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.AbortPolicy()
);
}
}
public void runTask(){
List<Order> orders = new ArrayList<>();
CompletableFuture.allOf(orders.stream().map(t ->
CompletableFuture.runAsync(() ->{
// todo：业务代码
}, executor)
).toArray(CompletableFuture[]::new)).join();
}
}
标签：java,队列,任务,线程,executor,new,ThreadPoolExecutor 来源： https://www.cnblogs.com/straycats/p/16353813.html