1. 线程池概述
在Java中，线程池（ThreadPool）是一种管理线程的技术，通过预先创建并管理一组线程，来减少频繁创建和销毁线程所带来的开销，从而提高系统的响应速度和吞吐量。ThreadPoolExecutor是Java并发包java.util.concurrent中的一个核心类，它提供了丰富的线程池功能。
2. ThreadPoolExecutor的核心概念
- 线程池状态：ThreadPoolExecutor有五种状态，包括RUNNING、SHUTDOWN、STOP、TIDYING和TERMINATED。这些状态之间通过调用shutdown()和shutdownNow()等方法进行转换。
- 核心参数：
- corePoolSize：线程池中核心线程数的最大值。
- maximumPoolSize：线程池中能拥有的最大线程数。
- workQueue：用于缓存任务的阻塞队列。
- keepAliveTime和TimeUnit：表示非核心线程的空闲存活时间及其时间单位。
- handler：当线程池无法处理新任务时所使用的拒绝策略。
3. ThreadPoolExecutor的工作原理
当向线程池提交一个任务时，ThreadPoolExecutor会按照以下步骤处理：
- 判断当前线程数：如果线程池中的线程数小于corePoolSize，则创建新线程执行任务。
- 判断工作队列：如果线程数已达到corePoolSize，则将任务加入workQueue中等待。
- 判断最大线程数：如果workQueue已满，且线程数小于maximumPoolSize，则创建新线程执行任务。
- 拒绝任务：如果线程数已达到maximumPoolSize，且workQueue已满，则根据handler指定的策略拒绝新任务。
4. 源码分析
4.1 核心字段
ThreadPoolExecutor 类中定义了一些关键的字段，这些字段对于理解其工作原理至关重要：
- ctl：一个 AtomicInteger 类型的字段，用于保存线程池的状态和线程数量。高三位表示线程池的状态，低29位表示线程数量。
- workQueue：一个阻塞队列，用于存放待执行的任务。
- corePoolSize：核心线程数。
- maximumPoolSize：最大线程数。
- keepAliveTime：非核心线程的空闲存活时间。
- threadFactory：用于创建新线程的线程工厂。
- handler：拒绝策略处理器，当线程池无法处理新任务时调用。
4.2 构造方法
构造方法用于初始化线程池的各个参数：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
// ... 省略了部分参数校验和初始化代码 ...
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
// ... 省略其他字段初始化 ...
}
4.3 任务提交与执行
execute 方法是提交任务到线程池的主要入口：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
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(