前言
在我们日常工作中，我们经常会用到多线程的来处理一些异步任务，提高系统吞吐量，但是线程毕竟是一种昂贵的系统的资源，我们不应该频繁的去申请销毁。在java的project loom还未正式release的时候，我们常用池化的方式来使用线程。所以了解线程池的设计思想能够有助于我们更好的使用它，所以本文将会去走读一下java的ThreadPoolExecutor的源码。在这之前我们也有必要读一下FutureTask。因为我们submit提交的任务会被封装成FutureTask去执行，而我们也能够通过FutureTask获取到执行的结果。
准备环境
jdk1.8
FutureTask源码分析
在正式阅读之前我们可以带着一些问题去阅读源码。
- FutureTask用来解决什么问题的? 为什么会出现?
- FutureTask类结构关系怎么样的?
- FutureTask的线程安全是由什么保证的?
- FutureTask结果返回机制?
- FutureTask内部运行状态的转变?
- FutureTask通常会怎么用? 举例说明。
类图关系如下：
可以看到FutureTask实现了RunnableFuture接口，而RunnableFuture接口继承了Runnable接口和Future接口，所以FutureTask既能当做一个Runnable直接被Thread执行，也能作为Future用来得到Callable的计算结果。
在Future中为我们设计了如下的方法
cancel():用来取消异步任务的执行。如果异步任务已经完成或者已经被取消，或者由于某些原因不能取消，则会返回false。如果任务还没有被执行，则会返回true并且异步任务不会被执行。如果任务已经开始执行了但是还没有执行完成，若mayInterruptIfRunning为true，则会立即中断执行任务的线程并返回true，若mayInterruptIfRunning为false，则会返回true且不会中断任务执行线程。isCanceled():判断任务是否被取消，如果任务在结束(正常执行结束或者执行异常结束)前被取消则返回true，否则返回false。isDone():判断任务是否已经完成，如果完成则返回true，否则返回false。需要注意的是：任务执行过程中发生异常、任务被取消也属于任务已完成，也会返回true。get():获取任务执行结果，如果任务还没完成则会阻塞等待直到任务执行完成。如果任务被取消则会抛出CancellationException异常，如果任务执行过程发生异常则会抛出ExecutionException异常，如果阻塞等待过程中被中断则会抛出InterruptedException异常。get(long timeout,Timeunit unit):带超时时间的get()版本，如果阻塞等待过程中超时则会抛出TimeoutException异常。
成员属性
state:表示当前的task状态NEW:当前任务尚未执行COMPLETING:当前任务正在结束，尚未完全结束，一种临界状态NORMAL:当前任务正常结束EXCEPTIONAL:当前任务执行过程中发生了异常。内部封装的callable.call（）向上抛出异常了CANCELLED:当前任务被取消INTERRUPTING:当前任务中段中INTERRUPTED:当前任务已中断callable:submit(runnable/callable) runnable使用装饰者模式伪装成Callable了outcome:正常情况下：任务正常执行结束，outcome保存执行结果,callable返回值;非正常情况：callable向上抛出异常，outcome保存异常runner:当前任务被线程执行期间，保存了当前执行任务的线程对象引用waiters:因为会有很多线程去get当前任务的结果，所以这里使用一种单项链表数据结构
// 表示当前的task状态
private volatile int state;
// 当前任务尚未执行
private static final int NEW = 0;
// 当前任务正在结束，尚未完全结束，一种临界状态
private static final int COMPLETING = 1;
// 当前任务正常结束
private static final int NORMAL = 2;
// 当前任务执行过程中发生了异常。内部封装的callable.call（）向上抛出异常了
private static final int EXCEPTIONAL = 3;
// 当前任务被取消
private static final int CANCELLED = 4;
// 当前任务中段中
private static final int INTERRUPTING = 5;
// 当前任务已中断
private static final int INTERRUPTED = 6;
// submit(runnable/callable) runnable使用装饰者模式伪装成Callable了
private Callable<V> callable;
// 正常情况下：任务正常执行结束，outcome保存执行结果。callable返回值
// 非正常情况：callable向上抛出异常，outcome保存异常
private Object outcome; // non-volatile, protected by state reads/writes
// 当前任务被线程执行期间，保存了当前执行任务的线程对象引用
private volatile Thread runner;
// 因为会有很多线程去get当前任务的结果，所以这里使用一种数据结构stack
private volatile WaitNode waiters;
构造方法
FutureTask 有两个构造方法，一个接受callable，一个接受一个runnable.
在callable的构造方法中会首先将任务状态置为NEW，然后将传入的任务存在callable这个成员变量中。
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW;
}
在runnable的构造方法中会首先将runnable转换为一个callable，再进行赋值操作。
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
简单的查看下这个转换操作，实际上是使用适配器的方式将runnable转换为一个callable，而call方法返回的结果即为我们传入的result。
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
static final class RunnableAdapter<T> implements Callable<T> {
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run();
return result;
}
}
run()方法
在new了一个FutureTask对象之后，接下来就是在另一个线程中执行这个Task,无论是通过直接new一个Thread还是通过线程池，执行的都是run()方法，接下来就看看run()方法的实现。
运行任务，如果任务状态为NEW状态，则利用CAS修改为当前线程。执行完毕调用set(result)方法设置执行结果。
// submit(runnable/callable) -> newTaskFor(runable)->execute(task)->pool
// 任务执行的入口
public void run() {
// 条件一：state!=NEW 说明当前的task已经被执行过了，或者是被cancel了，总之非NEW状态的任务，线程就不处理了。
// 条件二：!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()) 条件成立：cas失败，当前任务被其他线程抢占
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
// 执行到这里，当前的task是NEW状态，当前的线程也抢占task成功。
try {
// callable即为我们传入的业务callable/runable
Callable<V> c = callable;
// 条件一：c!=null 防止空指针
// 条件二：state==NEW 防止外部线程cancel掉当前任务
if (c != null && state == NEW) {
// 结果引用
V result;
// true表示callable.run代码块执行成功没抛出异常
boolean ran;
try {
// 调用业务的callable/runable
result = c.call();
// 未抛出异常 ran会被设置为true
ran = true;
} catch (Throwable ex) {
// 发生异常会将结果设为null
result = null;
ran = false;
setException(ex);
}
if (ran)
// 说明当前c.call正常执行结束，设置结果
set(result);
}
} finally {
runner = null;
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
set(result)方法如下：
首先利用cas修改state状态为COMPLETING，设置返回结果，然后使用UNSAFE.putOrderedInt的方式设置state状态为NORMAL。结果设置完毕后，调用finishCompletion()方法唤醒等待线程
protected void set(V v) {
// 使用cas方式设置当前任务为完成中...
// 有没有可能失败？ 外部线程等不及了，直接在set执行cas之前将task取消了
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
// 将结果赋值给outcome之后马上将会当前的任务状态修改为正常结束状态
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
finishCompletion()方法如下：
private void finishCompletion() {
// assert state > COMPLETING;
// q执行waiters链表的头节点
for (WaitNode q; (q = waiters) != null;) {
// 使用cas设置waiters为null 是因为怕外部线程使用cancel取消当前任务 业务触发finishCompletion方法
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
// 获取当前node节点封装的线程
Thread t = q.thread;
if (t != null) {
// 条件成立：说明当前线程不为null
q.thread = null; //help gc
// 唤醒当前节点对应的线程
LockSupport.unpark(t);
}
// 获取当前节点的下一个节点
WaitNode next = q.next;
if (next == null)
// 条件成立：当前节点为最后一个节点
break;
q.next = null; // unlink to help gc
// 将当前节点指向下一个节点
q = next;
}
break;
}
}
// 可扩展的点
done();
callable = null; // to reduce footprint
}
get()方法
FutureTask 通过get()方法获取任务执行结果。如果任务处于未完成的状态(state <= COMPLETING)，就调用awaitDone方法(后面单独讲解)等待任务完成。任务完成后，通过report方法获取执行结果或抛出执行期间的异常。
// 场景：多个线程等待当前任务执行完成后的结果。。。
public V get() throws InterruptedException, ExecutionException {
// 当前的任务状态
int s = state;
// 条件成立：未执行、正在执行、正完成。调用get的外部线程会被阻塞在get方法上
if (s <= COMPLETING)
// 返回task当前状态，可能当前线程在里面已经睡了一会了
s = awaitDone(false, 0L);
return report(s);
}
report(s)方法如下：
正常情况下 outcome 保存的是callable运行结束的结果，非正常情况下保存的是callable抛出的异常
@SuppressWarnings("unchecked")
private V report(int s) throws ExecutionException {
// 正常情况下 outcome 保存的是callable运行结束的结果
// 非正常情况下保存的是callable抛出的异常
Object x = outcome;
if (s == NORMAL)
// 条件成立当前任务状态正常结束直接返回结果
return (V)x;
if (s >= CANCELLED)
// 被取消状态
throw new CancellationException();
// 执行到这里说明业务逻辑内抛出了异常
throw new ExecutionException((Throwable)x);
}
awaitDone(boolean timed, long nanos) 方法：用于等待任务完成，或任务因为中断或超时而终止。通过自旋的方式来出队，q == null 条件成立，表示第一次自旋当前的线程还未创WaitNode对象，此时为当前线程创建WaitNode对象；!queued 条件成立，第二次自旋当前线程创建了WaitNode对象但是还未入队，则当前线程的node节点的next指向原队列的头节点，waiters一直指向队列的头！再用cas方式设置waiters引用指向当前线程的node，成功的话queued==ture，否则可能其他线程先你入队，最后再进入自旋，最终如果没有出队，则get操作会park掉当前线程，除非其他线程讲你唤醒或者当前线程中断。
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
// 0 不带超时
final long deadline = timed ? System.nanoTime() + nanos : 0L;
// 引用当前线程封装成WaitNode对象
WaitNode q = null;
// 表示当前线程waiteNode对象有没有入队/压栈
boolean queued = false;
// 自旋
for (;;) {
if (Thread.interrupted()) {
// 条件成立：说明当前线程唤醒 是被其他线程使用中断这种方式唤醒。interrupted()返回true后会将Thread的中断标记重置false
// 当前线程node出队
removeWaiter(q);
// get方法抛出中断异常
throw new InterruptedException();
}
// 假设当前线程是被其他线程使用unpark（thread）唤醒的话，会正常自旋执行下面逻辑
// 获取当前任务最新状态
int s = state;
if (s > COMPLETING) {
// 条件成立：说明当前任务已经有结果了，可能是好，可能是坏
if (q != null)
// 说明已经为当前线程创建了node，此时需要将node.thred指向null 为了help GC
q.thread = null;
// 返回当前状态
return s;
}
else if (s == COMPLETING) // cannot time out yet
// 条件成立：说明当前任务接近完成状态 这里再让当前线程释放cpu，进行下一次抢占cpu
Thread.yield();
else if (q == null)
// 条件成立 第一次自旋，当前线程还未创建WaitNode对象，此时为当前线程创建WaitNode对象
q = new WaitNode();
else if (!queued)
{
// 条件成立：第二次自旋，当前线程创建了WaitNode对象但是还未入队
// 当前线程的node节点的next指向原队列的头节点，waiters一直指向队列的头！
q.next = waiters;
// cas方式设置waiters引用指向当前线程的node，成功的话queued==ture，否则可能其他线程先你入队
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
waiters, q);
}
else if (timed) {
// 第三次自旋回来到这里
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
// 当前get操作的线程会被park掉，线程回变为WAITING状态，相当于休眠了。
// 除非有其他线程将你唤醒 或者当前线程中断
LockSupport.park(this);
}
}
removeWaiter(WaitNode node)方法：
首先我们将需要移除的node内持有的thread标记为空，然后进入自旋。定义了三个变量 pred，q，s 其中q赋值为node的队列，当这个q为null时候退出循环，每次循环结束将q的值赋值为s，进入for循环将s的值赋值为队列的下一个节点，如果q的thread == null说明这个节点即为我们标记的节点，如果pred == null所说我们移除的节点为队首节点，如果上一个节点不为空者将上一个节点的下一个节点指向当前节点的下一个节点，从而使得当前节点出队，如果上一个节点被其他线程标记为需要拿出去的节点我们现在这个线程再继续往后遍历就没有什么意义了，所以这时就调到retry处，从头再遍历。如果q.thread ！=null 那么我们就将pred指向队列，并且将s也就是q的下一个节点指向为q，进行第二次循环，直到找到我们标记的节点
// 从队列中移除节点
private void removeWaiter(WaitNode node) {
if (node != null) {
// 标记我们需要移除的节点
node.thread = null;
retry:
// 自旋操作
for (;;) {
// 定义了三个变量 pred，q，s 其中q赋值为node的队列，当这个q为null时候退出循环，每次循环结束将q的值赋值为s
for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
// 首先进入for循环将s的值赋值为队列的下一个节点
s = q.next;
// 如果q的thread==null说明这个节点即为我们标记的节点
if (q.thread != null)
// 如果q.thread ！=null 那么我们就将pred指向队列，并且将s也就是q的下一个节点指向为q，进行第二次循环，直到找到我们标记的节点
pred = q;
// 如果pred也==null所说我们移除的节点为队首节点
else if (pred != null) {
// 如果上一个节点不为空者将上一个节点的下一个节点指向当前节点的下一个节点，从而使得当前节点出队
pred.next = s;
// 如果上一个节点被其他线程标记为需要拿出去的节点我们现在这个线程再继续往后遍历就没有什么意义了，所以这时就调到retry处，从头再遍历。
if (pred.thread == null) // check for race
continue retry;
}
//用cas操作将栈顶元素设置成原栈顶节点的下一个节点。操作成功结束循环
else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
q, s))
continue retry;
}
break;
}
}
}
cancel(boolean mayInterruptIfRunning) 方法
尝试取消任务如果任务已经完成或已经被取消，此操作会失败
public boolean cancel(boolean mayInterruptIfRunning) {
// 条件一：state == NEW 表示当前任务处于运行中 或者处于线程池任务队列中
// 条件二： UNSAFE.compareAndSwapInt(this, stateOffset, NEW,mayInterruptIfRunning ? INTERRUPTING : CANCELLED)) 条件成立说明修改状态成功 可以执行下面的逻辑否则返回false cancel失败
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try {
if (mayInterruptIfRunning) {
try {
// 执行当前FutureTask的线程，有可能是null，表示当前任务在队列中表示还没有线程获取到这个任务。
Thread t = runner;
if (t != null)
// 如果正在执行给runner一个中断信号
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
// 唤醒所有get()阻塞线程。
finishCompletion();
}
return true;
}
完整源代码如下：
package java.util.concurrent;
import java.util.concurrent.locks.LockSupport;
public class FutureTask<V> implements RunnableFuture<V> {
// 表示当前的task状态
private volatile int state;
// 当前任务尚未执行
private static final int NEW = 0;
// 当前任务正在结束，尚未完全结束，一种临界状态
private static final int COMPLETING = 1;
// 当前任务正常结束
private static final int NORMAL = 2;
// 当前任务执行过程中发生了异常。内部封装的callable.call（）向上跑出异常了
private static final int EXCEPTIONAL = 3;
// 当前任务被取消
private static final int CANCELLED = 4;
// 当前任务中段中
private static final int INTERRUPTING = 5;
// 当前任务已中断
private static final int INTERRUPTED = 6;
// submit(runnable/callable) runnable使用装饰者模式伪装成Callable了
private Callable<V> callable;
// 正常情况下：任务正常执行结束，outcome保存执行结果。callable返回值
// 非正常情况：callable向上抛出异常，outcome保存异常
private Object outcome; // non-volatile, protected by state reads/writes
// 当前任务被线程执行期间，保存了当前执行任务的线程对象引用
private volatile Thread runner;
// 因为会有很多线程去get当前任务的结果，所以这里使用一种数据结构stack
private volatile WaitNode waiters;
@SuppressWarnings("unchecked")
private V report(int s) throws ExecutionException {
// 正常情况下 outcome 保存的是callable运行结束的结果
// 非正常情况下保存的是callable抛出的异常
Object x = outcome;
if (s == NORMAL)
// 条件成立当前任务状态正常结束直接返回结果
return (V)x;
if (s >= CANCELLED)
// 被取消状态
throw new CancellationException();
// 执行到这里说明业务逻辑内抛出了异常
throw new ExecutionException((Throwable)x);
}
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
// callable就是自己实现的callable业务类
this.callable = callable;
// 设置任务状态为new
this.state = NEW; // ensure visibility of callable
}
public FutureTask(Runnable runnable, V result) {
// 使用适配器模式将runnable转换为了callable接口，外部线程通过get获取当前任务执行的结果，结果可能会为null或者传进来的结果
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
public boolean isCancelled() {
return state >= CANCELLED;
}
public boolean isDone() {
return state != NEW;
}
public boolean cancel(boolean mayInterruptIfRunning) {
// 条件一：state == NEW 表示当前任务处于运行中 或者处于线程池任务队列中
// 条件二： UNSAFE.compareAndSwapInt(this, stateOffset, NEW,mayInterruptIfRunning ? INTERRUPTING : CANCELLED)) 条件成立说明修改状态成功 可以执行下面的逻辑否则返回false cancel失败
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try {
if (mayInterruptIfRunning) {
try {
// 执行当前FutureTask的线程，有可能是null，表示当前任务在队列中表示还没有线程获取到这个任务。
Thread t = runner;
if (t != null)
// 如果正在执行给runner一个中断信号
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
// 唤醒所有get()阻塞线程。
finishCompletion();
}
return true;
}
// 场景：多个线程等待当前任务执行完成后的结果。。。
public V get() throws InterruptedException, ExecutionException {
// 当前的任务状态
int s = state;
// 条件成立：未执行、正在执行、正完成。调用get的外部线程会被阻塞在get方法上
if (s <= COMPLETING)
// 返回task当前状态，可能当前线程在里面已经睡了一会了
s = awaitDone(false, 0L);
return report(s);
}
/**
* @throws CancellationException {@inheritDoc}
*/
public V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
if (unit == null)
throw new NullPointerException();
int s = state;
if (s <= COMPLETING &&
(s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
throw new TimeoutException();
return report(s);
}
protected void done() { }
protected void set(V v) {
// 使用cas方式设置当前任务为完成中...
// 有没有可能失败？ 外部线程等不及了，直接在set执行cas之前将task取消了
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
// 将结果赋值给outcome之后马上将会当前的任务状态修改为正常结束状态
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
protected void setException(Throwable t) {
// 使用cas方式设置当前任务为完成中...
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
// outcome引用的是callable向上抛出的异常
outcome = t;
// 将当前状态改为异常
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
// submit(runnable/callable) -> newTaskFor(runable)->execute(task)->pool
// 任务执行的入口
public void run() {
// 条件一：state!=NEW 说明当前的task已经被执行过了，或者是被cancel了，总之非NEW状态的任务，线程就不处理了。
// 条件二：!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()) 条件成立：cas失败，当前任务被其他线程抢占
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
// 执行到这里，当前的task是NEW状态，当前的线程也抢占task成功。
try {
// callable即为我们传入的业务callable/runable
Callable<V> c = callable;
// 条件一：c!=null 防止空指针
// 条件二：state==NEW 防止外部线程cancel掉当前任务
if (c != null && state == NEW) {
// 结果引用
V result;
// true表示callable.run代码块执行成功没抛出异常
boolean ran;
try {
// 调用业务的callable/runable
result = c.call();
// 未抛出异常 ran会被设置为true
ran = true;
} catch (Throwable ex) {
// 发生异常会将结果设为null
result = null;
ran = false;
setException(ex);
}
if (ran)
// 说明当前c.call正常执行结束，设置结果
set(result);
}
} finally {
runner = null;
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
/**
* Executes the computation without setting its result, and then
* resets this future to initial state, failing to do so if the
* computation encounters an exception or is cancelled. This is
* designed for use with tasks that intrinsically execute more
* than once.
*
* @return {@code true} if successfully run and reset
*/
protected boolean runAndReset() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return false;
boolean ran = false;
int s = state;
try {
Callable<V> c = callable;
if (c != null && s == NEW) {
try {
c.call(); // don't set result
ran = true;
} catch (Throwable ex) {
setException(ex);
}
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
return ran && s == NEW;
}
/**
* Ensures that any interrupt from a possible cancel(true) is only
* delivered to a task while in run or runAndReset.
*/
private void handlePossibleCancellationInterrupt(int s) {
if (s == INTERRUPTING)
while (state == INTERRUPTING)
Thread.yield(); // wait out pending interrupt
}
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
private void finishCompletion() {
// assert state > COMPLETING;
// q执行waiters链表的头节点
for (WaitNode q; (q = waiters) != null;) {
// 使用cas设置waiters为null 是因为怕外部线程使用cancel取消当前任务 业务触发finishCompletion方法
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
// 获取当前node节点封装的线程
Thread t = q.thread;
if (t != null) {
// 条件成立：说明当前线程不为null
q.thread = null; //help gc
// 唤醒当前节点对应的线程
LockSupport.unpark(t);
}
// 获取当前节点的下一个节点
WaitNode next = q.next;
if (next == null)
// 条件成立：当前节点为最后一个节点
break;
q.next = null; // unlink to help gc
// 将当前节点指向下一个节点
q = next;
}
break;
}
}
// 可扩展的点
done();
callable = null; // to reduce footprint
}
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
// 0 不带超时
final long deadline = timed ? System.nanoTime() + nanos : 0L;
// 引用当前线程封装成WaitNode对象
WaitNode q = null;
// 表示当前线程waiteNode对象有没有入队/压栈
boolean queued = false;
// 自旋
for (;;) {
if (Thread.interrupted()) {
// 条件成立：说明当前线程唤醒 是被其他线程使用中断这种方式唤醒。interrupted()返回true后会将Thread的中断标记重置false
// 当前线程node出队
removeWaiter(q);
// get方法抛出中断异常
throw new InterruptedException();
}
// 假设当前线程是被其他线程使用unpark（thread）唤醒的话，会正常自旋执行下面逻辑
// 获取当前任务最新状态
int s = state;
if (s > COMPLETING) {
// 条件成立：说明当前任务已经有结果了，可能是好，可能是坏
if (q != null)
// 说明已经为当前线程创建了node，此时需要将node.thred指向null 为了help GC
q.thread = null;
// 返回当前状态
return s;
}
else if (s == COMPLETING) // cannot time out yet
// 条件成立：说明当前任务接近完成状态 这里再让当前线程释放cpu，进行下一次抢占cpu
Thread.yield();
else if (q == null)
// 条件成立 第一次自旋，当前线程还未创建WaitNode对象，此时为当前线程创建WaitNode对象
q = new WaitNode();
else if (!queued)
{
// 条件成立：第二次自旋，当前线程创建了WaitNode对象但是还未入队
// 当前线程的node节点的next指向原队列的头节点，waiters一直指向队列的头！
q.next = waiters;
// cas方式设置waiters引用指向当前线程的node，成功的话queued==ture，否则可能其他线程先你入队
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
waiters, q);
}
else if (timed) {
// 第三次自旋回来到这里
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
// 当前get操作的线程会被park掉，线程回变为WAITING状态，相当于休眠了。
// 除非有其他线程将你唤醒 或者当前线程中断
LockSupport.park(this);
}
}
// 从队列中移除节点
private void removeWaiter(WaitNode node) {
if (node != null) {
// 标记我们需要移除的节点
node.thread = null;
retry:
// 自旋操作
for (;;) {
// 定义了三个变量 pred，q，s 其中q赋值为node的队列，当这个q为null时候退出循环，每次循环结束将q的值赋值为s
for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
// 首先进入for循环将s的值赋值为队列的下一个节点
s = q.next;
// 如果q的thread==null说明这个节点即为我们标记的节点
if (q.thread != null)
// 如果q.thread ！=null 那么我们就将pred指向队列，并且将s也就是q的下一个节点指向为q，进行第二次循环，直到找到我们标记的节点
pred = q;
// 如果pred也==null所说我们移除的节点为队首节点
else if (pred != null) {
// 如果上一个节点不为空者将上一个节点的下一个节点指向当前节点的下一个节点，从而使得当前节点出队
pred.next = s;
// 如果上一个节点被其他线程标记为需要拿出去的节点我们现在这个线程再继续往后遍历就没有什么意义了，所以这时就调到retry处，从头再遍历。
if (pred.thread == null) // check for race
continue retry;
}
//用cas操作将栈顶元素设置成原栈顶节点的下一个节点。操作成功结束循环
else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
q, s))
continue retry;
}
break;
}
}
}
// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long stateOffset;
private static final long runnerOffset;
private static final long waitersOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = FutureTask.class;
stateOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("state"));
runnerOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("runner"));
waitersOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("waiters"));
} catch (Exception e) {
throw new Error(e);
}
}
}
ThreadPoolExecutor源码分析
类图关系如下：
public class ThreadPoolExecutor extends AbstractExecutorService {
// 高3位：表示当前线程池运行状态 除去高三位后的低位：表示当前线程池所拥有的线程数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 表示ctl中 低COUNT_BITS位 是用于存放线程数量的位
private static final int COUNT_BITS = Integer.SIZE - 3;
// 低COUNT_BITS位 所能表达的最大数值
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 1110 0000 0000 0000 0000 0000 0000 0000转换成整数为一个负数
private static final int RUNNING = -1 << COUNT_BITS;
// 0000 0000 0000 0000 0000 0000 0000 0000
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 0010 0000 0000 0000 0000 0000 0000 0000
private static final int STOP = 1 << COUNT_BITS;
// 0100 0000 0000 0000 0000 0000 0000 0000
private static final int TIDYING = 2 << COUNT_BITS;
//0110 0000 0000 0000 0000 0000 0000 0000
private static final int TERMINATED = 3 << COUNT_BITS;
// 获取当前线程池运行状态
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 获取当前线程池线程数量
private static int workerCountOf(int c) { return c & CAPACITY; }
// 用在重置当前线程池ctl值会用到
// rs 表示线程池状态 wc表示当前线程池中worker（线程）数量
private static int ctlOf(int rs, int wc) { return rs | wc; }
// 比较当前线程池ctl所表示的状态是否小于某个状态s
// 所有情况下 RUNNING<SHUTDOWN<STOP<TIDYING<TERMINATED
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
// 比较当前线程池ctl所表示的状态是否大于等于于某个状态s
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
// 小于SHUTDOWN一定是RUNNING
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
// 使用cas方式让ctl值加1 成功返回true失败返回false
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
// 使用cas方式让ctl值减1 成功返回true失败返回false
private boolean compareAndDecrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect - 1);
}
// 将ctl值减1，一定保证成功 如果失败就一直重试
private void decrementWorkerCount() {
do {} while (! compareAndDecrementWorkerCount(ctl.get()));
}
// 任务队列，当线程池中的线程数达到核心线程数时，再提交任务就会提交到workQueue;
private final BlockingQueue<Runnable> workQueue;
// 线程池全局锁，增加worker减少worker时候，修改线程运行状态时候 需要持有锁
private final ReentrantLock mainLock = new ReentrantLock();
// 线程池中真正存放 worker->thread的地方
private final HashSet<Worker> workers = new HashSet<Worker>();
// 当外部线程调用 awaitTermination()方法时，外部线程会等待当前线程池状态为Termination为止。就是将外部线程封装成waitNode放入Condition队列中。waitNode.Thread就是外部线程会被park掉
// 当线程池状态变为Termination时，回去唤醒这些线程。通过termination.signalAll()，唤醒后这些线程会进入到阻塞队列。头节点会去抢占mainLock.抢占到线程会去继续执行awaitTermination()后面的程序。
private final Condition termination = mainLock.newCondition();
// 记录线程池生命周期内线程数最大值
private int largestPoolSize;
// 记录线程池所完成任务总数，当worker退出时会将worker完成的总数累计到count上
private long completedTaskCount;
// 创建线程时会使用线程工程，一般不建议使用Default线程工程
private volatile ThreadFactory threadFactory;
// 拒绝策略 juc包下提供了四种 默认采用Abort... 抛出异常这种
private volatile RejectedExecutionHandler handler;
// 空闲线程存活时间 allowCoreThreadTimeOut==false 会维护核心线程数量内的线程存活，超出的会被回收，allowCoreThreadTimeOut==true，核心线程空闲时也会被回收
private volatile long keepAliveTime;
// 控制核心线程数内的线程是否可以被回收
private volatile boolean allowCoreThreadTimeOut;
// 核心线程数量限制
private volatile int corePoolSize;
// 线程池最大线程数限制
private volatile int maximumPoolSize;
// 缺省的拒绝策略 为抛出异常的方式
private static final RejectedExecutionHandler defaultHandler =
new AbortPolicy();
private static final RuntimePermission shutdownPerm =
new RuntimePermission("modifyThread");
private final AccessControlContext acc;
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
// Worker采用AQS的独占模式
// 独占模式：两个重要的属性 state和ExclusiveOwnerThread
// state:0时 表示未被占用 >0时 表示被占用 <0时 表示初始状态 这种情况不能被抢锁
// ExclusiveOwnerThread：表示独占锁的线程
private static final long serialVersionUID = 6138294804551838833L;
// worker内封装的工作线程
final Thread thread;
// 假设firstTask不为空，那么当worker启动后（内部的线程启动）会优先执行firstTask，当执行完firstTask后回去queue 中拉去下一个任务
Runnable firstTask;
// 当前worker所完成任务数量
volatile long completedTasks;
// firstTask可以wznull。为null启动后回去queue中获取
Worker(Runnable firstTask) {
// 设置aqs独占模式为初始状态 不能被抢占锁
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 使用线程工厂创建了一个线程，并且将当前worker指定为Runnable，也就是说当thread启动时会以work.run()为入口
this.thread = getThreadFactory().newThread(this);
}
// 当work启动时会执行run方法
public void run() {
// ThreadPoolExecutor->runWorker()
runWorker(this);
}
// 判断当前worker的独占锁是否被独占
protected boolean isHeldExclusively() {
return getState() != 0;
}
// 尝试去占用worker的独占锁
protected boolean tryAcquire(int unused) {
// 使用cas修改aqs的state，修改成功表示抢占成功，那么则设置setExclusiveOwnerThread为当前线程
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
// 尝试去释放锁。外部不会直接调用这个方法 这个方法是aqs内部调用的
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
// 加锁 加锁失败时，会阻塞当前线程，直到获取到锁为止。
public void lock() { acquire(1); }
// 尝试去加锁。如果当前锁未被持有则返回成功
public boolean tryLock() { return tryAcquire(1); }
// 一般情况下 我们调用unlock要保证当前线程是持有锁的
// 特殊情况下 当woker的state==-1的时 调用unlock表示初始化state设置state==0
// 启动worker之前会先调用unlock()这个方法。会强制刷新ExclusiveOwnerThread=null state=0
public void unlock() { release(1); }
// 返回当前worker的lock是否被占用
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
private void advanceRunState(int targetState) {
for (;;) {
int c = ctl.get();
if (runStateAtLeast(c, targetState) ||
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
break;
}
}
/**
* Transitions to TERMINATED state if either (SHUTDOWN and pool
* and queue empty) or (STOP and pool empty). If otherwise
* eligible to terminate but workerCount is nonzero, interrupts an
* idle worker to ensure that shutdown signals propagate. This
* method must be called following any action that might make
* termination possible -- reducing worker count or removing tasks
* from the queue during shutdown. The method is non-private to
* allow access from ScheduledThreadPoolExecutor.
*/
final void tryTerminate() {
for (;;) {
int c = ctl.get();
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
/*
* Methods for controlling interrupts to worker threads.
*/
/**
* If there is a security manager, makes sure caller has
* permission to shut down threads in general (see shutdownPerm).
* If this passes, additionally makes sure the caller is allowed
* to interrupt each worker thread. This might not be true even if
* first check passed, if the SecurityManager treats some threads
* specially.
*/
private void checkShutdownAccess() {
SecurityManager security = System.getSecurityManager();
if (security != null) {
security.checkPermission(shutdownPerm);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers)
security.checkAccess(w.thread);
} finally {
mainLock.unlock();
}
}
}
/**
* Interrupts all threads, even if active. Ignores SecurityExceptions
* (in which case some threads may remain uninterrupted).
*/
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers)
w.interruptIfStarted();
} finally {
mainLock.unlock();
}
}
/**
* Interrupts threads that might be waiting for tasks (as
* indicated by not being locked) so they can check for
* termination or configuration changes. Ignores
* SecurityExceptions (in which case some threads may remain
* uninterrupted).
*
* @param onlyOne If true, interrupt at most one worker. This is
* called only from tryTerminate when termination is otherwise
* enabled but there are still other workers. In this case, at
* most one waiting worker is interrupted to propagate shutdown
* signals in case all threads are currently waiting.
* Interrupting any arbitrary thread ensures that newly arriving
* workers since shutdown began will also eventually exit.
* To guarantee eventual termination, it suffices to always
* interrupt only one idle worker, but shutdown() interrupts all
* idle workers so that redundant workers exit promptly, not
* waiting for a straggler task to finish.
*/
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
/**
* Common form of interruptIdleWorkers, to avoid having to
* remember what the boolean argument means.
*/
private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}
private static final boolean ONLY_ONE = true;
/*
* Misc utilities, most of which are also exported to
* ScheduledThreadPoolExecutor
*/
/**
* Invokes the rejected execution handler for the given command.
* Package-protected for use by ScheduledThreadPoolExecutor.
*/
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}
/**
* Performs any further cleanup following run state transition on
* invocation of shutdown. A no-op here, but used by
* ScheduledThreadPoolExecutor to cancel delayed tasks.
*/
void onShutdown() {
}
/**
* State check needed by ScheduledThreadPoolExecutor to
* enable running tasks during shutdown.
*
* @param shutdownOK true if should return true if SHUTDOWN
*/
final boolean isRunningOrShutdown(boolean shutdownOK) {
int rs = runStateOf(ctl.get());
return rs == RUNNING || (rs == SHUTDOWN && shutdownOK);
}
/**
* Drains the task queue into a new list, normally using
* drainTo. But if the queue is a DelayQueue or any other kind of
* queue for which poll or drainTo may fail to remove some
* elements, it deletes them one by one.
*/
private List<Runnable> drainQueue() {
BlockingQueue<Runnable> q = workQueue;
ArrayList<Runnable> taskList = new ArrayList<Runnable>();
q.drainTo(taskList);
if (!q.isEmpty()) {
for (Runnable r : q.toArray(new Runnable[0])) {
if (q.remove(r))
taskList.add(r);
}
}
return taskList;
}
/*
* Methods for creating, running and cleaning up after workers
*/
/**
* 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:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
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(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
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 {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());
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;
}
/**
* Rolls back the worker thread creation.
* - removes worker from workers, if present
* - decrements worker count
* - rechecks for termination, in case the existence of this
* worker was holding up termination
*/
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w);
decrementWorkerCount();
tryTerminate();
} finally {
mainLock.unlock();
}
}
/**
* Performs cleanup and bookkeeping for a dying worker. Called
* only from worker threads. Unless completedAbruptly is set,
* assumes that workerCount has already been adjusted to account
* for exit. This method removes thread from worker set, and
* possibly terminates the pool or replaces the worker if either
* it exited due to user task exception or if fewer than
* corePoolSize workers are running or queue is non-empty but
* there are no workers.
*
* @param w the worker
* @param completedAbruptly if the worker died due to user exception
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
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();
}
tryTerminate();
int c = ctl.get();
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
}
addWorker(null, false);
}
}
/**
* Performs blocking or timed wait for a task, depending on
* current configuration settings, or returns null if this worker
* must exit because of any of:
* 1. There are more than maximumPoolSize workers (due to
* a call to setMaximumPoolSize).
* 2. The pool is stopped.
* 3. The pool is shutdown and the queue is empty.
* 4. This worker timed out waiting for a task, and timed-out
* workers are subject to termination (that is,
* {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
* both before and after the timed wait, and if the queue is
* non-empty, this worker is not the last thread in the pool.
*
* @return task, or null if the worker must exit, in which case
* workerCount is decremented
*/
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
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;
}
}
}
/**
* Main worker run loop. Repeatedly gets tasks from queue and
* executes them, while coping with a number of issues:
*
* 1. We may start out with an initial task, in which case we
* don't need to get the first one. Otherwise, as long as pool is
* running, we get tasks from getTask. If it returns null then the
* worker exits due to changed pool state or configuration
* parameters. Other exits result from exception throws in
* external code, in which case completedAbruptly holds, which
* usually leads processWorkerExit to replace this thread.
*
* 2. Before running any task, the lock is acquired to prevent
* other pool interrupts while the task is executing, and then we
* ensure that unless pool is stopping, this thread does not have
* its interrupt set.
*
* 3. Each task run is preceded by a call to beforeExecute, which
* might throw an exception, in which case we cause thread to die
* (breaking loop with completedAbruptly true) without processing
* the task.
*
* 4. Assuming beforeExecute completes normally, we run the task,
* gathering any of its thrown exceptions to send to afterExecute.
* We separately handle RuntimeException, Error (both of which the
* specs guarantee that we trap) and arbitrary Throwables.
* Because we cannot rethrow Throwables within Runnable.run, we
* wrap them within Errors on the way out (to the thread's
* UncaughtExceptionHandler). Any thrown exception also
* conservatively causes thread to die.
*
* 5. After task.run completes, we call afterExecute, which may
* also throw an exception, which will also cause thread to
* die. According to JLS Sec 14.20, this exception is the one that
* will be in effect even if task.run throws.
*
* The net effect of the exception mechanics is that afterExecute
* and the thread's UncaughtExceptionHandler have as accurate
* information as we can provide about any problems encountered by
* user code.
*
* @param w the worker
*/
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
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
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);
}
}
// Public constructors and methods
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters and default thread factory and rejected execution handler.
* It may be more convenient to use one of the {@link Executors} factory
* methods instead of this general purpose constructor.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue} is null
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters and default rejected execution handler.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param threadFactory the factory to use when the executor
* creates a new thread
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue}
* or {@code threadFactory} is null
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, defaultHandler);
}
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters and default thread factory.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param handler the handler to use when execution is blocked
* because the thread bounds and queue capacities are reached
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue}
* or {@code handler} is null
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param threadFactory the factory to use when the executor
* creates a new thread
* @param handler the handler to use when execution is blocked
* because the thread bounds and queue capacities are reached
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue}
* or {@code threadFactory} or {@code handler} is null
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
/**
* Executes the given task sometime in the future. The task
* may execute in a new thread or in an existing pooled thread.
*
* If the task cannot be submitted for execution, either because this
* executor has been shutdown or because its capacity has been reached,
* the task is handled by the current {@code RejectedExecutionHandler}.
*
* @param command the task to execute
* @throws RejectedExecutionException at discretion of
* {@code RejectedExecutionHandler}, if the task
* cannot be accepted for execution
* @throws NullPointerException if {@code command} is null
*/
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();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);
}
/**
* Initiates an orderly shutdown in which previously submitted
* tasks are executed, but no new tasks will be accepted.
* Invocation has no additional effect if already shut down.
*
* <p>This method does not wait for previously submitted tasks to
* complete execution. Use {@link #awaitTermination awaitTermination}
* to do that.
*
* @throws SecurityException {@inheritDoc}
*/
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
/**
* Attempts to stop all actively executing tasks, halts the
* processing of waiting tasks, and returns a list of the tasks
* that were awaiting execution. These tasks are drained (removed)
* from the task queue upon return from this method.
*
* <p>This method does not wait for actively executing tasks to
* terminate. Use {@link #awaitTermination awaitTermination} to
* do that.
*
* <p>There are no guarantees beyond best-effort attempts to stop
* processing actively executing tasks. This implementation
* cancels tasks via {@link Thread#interrupt}, so any task that
* fails to respond to interrupts may never terminate.
*
* @throws SecurityException {@inheritDoc}
*/
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);
interruptWorkers();
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
public boolean isShutdown() {
return ! isRunning(ctl.get());
}
/**
* Returns true if this executor is in the process of terminating
* after {@link #shutdown} or {@link #shutdownNow} but has not
* completely terminated. This method may be useful for
* debugging. A return of {@code true} reported a sufficient
* period after shutdown may indicate that submitted tasks have
* ignored or suppressed interruption, causing this executor not
* to properly terminate.
*
* @return {@code true} if terminating but not yet terminated
*/
public boolean isTerminating() {
int c = ctl.get();
return ! isRunning(c) && runStateLessThan(c, TERMINATED);
}
public boolean isTerminated() {
return runStateAtLeast(ctl.get(), TERMINATED);
}
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (;;) {
if (runStateAtLeast(ctl.get(), TERMINATED))
return true;
if (nanos <= 0)
return false;
nanos = termination.awaitNanos(nanos);
}
} finally {
mainLock.unlock();
}
}
/**
* Invokes {@code shutdown} when this executor is no longer
* referenced and it has no threads.
*/
protected void finalize() {
SecurityManager sm = System.getSecurityManager();
if (sm == null || acc == null) {
shutdown();
} else {
PrivilegedAction<Void> pa = () -> { shutdown(); return null; };
AccessController.doPrivileged(pa, acc);
}
}
/**
* Sets the thread factory used to create new threads.
*
* @param threadFactory the new thread factory
* @throws NullPointerException if threadFactory is null
* @see #getThreadFactory
*/
public void setThreadFactory(ThreadFactory threadFactory) {
if (threadFactory == null)
throw new NullPointerException();
this.threadFactory = threadFactory;
}
/**
* Returns the thread factory used to create new threads.
*
* @return the current thread factory
* @see #setThreadFactory(ThreadFactory)
*/
public ThreadFactory getThreadFactory() {
return threadFactory;
}
/**
* Sets a new handler for unexecutable tasks.
*
* @param handler the new handler
* @throws NullPointerException if handler is null
* @see #getRejectedExecutionHandler
*/
public void setRejectedExecutionHandler(RejectedExecutionHandler handler) {
if (handler == null)
throw new NullPointerException();
this.handler = handler;
}
/**
* Returns the current handler for unexecutable tasks.
*
* @return the current handler
* @see #setRejectedExecutionHandler(RejectedExecutionHandler)
*/
public RejectedExecutionHandler getRejectedExecutionHandler() {
return handler;
}
/**
* Sets the core number of threads. This overrides any value set
* in the constructor. If the new value is smaller than the
* current value, excess existing threads will be terminated when
* they next become idle. If larger, new threads will, if needed,
* be started to execute any queued tasks.
*
* @param corePoolSize the new core size
* @throws IllegalArgumentException if {@code corePoolSize < 0}
* @see #getCorePoolSize
*/
public void setCorePoolSize(int corePoolSize) {
if (corePoolSize < 0)
throw new IllegalArgumentException();
int delta = corePoolSize - this.corePoolSize;
this.corePoolSize = corePoolSize;
if (workerCountOf(ctl.get()) > corePoolSize)
interruptIdleWorkers();
else if (delta > 0) {
// We don't really know how many new threads are "needed".
// As a heuristic, prestart enough new workers (up to new
// core size) to handle the current number of tasks in
// queue, but stop if queue becomes empty while doing so.
int k = Math.min(delta, workQueue.size());
while (k-- > 0 && addWorker(null, true)) {
if (workQueue.isEmpty())
break;
}
}
}
/**
* Returns the core number of threads.
*
* @return the core number of threads
* @see #setCorePoolSize
*/
public int getCorePoolSize() {
return corePoolSize;
}
/**
* Starts a core thread, causing it to idly wait for work. This
* overrides the default policy of starting core threads only when
* new tasks are executed. This method will return {@code false}
* if all core threads have already been started.
*
* @return {@code true} if a thread was started
*/
public boolean prestartCoreThread() {
return workerCountOf(ctl.get()) < corePoolSize &&
addWorker(null, true);
}
/**
* Same as prestartCoreThread except arranges that at least one
* thread is started even if corePoolSize is 0.
*/
void ensurePrestart() {
int wc = workerCountOf(ctl.get());
if (wc < corePoolSize)
addWorker(null, true);
else if (wc == 0)
addWorker(null, false);
}
/**
* Starts all core threads, causing them to idly wait for work. This
* overrides the default policy of starting core threads only when
* new tasks are executed.
*
* @return the number of threads started
*/
public int prestartAllCoreThreads() {
int n = 0;
while (addWorker(null, true))
++n;
return n;
}
/**
* Returns true if this pool allows core threads to time out and
* terminate if no tasks arrive within the keepAlive time, being
* replaced if needed when new tasks arrive. When true, the same
* keep-alive policy applying to non-core threads applies also to
* core threads. When false (the default), core threads are never
* terminated due to lack of incoming tasks.
*
* @return {@code true} if core threads are allowed to time out,
* else {@code false}
*
* @since 1.6
*/
public boolean allowsCoreThreadTimeOut() {
return allowCoreThreadTimeOut;
}
/**
* Sets the policy governing whether core threads may time out and
* terminate if no tasks arrive within the keep-alive time, being
* replaced if needed when new tasks arrive. When false, core
* threads are never terminated due to lack of incoming
* tasks. When true, the same keep-alive policy applying to
* non-core threads applies also to core threads. To avoid
* continual thread replacement, the keep-alive time must be
* greater than zero when setting {@code true}. This method
* should in general be called before the pool is actively used.
*
* @param value {@code true} if should time out, else {@code false}
* @throws IllegalArgumentException if value is {@code true}
* and the current keep-alive time is not greater than zero
*
* @since 1.6
*/
public void allowCoreThreadTimeOut(boolean value) {
if (value && keepAliveTime <= 0)
throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
if (value != allowCoreThreadTimeOut) {
allowCoreThreadTimeOut = value;
if (value)
interruptIdleWorkers();
}
}
/**
* Sets the maximum allowed number of threads. This overrides any
* value set in the constructor. If the new value is smaller than
* the current value, excess existing threads will be
* terminated when they next become idle.
*
* @param maximumPoolSize the new maximum
* @throws IllegalArgumentException if the new maximum is
* less than or equal to zero, or
* less than the {@linkplain #getCorePoolSize core pool size}
* @see #getMaximumPoolSize
*/
public void setMaximumPoolSize(int maximumPoolSize) {
if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)
throw new IllegalArgumentException();
this.maximumPoolSize = maximumPoolSize;
if (workerCountOf(ctl.get()) > maximumPoolSize)
interruptIdleWorkers();
}
/**
* Returns the maximum allowed number of threads.
*
* @return the maximum allowed number of threads
* @see #setMaximumPoolSize
*/
public int getMaximumPoolSize() {
return maximumPoolSize;
}
/**
* Sets the time limit for which threads may remain idle before
* being terminated. If there are more than the core number of
* threads currently in the pool, after waiting this amount of
* time without processing a task, excess threads will be
* terminated. This overrides any value set in the constructor.
*
* @param time the time to wait. A time value of zero will cause
* excess threads to terminate immediately after executing tasks.
* @param unit the time unit of the {@code time} argument
* @throws IllegalArgumentException if {@code time} less than zero or
* if {@code time} is zero and {@code allowsCoreThreadTimeOut}
* @see #getKeepAliveTime(TimeUnit)
*/
public void setKeepAliveTime(long time, TimeUnit unit) {
if (time < 0)
throw new IllegalArgumentException();
if (time == 0 && allowsCoreThreadTimeOut())
throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
long keepAliveTime = unit.toNanos(time);
long delta = keepAliveTime - this.keepAliveTime;
this.keepAliveTime = keepAliveTime;
if (delta < 0)
interruptIdleWorkers();
}
/**
* Returns the thread keep-alive time, which is the amount of time
* that threads in excess of the core pool size may remain
* idle before being terminated.
*
* @param unit the desired time unit of the result
* @return the time limit
* @see #setKeepAliveTime(long, TimeUnit)
*/
public long getKeepAliveTime(TimeUnit unit) {
return unit.convert(keepAliveTime, TimeUnit.NANOSECONDS);
}
/* User-level queue utilities */
/**
* Returns the task queue used by this executor. Access to the
* task queue is intended primarily for debugging and monitoring.
* This queue may be in active use. Retrieving the task queue
* does not prevent queued tasks from executing.
*
* @return the task queue
*/
public BlockingQueue<Runnable> getQueue() {
return workQueue;
}
/**
* Removes this task from the executor's internal queue if it is
* present, thus causing it not to be run if it has not already
* started.
*
* <p>This method may be useful as one part of a cancellation
* scheme. It may fail to remove tasks that have been converted
* into other forms before being placed on the internal queue. For
* example, a task entered using {@code submit} might be
* converted into a form that maintains {@code Future} status.
* However, in such cases, method {@link #purge} may be used to
* remove those Futures that have been cancelled.
*
* @param task the task to remove
* @return {@code true} if the task was removed
*/
public boolean remove(Runnable task) {
boolean removed = workQueue.remove(task);
tryTerminate(); // In case SHUTDOWN and now empty
return removed;
}
/**
* Tries to remove from the work queue all {@link Future}
* tasks that have been cancelled. This method can be useful as a
* storage reclamation operation, that has no other impact on
* functionality. Cancelled tasks are never executed, but may
* accumulate in work queues until worker threads can actively
* remove them. Invoking this method instead tries to remove them now.
* However, this method may fail to remove tasks in
* the presence of interference by other threads.
*/
public void purge() {
final BlockingQueue<Runnable> q = workQueue;
try {
Iterator<Runnable> it = q.iterator();
while (it.hasNext()) {
Runnable r = it.next();
if (r instanceof Future<?> && ((Future<?>)r).isCancelled())
it.remove();
}
} catch (ConcurrentModificationException fallThrough) {
// Take slow path if we encounter interference during traversal.
// Make copy for traversal and call remove for cancelled entries.
// The slow path is more likely to be O(N*N).
for (Object r : q.toArray())
if (r instanceof Future<?> && ((Future<?>)r).isCancelled())
q.remove(r);
}
tryTerminate(); // In case SHUTDOWN and now empty
}
/* Statistics */
/**
* Returns the current number of threads in the pool.
*
* @return the number of threads
*/
public int getPoolSize() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Remove rare and surprising possibility of
// isTerminated() && getPoolSize() > 0
return runStateAtLeast(ctl.get(), TIDYING) ? 0
: workers.size();
} finally {
mainLock.unlock();
}
}
/**
* Returns the approximate number of threads that are actively
* executing tasks.
*
* @return the number of threads
*/
public int getActiveCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int n = 0;
for (Worker w : workers)
if (w.isLocked())
++n;
return n;
} finally {
mainLock.unlock();
}
}
/**
* Returns the largest number of threads that have ever
* simultaneously been in the pool.
*
* @return the number of threads
*/
public int getLargestPoolSize() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
return largestPoolSize;
} finally {
mainLock.unlock();
}
}
/**
* Returns the approximate total number of tasks that have ever been
* scheduled for execution. Because the states of tasks and
* threads may change dynamically during computation, the returned
* value is only an approximation.
*
* @return the number of tasks
*/
public long getTaskCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
long n = completedTaskCount;
for (Worker w : workers) {
n += w.completedTasks;
if (w.isLocked())
++n;
}
return n + workQueue.size();
} finally {
mainLock.unlock();
}
}
/**
* Returns the approximate total number of tasks that have
* completed execution. Because the states of tasks and threads
* may change dynamically during computation, the returned value
* is only an approximation, but one that does not ever decrease
* across successive calls.
*
* @return the number of tasks
*/
public long getCompletedTaskCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
long n = completedTaskCount;
for (Worker w : workers)
n += w.completedTasks;
return n;
} finally {
mainLock.unlock();
}
}
/**
* Returns a string identifying this pool, as well as its state,
* including indications of run state and estimated worker and
* task counts.
*
* @return a string identifying this pool, as well as its state
*/
public String toString() {
long ncompleted;
int nworkers, nactive;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
ncompleted = completedTaskCount;
nactive = 0;
nworkers = workers.size();
for (Worker w : workers) {
ncompleted += w.completedTasks;
if (w.isLocked())
++nactive;
}
} finally {
mainLock.unlock();
}
int c = ctl.get();
String rs = (runStateLessThan(c, SHUTDOWN) ? "Running" :
(runStateAtLeast(c, TERMINATED) ? "Terminated" :
"Shutting down"));
return super.toString() +
"[" + rs +
", pool size = " + nworkers +
", active threads = " + nactive +
", queued tasks = " + workQueue.size() +
", completed tasks = " + ncompleted +
"]";
}
/* Extension hooks */
/**
* Method invoked prior to executing the given Runnable in the
* given thread. This method is invoked by thread {@code t} that
* will execute task {@code r}, and may be used to re-initialize
* ThreadLocals, or to perform logging.
*
* <p>This implementation does nothing, but may be customized in
* subclasses. Note: To properly nest multiple overridings, subclasses
* should generally invoke {@code super.beforeExecute} at the end of
* this method.
*
* @param t the thread that will run task {@code r}
* @param r the task that will be executed
*/
protected void beforeExecute(Thread t, Runnable r) { }
/**
* Method invoked upon completion of execution of the given Runnable.
* This method is invoked by the thread that executed the task. If
* non-null, the Throwable is the uncaught {@code RuntimeException}
* or {@code Error} that caused execution to terminate abruptly.
*
* <p>This implementation does nothing, but may be customized in
* subclasses. Note: To properly nest multiple overridings, subclasses
* should generally invoke {@code super.afterExecute} at the
* beginning of this method.
*
* <p><b>Note:</b> When actions are enclosed in tasks (such as
* {@link FutureTask}) either explicitly or via methods such as
* {@code submit}, these task objects catch and maintain
* computational exceptions, and so they do not cause abrupt
* termination, and the internal exceptions are <em>not</em>
* passed to this method. If you would like to trap both kinds of
* failures in this method, you can further probe for such cases,
* as in this sample subclass that prints either the direct cause
* or the underlying exception if a task has been aborted:
*
* <pre> {@code
* class ExtendedExecutor extends ThreadPoolExecutor {
* // ...
* protected void afterExecute(Runnable r, Throwable t) {
* super.afterExecute(r, t);
* if (t == null && r instanceof Future<?>) {
* try {
* Object result = ((Future<?>) r).get();
* } catch (CancellationException ce) {
* t = ce;
* } catch (ExecutionException ee) {
* t = ee.getCause();
* } catch (InterruptedException ie) {
* Thread.currentThread().interrupt(); // ignore/reset
* }
* }
* if (t != null)
* System.out.println(t);
* }
* }}</pre>
*
* @param r the runnable that has completed
* @param t the exception that caused termination, or null if
* execution completed normally
*/
protected void afterExecute(Runnable r, Throwable t) { }
/**
* Method invoked when the Executor has terminated. Default
* implementation does nothing. Note: To properly nest multiple
* overridings, subclasses should generally invoke
* {@code super.terminated} within this method.
*/
protected void terminated() { }
/* Predefined RejectedExecutionHandlers */
/**
* A handler for rejected tasks that runs the rejected task
* directly in the calling thread of the {@code execute} method,
* unless the executor has been shut down, in which case the task
* is discarded.
*/
public static class CallerRunsPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code CallerRunsPolicy}.
*/
public CallerRunsPolicy() { }
/**
* Executes task r in the caller's thread, unless the executor
* has been shut down, in which case the task is discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
/**
* A handler for rejected tasks that throws a
* {@code RejectedExecutionException}.
*/
public static class AbortPolicy implements RejectedExecutionHandler {
/**
* Creates an {@code AbortPolicy}.
*/
public AbortPolicy() { }
/**
* Always throws RejectedExecutionException.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
* @throws RejectedExecutionException always
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
/**
* A handler for rejected tasks that silently discards the
* rejected task.
*/
public static class DiscardPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardPolicy}.
*/
public DiscardPolicy() { }
/**
* Does nothing, which has the effect of discarding task r.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
/**
* A handler for rejected tasks that discards the oldest unhandled
* request and then retries {@code execute}, unless the executor
* is shut down, in which case the task is discarded.
*/
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardOldestPolicy} for the given executor.
*/
public DiscardOldestPolicy() { }
/**
* Obtains and ignores the next task that the executor
* would otherwise execute, if one is immediately available,
* and then retries execution of task r, unless the executor
* is shut down, in which case task r is instead discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
}