ThreadPoolExecutor管理异步线程笔记

为什么使用线程池？

• 线程的创建和销毁都需要不小的系统开销，不加以控制管理容易发生OOM错误。
• 避免线程并发抢占系统资源导致系统阻塞。
• 具备一定的线程管理能力（数量、存活时间，任务管理）

new ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory) ;

参数说明

• corePoolSize: 线程池中的线程数量
• maximumPoolSize: 线程池中的最大线程数量
• keepAliveTime: 当线程池线程数量超过corePoolsize时，多余的空闲线程的存活时间，即超过corePoolSize的空闲线程，在keepAliveTime时间内会被销毁
• TimeUnit unit: keepAliveTime的单位
• BlockingQueue<Runnable> workQueue: 任务队列，管理被提交但尚未被执行的任务
• ThreadFactory threadFactory: 线程工厂，用于创建线程
• RejectedExecutionHandler handler: 拒绝策略。当任务太多来不及处理时，如何拒绝任务

BlockingQueue的几种形式：

• SynchronousQueue：这个队列接收到任务的时候，会直接提交给线程处理，而不保留它，如果所有线程都在工作怎么办？那就新建一个线程来处理这个任务！所以为了保证不出现<线程数达到了maximumPoolSize而不能新建线程>的错误，使用这个类型队列的时候，maximumPoolSize一般指定成Integer.MAX_VALUE，即无限大（但是这样就容易成OOM，因为Spring的工具类Executors创建线程池的底层也是使用MAX_VALUE所以并不是很推荐）。

• LinkedBlockingQueue：这个队列接收到任务的时候，如果当前线程数小于核心线程数，则新建线程(核心线程)处理任务；如果当前线程数等于核心线程数，则进入队列等待。由于这个队列没有最大值限制，即所有超过核心线程数的任务都将被添加到队列中，这也就导致了maximumPoolSize的设定失效，因为总线程数永远不会超过corePoolSize。（该方式需要协调好任务处理时间，否则容易造成任务数量过多，最差的情况会耗尽系统资源）

• ArrayBlockingQueue：可以限定队列的长度，接收到任务的时候，如果没有达到corePoolSize的值，则新建线程(核心线程)执行任务，如果达到了，则入队等候，如果队列已满，则新建线程(非核心线程)执行任务，又如果总线程数到了maximumPoolSize，并且队列也满了，则发生错误。

• DelayQueue：队列内元素必须实现Delayed接口，这就意味着你传进去的任务必须先实现Delayed接口。这个队列接收到任务时，首先先入队，只有达到了指定的延时时间，才会执行任务。

示例Demo

    @Testpublic void testExecutor() {ThreadPoolExecutor pool = new ThreadPoolExecutor(50, 50, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(),new NamedThreadFactory("CustomerThreadName01") // 自定义线程池名称);// 默认Runnablefor (int i = 0; i < 10; i++) {pool.execute(new Runnable() {@Overridepublic void run() {// doSomeThing}});}// 自定义Runnablefor (int i = 0; i < 10; i++) {pool.execute(new MyRunnable("线程" + i, array[i]));}}

这里使用自定义线程池和自定义MyRunnable的目的是为了当线程出现异常的时候，通过日志可以更具自定线程池的名称和自定义Runnable的名称知道是哪个线程的池发生的异常，所以一般推荐不同的业务使用不同线程池的时候，便于线程异常的时候追查。

package com.lg.demo.thread.factory;import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;/*** @Description: 自定义线程池名称* @Author: GE LIANG* @Date: 2023/1/30 15:11*/
public class NamedThreadFactory implements ThreadFactory {private static final AtomicInteger poolNumber = new AtomicInteger(1);private final ThreadGroup group;private final AtomicInteger threadNumber = new AtomicInteger(1);private final String namePrefix;public NamedThreadFactory(String name){SecurityManager s = System.getSecurityManager();group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();if (null == name || name.isEmpty()){name = "pool";}namePrefix = name + "-" + poolNumber.getAndIncrement() + "-thread-";}@Overridepublic Thread newThread(Runnable r){Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);if (t.isDaemon()){t.setDaemon(false);}if (t.getPriority() != Thread.NORM_PRIORITY){t.setPriority(Thread.NORM_PRIORITY);}return t;}
}

package com.lg.demo.thread.funnable;import lombok.Data;/*** @Description: 可自定义Runnable* @Author: GE LIANG* @Date: 2023/1/30 15:27*/
public class MyRunnable implements Runnable {public String name;public Integer index;public MyRunnable(String name, Integer index) {this.name = name;this.index = index;}@Overridepublic void run() {System.out.println(name + ">>>" + index);}public String getName() {return name;}public void setName(String name) {this.name = name;}public Integer getIndex() {return index;}public void setIndex(Integer index) {this.index = index;}
}

引用

《阿里巴巴Java开发规范》
《Java常用四大线程池用法以及ThreadPoolExecutor详解》