使用 ThreadPoolExecutor 管理线程池
在多线程编程中，线程池是一个关键的工具，可以有效地管理线程的生命周期，提高程序的性能和资源利用率。Java提供了一个强大的线程池实现，称为 ThreadPoolExecutor。
1. 引入ThreadPoolExecutor
ExecutorService executor = new ThreadPoolExecutor(
corePoolSize, // 核心线程数
maximumPoolSize, // 最大线程数
keepAliveTime, // 线程空闲时间
timeUnit, // 时间单位
workQueue, // 工作队列
threadFactory, // 线程工厂
rejectionPolicy // 拒绝策略
);2. 参数解释
a. corePoolSize (核心线程数)
- 核心线程数定义了线程池中保持活动状态的最少线程数量。
- 当提交一个任务时，线程池会创建一个线程来处理任务，即使此时线程池中已经存在空闲的线程。
- 如果线程池中的线程数小于 corePoolSize，即使其他线程都是空闲的，ThreadPoolExecutor 也会优先创建新的线程来处理任务。
b. maximumPoolSize (最大线程数)
- 最大线程数定义了线程池中允许的最大线程数量。
- 如果工作队列已满且活动线程数小于 maximumPoolSize，线程池会创建新线程来处理任务。
c. keepAliveTime (线程空闲时间)
- 当线程数超过 corePoolSize，空闲线程的最大存活时间。
- 超过这个时间，空闲线程会被终止，直到线程数量回到 corePoolSize。
d. timeUnit (时间单位)
- 用于指定 keepAliveTime 的单位，可以是秒、毫秒等。
e. workQueue (工作队列)
- 任务的排队队列，用于暂时存放等待执行的任务。
- 可以是
LinkedBlockingQueue（无界队列）或ArrayBlockingQueue（有界队列）等。 - 以下是一些常用的工作队列：
- LinkedBlockingQueue（链表阻塞队列）：
-
LinkedBlockingQueue是一个基于链表实现的无界队列，可以无限制地增加元素。 - 它的容量理论上可以是无限大的，适用于任务量比较大的场景。
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>();- ArrayBlockingQueue（数组阻塞队列）：
-
ArrayBlockingQueue是一个基于数组实现的有界队列，必须指定队列的容量。 - 当队列已满时，会拒绝新的任务。
BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(capacity);- SynchronousQueue（同步队列）：
-
SynchronousQueue是一个没有存储元素的队列，每个插入操作必须等待另一个线程的对应移除操作。 - 适用于直接将任务交给线程执行，而不需要缓冲任务的场景。
BlockingQueue<Runnable> workQueue = new SynchronousQueue<>();- PriorityBlockingQueue（优先级队列）：
-
PriorityBlockingQueue是一个无界的优先级队列，可以根据元素的优先级顺序进行处理。 - 需要任务实现
Comparable接口或者传入自定义的Comparator。
BlockingQueue<Runnable> workQueue = new PriorityBlockingQueue<>();- DelayQueue（延迟队列）：
-
DelayQueue是一个无界的队列，用于存放实现了Delayed接口的元素。 - 元素只有在其指定的延迟时间到了才能从队列中取出。
BlockingQueue<Runnable> workQueue = new DelayQueue<>();- LinkedTransferQueue（链表传输队列）：
-
LinkedTransferQueue是一个无界队列，可以在生产者和消费者之间传输元素。 - 可以通过
tryTransfer方法尝试直接将元素传输给消费者，如果不成功则添加到队列中。
BlockingQueue<Runnable> workQueue = new LinkedTransferQueue<>();- LinkedBlockingDeque（链表双向阻塞队列）：
-
LinkedBlockingDeque是一个基于链表实现的无界双向队列，可以在队头和队尾插入、移除元素。
BlockingQueue<Runnable> workQueue = new LinkedBlockingDeque<>();- 自定义工作队列：
- 你也可以实现自己的工作队列，只需要实现
BlockingQueue<Runnable>接口。
总的来说，选择工作队列的类型取决于你的具体需求和场景。不同的队列类型在不同的场景下有着不同的优劣势，需要根据实际情况来选择。
f. threadFactory (线程工厂)
- 用于创建新线程的工厂，可以自定义线程的名字、优先级等。
g. rejectionPolicy (拒绝策略)
- 当工作队列和线程池都满了，如何处理新提交的任务。有几种策略可选：
-
ThreadPoolExecutor.AbortPolicy：直接抛出异常。 -
ThreadPoolExecutor.CallerRunsPolicy：由调用者所在的线程执行任务。 -
ThreadPoolExecutor.DiscardPolicy：直接丢弃任务。 -
ThreadPoolExecutor.DiscardOldestPolicy：丢弃最老的任务。
3. 使用示例
import java.util.concurrent.*;
public class ThreadPoolExample {
public static void main(String[] args) {
int corePoolSize = 5;
int maximumPoolSize = 10;
long keepAliveTime = 60L;
TimeUnit timeUnit = TimeUnit.SECONDS;
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(100);
ThreadFactory threadFactory = Executors.defaultThreadFactory();
RejectedExecutionHandler rejectionPolicy = new ThreadPoolExecutor.CallerRunsPolicy();
ExecutorService executor = new ThreadPoolExecutor(
corePoolSize,
maximumPoolSize,
keepAliveTime,
timeUnit,
workQueue,
threadFactory,
rejectionPolicy
);
// 提交任务给线程池
for (int i = 0; i < 20; i++) {
executor.execute(new Task(i));
}
// 关闭线程池
executor.shutdown();
}
static class Task implements Runnable {
private final int taskId;
public Task(int taskId) {
this.taskId = taskId;
}
@Override
public void run() {
System.out.println("Task " + taskId + " is running.");
}
}
}在这个示例中，我们创建了一个 ThreadPoolExecutor，配置了各种参数，然后提交了20个任务给线程池执行。
protected ExecutorService executorService= new ThreadPoolExecutor(
10,
15,
30,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(1024),
r -> {
Thread t = new Thread(r);
t.setName("SAVE_D_STORAGE");
t.setPriority(Thread.NORM_PRIORITY + 2); // 增加优先级
return t;
},
new ThreadPoolExecutor.CallerRunsPolicy());4. 应用场景
ThreadPoolExecutor 适用于需要在后台执行异步任务的场景，比如：
- Web 服务器处理请求时，可以使用线程池来处理每个请求，提高并发处理能力。
- 后台任务的批量处理，比如数据同步、清理等。
- 提升程序性能，避免频繁创建和销毁线程的开销。
结语
ThreadPoolExecutor 是 Java 多线程编程中非常重要的工具，能够高效地管理线程的生命周期，提高程序性能和资源利用率。合理配置线程池参数，选择适当的拒绝策略，是保证系统稳定性和性能的关键。