定义
线程池
Java中的线程池为
ThreadPoolExecutor,通过统一创建、销毁和管理线程,为业务中使用线程提供保障。
优点:
- 降低资源消耗:通过复用已存在的线程,降低线程创建和销毁造成的性能开销。
- 提高响应速度:业务不需要等待线程创建。
- 统一管理:统一分配、监控、监控,防止线程无上限创建导致内存耗尽。
构造函数
public ThreadPoolExecutor(
int corePoolSize, // 核心线程数
int maximumPoolSize, // 最大线程数
long keepAliveTime, // 空闲临时线程存活时间
TimeUnit unit, // 时间单位
BlockingQueue<Runnable> workQueue, // 任务阻塞队列
ThreadFactory threadFactory, // 线程工厂
RejectedExecutionHandler handler // 拒绝策略
)execute()
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) { // 1.检查工作线程数是否 < corePoolSize
if (addWorker(command, true)) // 工作线程数 < corePoolSize,创建核心线程
return;
c = ctl.get();
} // 工作线程数 ≥ corePoolSize,核心线程已满
if (isRunning(c) && workQueue.offer(command)) { // 2.尝试放入任务阻塞队列
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false)) // 3.入队失败(队列已满),尝试创建非核心线程
// (工作线程数是否 < maximumPoolSize)
reject(command); // 4.工作线程数 > maximumPoolSize,触发拒绝策略
}拒绝策略
JDK提供了4种内置拒绝策略:
- AbortPolicy(默认):直接抛出
RejectedExecutionException异常。让上游主动处理异常。 - CallerRunsPolicy:由提交任务的线程直接执行该任务。
- 优点:负反馈机制:提交任务的线程手动执行任务,暂时无法提交新任务,为线程池提供缓冲时间。
- DiscardPolicy:静默丢弃当前任务,不处理也不抛异常。非关键任务可用。
- DiscardOldestPolicy:丢弃队列中等待时间最长的任务,然后重试当前任务。新任务会覆盖旧任务时可用。
相关机制
ctl
ctl是一个AtomicInteger类型的变量。
- 高3位:表示当前线程的运行状态。
- 低29位:表示当前的工作线程数。
通过一次CAS操作就可以同时更新线程池的状态和线程数量,保证了并发安全性和性能。
线程池状态
- RUNNING:接受新任务,处理队列任务。
- SHUTDOWN:不接受新任务,但继续处理队列中已有的任务(调用
shutdown())。 - STOP:不接受新任务,中断所有正在执行的任务,且抛弃队列任务(调用
shutdownNow())。 - TIDYING:所有任务已终止,有效线程为0,即将调用
terminated()方法。 - TERMINATED:
terminated()方法执行完成。
线程复用
线程池将线程封装为内部类Worker,继承AQS的同时实现了Runnable接口。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
final Thread thread; // 真正的工作线程
Runnable firstTask; // 创建时带的首个任务(可为 null)
}Worker通过runWorker通过getTask()循环不断从阻塞队列中获取任务。
getTask:
- 如果是核心线程(且没有设置allowCoreThreadTimeOut=true),调用
workQueue.take(),线程会无限期阻塞,直到有新任务唤醒线程。 - 如果是非核心线程,调用
workQueue.poll(keepAliveTime, unit),带超时阻塞,超时后返回null,退出循环,进入销毁流程。
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // 允许被中断(变为空闲)
boolean completedAbruptly = true;
try { // 执行首个任务或不断从阻塞队列中获取任务
while (task != null || (task = getTask()) != null) {
w.lock(); // 加锁标明 Worker 正在工作
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run(); // 直接调用 run(),不是 start()
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}问题
为什么严禁使用Executors创建线程池?
Executors.newFixedThreadPool()和Executors.newSingleThreadExecutor()的底层工作队列是无界的LinkedBlockingQueue(默认容量为Integer.MAX_VALUE),当请求速度大于处理速度时,会导致OOM。Executors.newCachedThreadPool()的最大线程数为Integer.MAX_VALUE,高并发下会无限创建线程,导致OOM。
线程池执行任务时任务抛出异常会怎么样?
execute:会销毁。在runWorker中会被catch并抛出,然后在finally中执行processWorkerExit。在processWorkerExit中,会将当前异常死亡的Worker从workers中移除,并调用addWorker(null, false)补充创建一个新的worker线程。submit:不会销毁。submit把任务封装为了FutureTask,FutureTask.run()内部用trycatch拦截了异常,并将异常存入内部字段,工作线程感知不到异常。
线程池大小怎么设置?
- CPU密集型:CPU核数+1。当某个线程偶尔因为缺页中断或其他原因暂停时,CPU处于空闲状态,此时多一个线程可继续工作。
- IO密集型:CPU核数x2或CPU核数 / (1 - 等待时间/计算时间)。IO大部分时间需要等待,需要更多的线程提高并发。
- 动态线程池:实际生产中,静态配置无法应对动态的业务场景。
ThreadPoolExecutor原生提供了setCorePoolSize()和setMaximumPoolSize()方法,允许运行时动态修改线程池大小。通过动态线程池框架(如Hippo4j、Dynamic-TP等),将核心参数推送至配置中心(如Nacos),并通过Prometheus/Grafana等监控实施状态,实现动态调整线程池大小。