FutureTask源码详解
FutureTask的作用
FutureTask可以用于执行异步计算、获取线程运行的结果、可以判断程序是否执行完毕、取消任务的执行。
FutureTask源码解析
先看看这个类的一些成员变量
// 这个属性很关键，表示当前的一个状态
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;
// 表示任务未执行之前被调用了cancel(false)方法
private static final int CANCELLED = 4;
// 当任务调用cancel(true)方法中断程序时,处于中间状态
private static final int INTERRUPTING = 5;
// 当任务调用cancel(true)方法中断程序时,程序调用线程的interrupt()方法,状态由INTERRUPTING->INTERRUPTED
private static final int INTERRUPTED = 6;
// 保存Callable对象，这里使用了适配器模式
private Callable<V> callable;
// 程序执行的结果
private Object outcome;
// 正在运行的线程
private volatile Thread runner;
// 等待队列
private volatile WaitNode waiters;
// 在类加载时进行初始化
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);
}
}
状态的变化可能有以下几种情况
* NEW -> COMPLETING -> NORMAL 任务正常结束
* NEW -> COMPLETING -> EXCEPTIONAL 任务出现异常而结束
* NEW -> CANCELLED 任务被取消
* NEW -> INTERRUPTING -> INTERRUPTED 任务被中断
继承关系图
可以看到FutureTask实现了Runnable接口和Future接口，并且有个Callable类型的成员变量，在创建FutureTask的时候进行属性的赋值，我认为这里是运用了适配器模式，这里不展开细讲。
run()方法
public void run() {
// 如果状态不为NEW，或者CAS不成功，则直接返回;若CAS成功，则会将内存中偏移量为runnerOffset的值设置为当前线程
// runnerOffset代表一个内存的偏移量，不了解CAS的可以先去了解一下
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
// 这里的逻辑主要是执行callable的call()方法，获取返回值，并将结果赋值给outcome
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
// 执行call方法
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
// 如果出现异常，则将异常结果赋值给outcome
setException(ex);
}
// 如果执行成功，则将执行结果赋值给outcome
if (ran)
set(result);
}
} finally {
// 在执行结束之后，需要将runner设置为null，避免其他线程CAS不成功
runner = null;
int s = state;
// 如果当前的状态为INTERRUPTING，调用handlePossibleCancellationInterrupt方法处理中断
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
setException(Throwable t)方法
protected void setException(Throwable t) {
// 通过CAS将状态从NEW改为COMPLETING
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
// 将异常信息赋值给outcome，并将状态改为EXCEPTIONAL
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL);
finishCompletion();
}
}
finishCompletion()方法
private void finishCompletion() {
// 遍历整个等待队列
for (WaitNode q; (q = waiters) != null;) {
// 利用CAS将偏移量为waitersOffset的值从q改为null
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
// 这个循环的逻辑主要是遍历整个等待队列，并将等待队列里面的等待线程唤醒，转为就绪状态，等待获取cpu的执行权
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，借助gc回收
q.next = null;
q = next;
}
break;
}
}
// 这里是一个扩展，根据具体情况去重写
done();
callable = null; // to reduce footprint
}
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;
// 当s>COMPLETING时，都是已完成的状态
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
// 如果任务正在执行中，则让出cpu的执行权，有可能下一次循环时s的状态就是>COMPLETING
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
// 若q为null，则new一个WaitNode对象
else if (q == null)
q = new WaitNode();
// 未入队时，将q设置为等待队列的首节点
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
// timed为true时
else if (timed) {
nanos = deadline - System.nanoTime();
// 若当前时间超过了阻塞等待时间，则将此线程从等待队列中出队
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
// 若当前时间未超过阻塞等待时间，则阻塞
LockSupport.parkNanos(this, nanos);
}
// 阻塞当前线程
else
LockSupport.park(this);
}
}
removeWaiter(WaitNode node)方法
private void removeWaiter(WaitNode node) {
if (node != null) {
// 将node的thread属性设为null，便于后面出队操作
node.thread = null;
retry:
for (;;) {
for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
s = q.next;
// 若q.thread != null，则继续往下一个节点寻找thread为null的节点(因为上一个注释处已经将node的thread置为null)
if (q.thread != null)
pred = q;
// 找到要移除的节点，将该节点的上一个节点的next指向下一个节点，这样就做到从等待队列中删除该节点的效果
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;
}
}
}
WaitNode结构
// 这里其实就是对当前线程做了一层封装
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
get()方法
public V get() throws InterruptedException, ExecutionException {
int s = state;
// 若任务未执行完成，将当前线程进入等待队列
if (s <= COMPLETING)
s = awaitDone(false, 0L);
// 返回任务执行的结果
return report(s);
}
get(long timeout, TimeUnit unit)方法
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);
}
cancel(boolean mayInterruptIfRunning)方法
public boolean cancel(boolean mayInterruptIfRunning) {
// 如果CAS成功，并且mayInterruptIfRunning为true，则将状态从NEW改为INTERRUPTING，这是一个中间状态
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try {
// 如果mayInterruptIfRunning为true
if (mayInterruptIfRunning) {
try {
// 中断，并设置中断标志位，不会中断正在执行的线程
Thread t = runner;
if (t != null)
t.interrupt();
} finally {
//
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}
版权声明：本文为weixin_43927838原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接和本声明。