jdk1.8 ThreadPoolExecutor
2018-11-25 本文已影响0人
timar
菜鸟一枚。理解不对的地方，望能直接指出~
1、线程池的5种状态
RUNNING：运行中，正常接受新的任务
SHUTDOWN：关闭状态，不接受新的task，但是会继续处理workQueue中的任务
STOP：停止状态，不接受新的task，并立即停止所有的工作线程
TIDYING：整理状态，此时workCount为0，workQueue为空
TERMINATED ：线程池已经完全中止（workCount = 0 ， workQueue is empty）
状态的流转
- RUNNING -> SHUTDOWN 调用了shutdown()
- RUNNING -> STOP 调用了shutdownNow()
- SHUTDOWN -> TIDYING 最后一个任务处理完，程序退出时，调用tryTerminate()
- STOP -> TIDYING 调用shutdownNow->所有线程停止->TIDYING
- TIDING -> TERMINATED TIDING之后，执行terminated()后，会立即修改为TERMINATED
2、成员变量介绍
// 任务队列
private final BlockingQueue<runnable> workQueue;
// 当操作workers集合,以及获取largestPoolSize/poolSzie/completedTaskCount值,以及shutdown(),shutdownNow()等时会用到
private final ReentrantLock mainLock = new ReentrantLock();
//存放所有的worker，一个worker绑定一个线程
private final HashSet<worker> workers = new HashSet<worker>();
// ？
private final Condition termination = mainLock.newCondition();
//线程池曾经达到的最大线程数量，只是记录一个最大值
private int largestPoolSize;
// 线程池已经执行的任务数量，成功或失败都会+1，但是并非是实时修改的，只有当一个worker停止的时候才会累加，实时修改的是worker中的一个属性->completedTasks
private long completedTaskCount;
// 线程工厂，在new Worker()的时候会由线程工厂创建一个线程
private volatile ThreadFactory threadFactory;
// 队列满了之后执行的拒绝策略
private volatile RejectedExecutionHandler handler;
// 达到coreThreadCount之后，从队列中获取任务的超时时间，为0表示立即获得，否则返回null
private volatile long keepAliveTime;
//在线程数量没有达到coreThreadCount时，从队列中获取是否需要使用超时时间
// 默认是false，函数allowCoreThreadTimeOut(value)可以更改值
private volatile boolean allowCoreThreadTimeOut;
// 最小线程数量，但是初始化时不会创建worker，可以为0
private volatile int corePoolSize;
// 最大线程数量，大于0
private volatile int maximumPoolSize;
// 默认拒绝策略,直接抛出RejectedExecutionException
private static final RejectedExecutionHandler defaultHandler =
new AbortPolicy();
//
private static final RuntimePermission shutdownPerm =
new RuntimePermission("modifyThread");
3、 ctl 线程池控制标志
// Integer类型值，前3位保存运行状态，后29位保存线程数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 29
private static final int COUNT_BITS = Integer.SIZE - 3;
// 2^29 - 1，二进制表示为 000111111（后面29个1）
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// ~CAPACITY二进制表示为 1110000000(后面29个0)
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 计算工作线程数
private static int workerCountOf(int c) { return c & CAPACITY; }
// 线程池状态和线程数计算得出ctl值
private static int ctlOf(int rs, int wc) { return rs | wc; }
4、execute(Runnable command)，提交新的任务到线程池中
提交的Worker实现了Runnable接口
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.
*/
分为3种情况
1、当前线程数量小于corePoolSize，即最小线程数量，直接创建新的线程
2、线程数量已经达到corePoolSize，或者创建新线程失败，则直接将任务添加进队列....
3、加入队列不成功，尝试创建一个新线程，如果创建失败，那一定是线程池停止或者饱和了
// 取ctl值
int c = ctl.get();
// 如果当前线程数小于最小线程数corePoolSize
if (workerCountOf(c) < corePoolSize) {
// 创建一个新的Worker
if (addWorker(command, true))
return;
// 创建失败，重新获取ctl值
c = ctl.get();
}
// 如果是运行中状态，将任务添加到任务队列中workQueue
if (isRunning(c) && workQueue.offer(command)) {
// 再次刷新ctl值，做二次检测
int recheck = ctl.get();
// 运行状态不是运行中，将任务从队列中取出
if (! isRunning(recheck) && remove(command))
// 执行拒绝策略
reject(command);
// 线程池状态未改变，但是工作线程数量为0，创建新的线程去处理任务队列中任务
// 例如，corePoolSize为1，但是在这个任务添加进队列之前，这个线程就已经因为获取不到新的线程而被interrupt()
// 这里worker已经添加到队列中，所以是传null，但是不太明白这里为什么用false，防止corePoolSize为0吗？
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
} // 线程池已经停止或者任务队列已经满了，尝试创建新的线程去处理这个任务，
// 这里没有STOP/TIDYING/TERMINATED等状态，是因为addWorker中会判断
else if (!addWorker(command, false))
reject(command);
}
5、addWork()，创建新的Worker(或者说线程)
/**
* Checks if a new worker can be added with respect to current
* pool state and the given bound (either core or maximum). If so,
* the worker count is adjusted accordingly, and, if possible, a
* new worker is created and started, running firstTask as its
* first task. This method returns false if the pool is stopped or
* eligible to shut down. It also returns false if the thread
* factory fails to create a thread when asked. If the thread
* creation fails, either due to the thread factory returning
* null, or due to an exception (typically OutOfMemoryError in
* Thread.start()), we roll back cleanly.
*
* @param firstTask the task the new thread should run first (or
* null if none). Workers are created with an initial first task
* (in method execute()) to bypass queuing when there are fewer
* than corePoolSize threads (in which case we always start one),
* or when the queue is full (in which case we must bypass queue).
* Initially idle threads are usually created via
* prestartCoreThread or to replace other dying workers.
*
* @param core if true use corePoolSize as bound, else
* maximumPoolSize. (A boolean indicator is used here rather than a
* value to ensure reads of fresh values after checking other pool
* state).
* @return true if successful
*/
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
// 大循环，定义breakpoint
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 这里反向看，排除了RUNNING状态，排除了SHUTDOWN 状态且workQueue不为空且传入的firstTask 为空的场景
// 因为SHUTDOWN 状态不接受新的任务，但是会继续处理队列中任务
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
// 内循环
for (;;) {
// 当前工作线程数
int wc = workerCountOf(c);
// 不大可能会超过2^29-1，或者线程数量大于了corePoolSize /maximumPoolSize，直接返回
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// cas替换，将workCount+1，成功跳出大循环
if (compareAndIncrementWorkerCount(c))
break retry;
// cas替换失败，重新检查运行状态，如果状态改变，重新执行大循环
// 状态未改变，重试内循环
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
// 执行到这里，表示cas将线程数+1已经成功了
// 本次创建的新的线程，是否启动成功了
boolean workerStarted = false;
// 本次创建的Worker是否添加到workers集合中了
boolean workerAdded = false;
Worker w = null;
try {
// 构造方法中，会用threadFactory new一个Thread和这个worker绑定
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
// 再次检查运行状态，例如变成STOP
int rs = runStateOf(ctl.get());
// 必须是RUNNING状态或者SHUTDOWN状态而且传入firstTask 为空
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// 新创建的线程已经被启动过了，报错
if (t.isAlive()) // precheck that t is startable
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;
}
6、runWorker，线程池如何处理任务
先来看看Woker类
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
// 继承AQS，实现Runnable接口
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;** Thread this worker is running in. Null if factory fails. */
// worker中实际负责处理任务的线程，由threadFactory创建
final Thread thread;
/** Initial task to run. Possibly null. */
// 创建Worker时的构造参数，可能为null（入列队了）
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 这里创建Thread时，把worker对象自己丢进去了，所以在addWork()方法中，创建Worker对象成功后，t.start()就会执行这个worker对象的run()方法
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
---------------------------分隔一下---------------------------------------
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
// 不是正在处理某一个任务中，都可以被中断，如shutdownNow()
w.unlock(); // allow interrupts
// 是否异常中断标志
boolean completedAbruptly = true;
try {
// firstTask不为空，或者从队列中取到任务了
while (task != null || (task = getTask()) != null) {
// 给自己上锁，防止被中断？
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
// 这一段也不是特别明白，根据上面的原谅注释大胆猜测一下
// 在线程池状态是STOP/TIDYING/TERMINATED时，而且当前线程未被中止，那么直接中止
// 在第一次检查时，状态还不是STOP/TIDYING/TERMINATED，此用用户调用了showDownNow()，导致当前线程已经被中断，然后进行二次检测
// 这里状态已经是STOP，而且 !wt.isInterrupted() 会返回true，因为*Thread.interrupted()会清除线程内部的中断状态*
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();
}
}
// 没有取到任务会从这里退出，设置异常结束标志为false
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
从队列中获取任务
*/
private Runnable getTask() {
// 本次获取任务是否超时标志
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// RUNNING状态 和 SHUTDOWN状态而且 workQueue 不为空除外
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
// 工作线程数 -1（取不到任务线程会interupt）
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
// allowCoreThreadTimeOut默认是false，可以人为修改，如果为true，所以线程获取任务都需要设置超时时间，为false，只有wc > corePoolSize时才会设置超时时间
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// wc > max ， 在人为调用setMaximumPoolSize(int maximumPoolSize)时可能出现
// 工作线程数大于最大线程数或者上次超时 而且 线程池中还有别的线程或者队列为空，那么将当前线程中断
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因为取不到任务退出或者异常退出干了啥
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 异常退出时，workCount -1 ，正常退出时，在getTask()中已经做了-1操作
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 线程池执行过的任务数累加（不管是否成功）
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
// *为啥要尝试终止？个人猜测，在调用shutdown之后，仍然有线程在运行，所以线程正常退出后，需要调用tryTerminate()来终止线程池*
tryTerminate();
int c = ctl.get();
// 线程状态还未到STOP
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
// 异常中断，直接new一个新的Worker代替这个异常退出的
// 正常退出，而且当前线程数为0，workQueue不为空，也创建一个新Worker
addWorker(null, false);
}
}
7、退出 shutdown() shutdownNow()
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// cas替换状态为SHUTDOWN
advanceRunState(SHUTDOWN);
// 中断空间的worker
interruptIdleWorkers();
// 空实现
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// cas替换ctl值，变更线程池状态
advanceRunState(STOP);
// 中断所有线程
interruptWorkers();
// 取出队列中所有还未执行的任务返回，即队列被清空
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
tryTerminate()被调用的地方有很多，这里以shutdown()和shutdownNow()分析
final void tryTerminate() {
for (;;) {
int c = ctl.get();
// 状态判断，如果调用 shutdown()，状态是SHUTDOWN，队列可能不为空
// 如果调用shutdownNow()，状态是STOP，会清空队列
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
// 工作线程不为0，尝试中断一个空间的线程
// shutdown：workCount可能不为0，shutdownNow一定 为0
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// cas改变状态为TIDYING，线程数为0，失败了就重试大循环
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
// 空实现
terminated();
} finally {
// cas改变状态为TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
// 不明白
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
调用tryTerminate() 的地方有机会再补
