FutureTask源码分析

FutureTask是JDK1.6新增的内容,间接实现了Future, Runnable接口。前面已经介绍过了Future的get方法会阻塞在那,等待执行完成,获取结果,那么FutureTask是怎么实现这一功能的呢?

相关类简介

LockSupport

LockSupport是用来创建锁及其他同步类的基本线程阻塞元素,它的park和 unpark能够分别阻塞线程和解除线程阻塞。它提供了可以指定阻塞时长的park方法。park和unpark的基本接口为:

1
2
3
4
5
6
7
public static void park() {
unsafe.park(false, 0L);
}
public static void unpark(Thread thread) {
if (thread != null)
unsafe.unpark(thread);
}

unpark需要指定对应的线程,而park是将当前线程阻塞。其实这里也提供阻塞线程新的方式,我们已经或者用忙等待,或者用中断睡眠的方式阻塞唤醒线程,这里可以通过LockSupport实现。FutureTask就是用这种方式实现的。下面看一下unsafe,它是一个Unsafe类

Unsafe

Java不能够直接访问操作系统底层,而是通过本地方法来访问。Unsafe提供了硬件级别的原子访问,主要提供一下功能:

  1. 分配释放内存
  2. 定位某个字段的内存位置
  3. 挂起一个线程和恢复,更多的是通过LockSupport来访问。park和unpark
  4. CAS操作,比较一个对象的某个位置的内存值是否与期望值一致,一致则更新对应值,此更新是不可中断的。主要方法是compareAndSwap*。

FutureTask源码

FutureTask主要是通过这两个类来实现的,调用get的时候如果异步线程还没有执行完,就是先park当前线程,如果超时则返回,这个时候如果异步线程还是没有执行完成,则返回失败。如果异步线程执行完成了,则会调用LockSupport的unpark方法,将主线程(与异步线程对应)解除阻塞,而主线程解除阻塞后就会获取执行的结果,get函数返回结果。下面看一下get方法的源码

1
2
3
4
5
6
7
8
9
10
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);
}

它实际实现wait的是awaitDone方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
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;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING) // cannot time out yet, 已经完成了,但是状态还在COMPLETING,yield一下下次再判断。
Thread.yield();
else if (q == null) //如果队列还是为创建一个等待节点
q = new WaitNode();
else if (!queued) //如果还没有加入waiters队列,则先将waiters赋值给q.next,然后判断waiters是否已经是q.next,如果是,则将q赋值给waiters(waitersOffset对应waiters在FutureTask中的偏移量)。关键在于此一系列操作能保证一致性。
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) { //如果是需要判断超时,则使用parkNanos
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
LockSupport.park(this);
}
}

waitDone就是将当前线程加入等待队列(WaitNode有当前Thread的Thread变量),然后用LockSupport将自己阻塞,等待超时或者被解除阻塞后,判断是否已经完成(state为>= COMPLETING),如果未完成(state< COMPLETING)抛出超时异常,如果已完成则稍等或者直接返回结果。

下面看看异步线程如何唤醒当前线程,从run方法开始:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread())) //给runner赋值
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) //已经运行完则设置结果,前面一段是调用Callable运行需要做的操作
set(result);
}
} finally {
...
}
}

set方法:

1
2
3
4
5
6
7
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { //原子操作state
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();//通知线程
}
}

finishCompletion:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {//如果已经赋值给q ,则对waiter置null
for (;;) { //一个一个地unpark。
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
}

上面的代码就是遍历waiters,然后解除他们的阻塞。这样整个流程就算完了

总结

在整个FeatureTask中没有直接使用锁机制,而是通过LockSupport来阻塞线程,唤醒线程。对于多线程访问FeatureTask的waiters,state,都是采用Unsafe来操作,避免使用锁(毕竟锁其实很耗时),改为直接原子操作对应的变量。FeatureTask是一个非常好的Unsafe和LockSupport例子。