FutureTask
使用线程池的时候，只能传入Runnable接口的对象，而Runnable接口不能获得线程的返回值，所以需要使用callable接口，所以封装一个FutureTask对象，FutureTask继承与Runnable接口和Future接口，future接口的get方法可以获取到返回值。
FutureTask的初始化可以传入runnable接口 和 callable接口，如果传入runnable接口的话使用一个RunnableAdapter适配器变成callable接口(执行call方法，实际上执行的是runnable的run方法)。
FutureTask 统一了 Runnnable 和 Callable，方便了我们后续对线程池的使用.
FutureTask对象内部维护了线程的状态、callable对象、outcome输出结果，WaitNode等待队列。
FutureTask 中等待任务返回结果的队列，都会进行等待队列里面，是一个单向队列，队列中每个结点都是WaitNode。 这个队列被当成线程安全的栈来使用的，出栈和入栈使用CAS来保证线程安全。
public class FutureTask<V> implements RunnableFuture<V> {
private volatile int state;
private static final int NEW = 0;
private static final int COMPLETING = 1;
private static final int NORMAL = 2;
private static final int EXCEPTIONAL = 3;
private static final int CANCELLED = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED = 6;
/** 要执行的任务，如果传入的是runnable，则会适配器成callable */
private Callable<V> callable;
/** 输出结果，可以通过get获得 */
private Object outcome;
/** The thread running the callable; CASed during run() */
private volatile Thread runner;
// 指向栈顶指针
private volatile WaitNode waiters;
static final class WaitNode {
volatile Thread thread;// 等待的线程
volatile WaitNode next;// next 指针
WaitNode() { thread = Thread.currentThread(); }
}
}
run方法
FutureTask的run方法是线程执行的入口
- 如果当前线程不是new状态，说明已经被执行过了，所以直接退出；
- CAS设置当前线程为运行线程，失败则退出；
- 执行FutureTask.Callable线程的call 方法，并保存结果，如果有异常，结果里面保存异常信息，并设置状态为异常状态。
- CAS设置状态为完成状态中，把结果保存到outcome，然后设置状态为normal正常状态。
CAS设置阻塞队列里面的头结点为null。- 当前节点封装的线程不为空，则赋值为null，等待GC, 然后则唤醒当前线程
public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {// 本身程序导致的中断异常
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result);// CAS设置状态为完成状态中，把结果保存到outcome，然后设置状态为normal正常状态，然后清空并唤醒阻塞队列里面的结点
}
} 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
int s = state;
if (s >= INTERRUPTING)// 判断是否是cancle 导致的中断异常
handlePossibleCancellationInterrupt(s);
}
}
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
protected void set(V v) {
// CAS设置状态为完成状态中
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;// 把结果保存到outcome，
// CAS设置状态为完成
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion(); //清空并唤醒阻塞队列里面的结点
}
}
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
// 清空WaitNode链表
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) { // 遍历链表并唤醒结点
Thread t = q.thread;
if (t != null) {
q.thread = null;
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
可能有多个线程等待获取get的结果。 如果线程状态小于等于完成中状态，则调用get的外部线程会被阻塞：
- 当前线程的等待状态小于等于完成中COMPETTING，则阻塞，
为当前线程创建WaitNode结点，并进入阻塞队列 - 阻塞当前线程，直到被唤醒或者中断唤醒
- 如果是中断唤醒，当前结点从阻塞队列出对，并抛出异常
- 如果是唤醒操作，且当前任务状态已经完成，退出
- 如果当前状态是正在完成(state==COMPETTING)，则yield释放CPU资源，进行下一次抢占CPU。
如果线程状态是正常状态，则返回结果- 如果是大于取消状态，即中断状态 ，则抛出异常
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
// 引用当前线程 封装成 WaitNode 对象
WaitNode q = null;
boolean queued = false;// 表示当前线程waitNode对象，有没有入队/压栈
for (;;) {
if (Thread.interrupted()) {// 当前线程刚被中断唤醒
removeWaiter(q); // 把当前结点从阻塞队列里面出队
throw new InterruptedException();
}
// 当前线程被unpark唤醒
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s; // 当前任务已有结果，退出
}
// 当前任务已经快完成， 当前线程释放CPU，进行下一次抢占CPU
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
//第一次自旋，为当前线程创建WaitNode对象
else if (q == null)
q = new WaitNode();
// 第二次自旋，当前q的next指向队列头节点，q结点入队，并CAS设置头结点为q
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
// 当前get操作被park，线程状态变为WAITTING状态，
//等待被其他线程唤醒，或者 当前线程中断
// 唤醒后继续自旋
LockSupport.park(this);
}
}
cancle
如果当前不为New 状态，表示正在执行，则直接退出。
如果当前线程为NEW状态，则CAS 设置状态为中断、或者取消状态，CAS设置失败则退出；
如果是中断，则执行线程中断程序，并设置状态为中断状态
public boolean cancel(boolean mayInterruptIfRunning) {
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try { // 通过中断结束任务
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}