一.FutureTask继承接口
FutureTask是Future的具体实现。FutureTask实现了RunnableFuture接口。RunnableFuture接口又同时继承了Future 和 Runnable 接口。所以FutureTask既可以作为Runnable被线程执行，又可以作为Future得到Callable的返回值。
二.FutureTask 属性
state 其初始状态为"NEW"（新建）。任务的运行状态仅在三个方法中被转移到终止状态：set、setException和cancel。任务的完成过程中，状态可能会临时为"COMPLETING"（正在完成，当结果正在被设置时）或"INTERRUPTING"（仅当为了响应cancel(true)而中断运行者时）。
可能的状态转换如下：
- NEW -> COMPLETING -> NORMAL：从新建状态到正在完成状态，再到正常完成状态。
- NEW -> COMPLETING -> EXCEPTIONAL：从新建状态到正在完成状态，再到异常完成状态。
- NEW -> CANCELLED：从新建状态直接到取消状态。
- NEW -> INTERRUPTING -> INTERRUPTED：从新建状态到正在中断状态，再到中断状态。
这个状态设计允许任务在执行过程中根据不同的操作（如正常完成、异常完成、取消或中断）来更新其状态，并且确保了状态转换的效率和正确性。
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;
private Callable<V> callable;
private Object outcome;
private volatile Thread runner;
private volatile WaitNode waiters;
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
-
定义
Unsafe实例和字段偏移量：-
private static final sun.misc.Unsafe UNSAFE;：声明一个Unsafe类型的静态最终变量，用于后续操作。 -
private static final long stateOffset;、private static final long runnerOffset;、private static final long waitersOffset;：声明三个长整型静态最终变量，用于存储FutureTask类中state、runner、waiters字段的内存偏移量。
-
-
静态初始化块：
-
在静态初始化块中，首先尝试获取
Unsafe实例。由于Unsafe的getUnsafe()方法是受限的，通常只能在引导类加载器加载的类中调用，因此这里可能会抛出SecurityException。为了绕过这个限制，实际使用中可能需要通过反射获取Unsafe的私有构造器或私有字段来创建实例（这部分代码未展示）。 -
Class<?> k = FutureTask.class;：获取FutureTask类的Class对象。 -
通过
UNSAFE.objectFieldOffset方法获取FutureTask类中state、runner、waiters字段的内存偏移量。这些偏移量用于后续直接通过内存地址修改这些字段的值。
-
// 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);
}
}
三.核心方法
1.run方法
public void run() {
//当当前状态不是new，或者cas 操作设置成员变量thread为当前线程失败时直接返回false
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);
}
} finally {
设置成员变量thread为空
runner = null;
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}2.setException（）
发生异常时设置异常方法
(1) 通过cas操作将状态由new->completing
(2)若设置成功，则将返回结果设置为异常
同样通过cas操作将状态更新为异常完成状态
执行唤醒正在等待结果的阻塞线程
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
3.set（）
正常设置结果方法
(1) 通过cas操作将状态由new->completing
(2)若设置成功，则将返回结果设置为正常返回值
同样通过cas操作将状态更新为正常完成状态
执行唤醒正在等待结果的阻塞线程
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}4.finishcompletion()
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
//设置等待waiter为空
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
//循环遍历将等待线程逐个唤醒
for (;;) {
Thread t = q.thread;
if (t != null) {
将每个waiter中的线程都设置为空
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
}5.handlePossibleCancellationInterrupt(int s)
private void handlePossibleCancellationInterrupt(int s) {
// It is possible for our interrupter to stall before getting a
// chance to interrupt us. Let's spin-wait patiently.
if (s == INTERRUPTING)
while (state == INTERRUPTING)
Thread.yield(); // wait out pending interrupt
// assert state == INTERRUPTED;
// We want to clear any interrupt we may have received from
// cancel(true). However, it is permissible to use interrupts
// as an independent mechanism for a task to communicate with
// its caller, and there is no way to clear only the
// cancellation interrupt.
//
// Thread.interrupted();
}6.awaitDone(boolean timed, long nanos)
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
//若当前线程被中断，则移除当前等待节点，抛出异常
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
//当状态在完成中状态以后，意味着已经流程结束不需要等待，当waiter 不为空时设置线程为空清除节点，返回状态
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
//当当前状态为完成中，则当前线程让出执行，辅助流程尽快走完
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
//以上都不是则新建waiter放入等待队列
else if (q == null)
q = new WaitNode();
else if (!queued)
// 如果当前线程还没有被加入到等待队列，则尝试使用CAS（Compare-And-Swap）操作将当前节点加入到等
//待队列的头部
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
//如果等待有超时时间，则计算剩余等待时间，如果超时则移除当前节点并返回状态。如果没有超时，则调用
//LockSupport.parkNanos(this, nanos)等待指定的纳秒数。
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
//如果没有超时时间，则调用LockSupport.park(this)无限期等待，直到被唤醒。
LockSupport.park(this);
}
}7.removeWaiter(WaitNode node)
private void removeWaiter(WaitNode node) {
if (node != null) {
node.thread = null;
retry:
for (;;) { // restart on removeWaiter race
for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
s = q.next;
if (q.thread != null)
pred = q;
else if (pred != null) {
pred.next = s;
if (pred.thread == null) // check for race
continue retry;
}
else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
q, s))
continue retry;
}
break;
}
}
}8.其他简单方法
@SuppressWarnings("unchecked")
private V report(int s) throws ExecutionException {
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();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
public FutureTask(Runnable runnable, V result) {
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) {
//当当前状态为new 根据传入值设置状态，失败直接返回false
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;
}
public V get() throws InterruptedException, ExecutionException {
int s = state;
//如果当前状态为新建，将当前线程加入等待
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
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);
}