ThreadPoolExecutor是 JDK1.5版本推出的一个线程池。是 ExecutorService接口的实现之一，也是阿里推荐使用的一种线程池。线程池解决了两个不同的问题：
ThreadPoolExecutor会根据 corePoolSize和 maximumPoolSize参数自动调整线程池的大小。当方法 execute(Runnable)提交一个新任务的时候，如果运行的线程数量小于 corePoolSize，即使存在空闲的线程也会创建一个新的线程来处理请求，如果运行的线程数量大于 corePoolSize但小于 maximumPoolSize时，并且队列已经满了，则会创建一个新的线程执行任务。 corePoolSize和 maximumPoolSize的数量也可以通过 setCorePoolSize和 setMaximumPoolSize动态的修改。
默认情况下 ThreadPoolExecutor核心线程在任务到达的时候创建和启动的。你可以使用 prestartCoreThread、prestartAllCoreThreads来动态覆盖来预先启动线程，或者使用非空队列的时候，可能需要虚线启动线程。
使用 ThreadFactory创建新的线程，如果不指定，默认使用 Executors.defaultThreadFactory，他创建的线程具有相同的 TheadGroup、相同的优先级、相同的非守护状态。通过不同的 ThreadFactory，你可以方便的更改线程的名字、线程的优先级、线程的守护状态。
当前线程池中超过了 corePoolSize数量的线程，多余的空闲线程在 keepAliveTime时间到达后将被终止。这提供了一种减少资源消耗的方法。如果稍后任务量更加大的时候，线程池又会创建新的线程执行任务。默认情况下只有线程达到了 corePoolSize数量时，才会应用 keepAliveTime策略。但是方法 allowCoreThreadTimeOut(boolean)也可以将这个超时时间策略应用在核心线程上，前提是 keepAliveTime不是0。
任何的 BlockingQueue队列都可以用来传输和提交任务。
直接交换队列的一个很好的选择是 SynchronousQueue。它本身不存储任何任务，只是将任务传递给线程，如果此时没有任何线程来执行这个任务，并且下一个任务已经到达，那么任务就排队就会失败，导致线程池创建一个新的非核心线程。直接传递需要无限的 maxmunPoolsize,以避免决绝新提交的任务。然任务的到达速度高于线程处理任务的速度，那么就会导致线程无限增长。
无界队列 LinkedBlockingQueue表示没有设置具体容量的队列，这将导致在 corePoolSize都处于繁忙时，任务都会放入队列中，因而不会创建超过 corePoolSize的线程，因此 maxmunPoolSize没有任何效果。如果任务平均到达速度高于线程处理任务的速度，将可能造成队列的任务过多导致OOM。
有界队列，例如 ArrayBlockingQueue可以使用 maxmumPoolSize防止系统的资源被耗尽，种情况下更加难以调优和控制。线程池大小和队列大小可以相互权衡，大队列小线程池可以减少 CPU的开销，但系统的吞吐量就会变低。大线程池小队列可以增加 CPU的开销，可能会遇到不可预测的调度开销，也会降低吞吐量。
当线程池已经关闭或者使用有界队列的前提下，线程的数量已经达到了 maxmunPoolSize并且有界队列已经饱和，那么通过 execute(Runnable)方法提交的新任务将被拒绝。这种情况下 execute方法会调用 RejectedExecutionHandler#rejectedExecution(Runnable,ThreadPoolExecutor)方法。提供了四个预定义的拒绝粗略：
ThreadPoolExecutor提供可覆写的 preExecute(Thread,Runbale)、afterExecute(Thread,Runnable)方法，在任务执行之前和执行之后调用。这些可以用来控制执行环境，例如可以重新初始化 ThreadLocals、收集统计数据或者增加日志。此外，还可以覆写的 terminated方法，可以在所有 Executor全部终止后执行所需要的特殊处理逻辑。
如果 Hook函数或者回调方法失败，那么内部的工作线程将会失败并突然终止。
getQueue()方法允许访问工作队列，以进行监视和调试。强烈反对将此方法用于其他目的。提供了两个 remove(Runnable)和 purge方法，可用于在大量任务被取消执行的时候进行回收存储。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
线程池控制状态ctl是一个原子整数，包含了两个概念字段：
runState提供了主要生命周期控制，接受以下值：
execute(Runnbale)
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
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();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);
}
在将来某个时间执行给定的任务，可以是新创建的线程来执行，也可以是线程池中已存在的线程来执行。如果线程池已关闭或者已经达到其最大容量，则会执行 RejectedExecutionHandler的拒绝策略。
该方法时一个核心方法，是任务提交的入口。该方法中最核心的方法就是 addWorker方法。
该方法的主要逻辑：
addWorker(Runnable,boolean)
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();
if (runStateOf(c) != rs)
continue retry;
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive())
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
该方法的主要作用就是根据当前线程池的状态和容量用给定的线程添加到worker中.如果添加成功,则调整worker的计数,并且可能会启动一个新的Worker,将 Runnable作为其第一个任务执行.如果线程池已停止或者符合关闭条件,则直接返回 fasle,如果线程工厂返回 null,或者线程创建失败,或者由于异常导致创建失败,则会回滚.
runWorker(Worker)
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();
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();
} 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);
}
}
该方法主要时循环从队列中获取任务并执行它们,同时处理一些其他的问题.
getTask()
private Runnable getTask() {
boolean timedOut = false;
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
该方法主要作用就是根据当前线程池的配置，对任务进行阻塞或者超时等待(该方法最核心的功能就是核心线程复用)。如果worker必须退出，则返回null。worker退出条件如下:
Original: https://blog.csdn.net/Suubyy/article/details/127826758
Author: 666呀
Title: 线程池ThreadPoolExecutor
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/661199/
转载文章受原作者版权保护。转载请注明原作者出处！