FutureTask
相较于传统线程Thread
就是可以阻塞等待返回线程执行结果, 这点在有些场景下是非常重要的.
那么来看下并发大神Doug Lea
是如何实现的吧.
继承关系 - Future接口
先看类注释
A {@code Future} represents the result of an asynchronous computation. Methods are provided to check if the computation is complete, to wait for its completion, and to retrieve the result of the computation.
-- Future 代表一个异步计算的结果, 此接口提供如下方法: 1.判断计算是否完成; 2.等待计算完成; 3.返回计算结果
The result can only be retrieved using method {@code get} when the computation has completed, blocking if necessary until it is ready.
-- 这个计算结果只能被 {@code get} 返回, 当计算完成或必要的阻塞直到其准备好(去返回).
Cancellation is performed by the {@code cancel} method. Additional methods are provided to determine if the task completed normally or was cancelled.
-- 取消任务需要调用 {@code cancel} 方法,其他的方法({@code isCancelled}, {@code isDone})用于判断此任务是否完成或被取消
Once a computation has completed, the computation cannot be cancelled.
-- 不能取消一个已经完成的任务
If you would like to use a {@code Future} for the sake of cancellability but not provide a usable result, you can declare types of the form {@code Future<?>} and return {@code null} as a result of the underlying task.
-- 如果只是想使用其可取消而不使用其返回结果, 那么可以声明 {@code Future<?>} 类型并以 {@code null} 作为其底层任务返回值 返回.
然后提供了两种用法:
ArchiveSearcher
interface ArchiveSearcher { String search(String target); }
class App {
ExecutorService executor = ...
ArchiveSearcher searcher = ...
void showSearch(final String target) throws InterruptedException {
Future<String> future = executor.submit(new Callable<String>() {
public String call() {
return searcher.search(target);
}
});
displayOtherThings(); // do other things while searching
try {
displayText(future.get()); // use future
} catch (ExecutionException ex) {
cleanup();
return;
}
}
}
FutureTask
FutureTask<String> future = new FutureTask<String>(new Callable<String>() {
public String call() {
return searcher.search(target);
}
});
executor.execute(future);
最后一段有点没看懂, 内存一致性: 异步计算的get()
方法由另一个线程调用
<p>Memory consistency effects: Actions taken by the asynchronous computation
<a href="package-summary.html#MemoryVisibility"> <i>happen-before</i></a>
actions following the corresponding {@code Future.get()} in another thread.
Future
接口提供了五个方法
/**
* Attempts to cancel execution of this task. This attempt will
* fail if the task has already completed, has already been cancelled,
* or could not be cancelled for some other reason. If successful,
* and this task has not started when {@code cancel} is called,
* this task should never run. If the task has already started,
* then the {@code mayInterruptIfRunning} parameter determines
* whether the thread executing this task should be interrupted in
* an attempt to stop the task.
*
* <p>After this method returns, subsequent calls to {@link #isDone} will
* always return {@code true}. Subsequent calls to {@link #isCancelled}
* will always return {@code true} if this method returned {@code true}.
*
* @param mayInterruptIfRunning {@code true} if the thread executing this
* task should be interrupted; otherwise, in-progress tasks are allowed
* to complete
* @return {@code false} if the task could not be cancelled,
* typically because it has already completed normally;
* {@code true} otherwise
*/
boolean cancel(boolean mayInterruptIfRunning);
/**
* Returns {@code true} if this task was cancelled before it completed
* normally.
*/
boolean isCancelled();
/**
* Returns {@code true} if this task completed.
*
* Completion may be due to normal termination, an exception, or
* cancellation -- in all of these cases, this method will return
* {@code true}.
*
* @return {@code true} if this task completed
*/
boolean isDone();
/**
* Waits if necessary for the computation to complete, and then
* retrieves its result.
*
* @return the computed result
* @throws CancellationException if the computation was cancelled
* @throws ExecutionException if the computation threw an
* exception
* @throws InterruptedException if the current thread was interrupted
* while waiting
*/
V get() throws InterruptedException, ExecutionException;
/**
* Waits if necessary for at most the given time for the computation
* to complete, and then retrieves its result, if available.
*
* @param timeout the maximum time to wait
* @param unit the time unit of the timeout argument
* @return the computed result
* @throws CancellationException if the computation was cancelled
* @throws ExecutionException if the computation threw an
* exception
* @throws InterruptedException if the current thread was interrupted
* while waiting
* @throws TimeoutException if the wait timed out
*/
V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
其注释也写的很清楚, 这里挑几点说一下:
cancel()
-
此方法将在以下情况下失败(注意这里没有未启动的状态)
- 已经完成
- 已经取消
- 其他无法被取消的原因 (这个没再细说感觉有点迷的~)
如果取消任务成功, 而此任务尚未启动, 则此任务永远不应进入运行
此方法返回
true
后, 再调用{@link #isDone}
和{@link #isCancelled}
都会返回true
mayInterruptIfRunning
这个参数就是其语义, 是否要中断已经启动的任务
isDone()
返回true
有以下几种情况
-
normal termination
正常结束 -
an exception
执行异常 -
cancellation
任务被取消
-
get()
与get(long timeout, TimeUnit unit)
就是要等待任务完成(阻塞blocking
)并返回其结果, 后者带超时时间.
实现类 - FutureTask
结构
public class FutureTask<V> implements RunnableFuture<V>
- public interface RunnableFuture<V> extends Runnable, Future<V>
照例看下类注释
A cancellable asynchronous computation. This class provides a base implementation of {@link Future}, with methods to start and cancel a computation, query to see if the computation is complete, and retrieve the result of the computation.
-- 一个可以取消的异步任务,提供了一些方法,或启动,或取消,或查询是否结束,或返回结果
The result can only be retrieved when the computation has completed; the {@code get} methods will block if the computation has not yet completed.
-- 任务完成后悔返回结果,在这之前会呗阻塞。
once the computation has completed, the computation cannot be restarted or cancelled (unless the computation is invoked using {@link #runAndReset}).
-- 一旦任务完成,此任务不能被重新启动(除非任务是通过 {@link #runAndReset}) 被调用的)
<p>A {@code FutureTask} can be used to wrap a {@link Callable} or {@link Runnable} object. Because {@code FutureTask} implements {@code Runnable}, a {@code FutureTask} can be submitted to an {@link Executor} for execution.
-- FutureTask 可以被封装成 {@link Callable} 或 {@link Runnable}
Callable 与Runnable 的关系
在Executors
工具类中有如下代码
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
/**
* A callable that runs given task and returns given 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;
}
}
- 不需要返回值可考虑调用
callable(target, null)
因为
FutureTask
实现了Runnable
接口, 其可以转化为Runnable
与Callable
对象进行操作.
继续看注释
/*
* Revision notes: This differs from previous versions of this
* class that relied on AbstractQueuedSynchronizer, mainly to
* avoid surprising users about retaining interrupt status during
* cancellation races. Sync control in the current design relies
* on a "state" field updated via CAS to track completion, along
* with a simple Treiber stack to hold waiting threads.
*
* Style note: As usual, we bypass overhead of using
* AtomicXFieldUpdaters and instead directly use Unsafe intrinsics.
*/
- 这里说了点与之前版本的不同, 使用了
AbstractQueuedSynchronizer
来处理中断相关. 通过字段state
控制.
又来了!通过
AQS
控制中断, 之前我们在线程池的Worker
中碰到过, 继续往下看FutureTask
是怎么玩的~(但源码中并没有看到AQS
的使用诶......)
- 最后一句提到通过直接使用
Unsafe
内部函数来绕过使用AtomicXFieldUpdaters
的开销
AtomicXFieldUpdaters
, 通过反射技术来对volatile
修饰的X型
属性进行原子更新, 很明显这种方式很消耗资源
列一下这几种方式CPU消耗程度,由小到大
- 基本类型
X
: 多线程不安全 - volatile
X
: 多线程读取安全,但无法进行原子操作 -
AtomicX
: 多线程读安全,可进行原子操作 -
AtomicXFieldUpdate
:使用反射,多线程安全和可原子操作
啰嗦了这么多, 到现在终于可以正儿八经地来看下FutureTask
这个类的属性了.
字段属性
state
/**
* The run state of this task, initially NEW. The run state
* transitions to a terminal state only in methods set,
* setException, and cancel. During completion, state may take on
* transient values of COMPLETING (while outcome is being set) or
* INTERRUPTING (only while interrupting the runner to satisfy a
* cancel(true)). Transitions from these intermediate to final
* states use cheaper ordered/lazy writes because values are unique
* and cannot be further modified.
*
* Possible state transitions:
* NEW -> COMPLETING -> NORMAL
* NEW -> COMPLETING -> EXCEPTIONAL
* NEW -> CANCELLED
* NEW -> INTERRUPTING -> INTERRUPTED
*/
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;
首先告诉我们初始状态是
NEW
, 然后只能通过三个方法来修改此字段值:set
/setException
/cancel
, 等会一一讲这三个方法.在任务完成(
normal
/cancelled
/interrupted
)之前,FutureTask
可能会短暂地存在于COMPLETING
(已经设置好返回值) 或INTERRUPTING
(调用cancel(true)
方法), 从这些中间状态到最终状态的转换使用更廉价的有序且惰性写入, 因为其值唯一且无法进一步更改.
廉价有序惰性, 意味着占用资源更少, 后文可以看到这一步转换都是使用
putOrderedInt
方法, 此方法执行时会插入StoreStore内存屏障,避免发生写操作重排序。与之类似的还有putOrderedObject
/putOrderedLong
, 均属于Unsafe
类.
- 其他属性
/** The underlying callable; nulled out after running */
private Callable<V> callable;
/** The result to return or exception to throw from get() */
private Object outcome; // non-volatile, protected by state reads/writes
/** The thread running the callable; CASed during run() */
private volatile Thread runner;
/** Treiber stack of waiting threads */
private volatile WaitNode waiters;
-
callable
构造传入, 真正执行的任务 -
outcome
返回结果 -
runner
当前执行线程 -
waiters
get
方法等待线程列表
/**
* Simple linked list nodes to record waiting threads in a Treiber
* stack. See other classes such as Phaser and SynchronousQueue
* for more detailed explanation.
*/
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
可见这是个单链表, 注释中提到了一个叫做Treiber Stack
的东西
Treiber Stack
, 简单而言是一个栈, 既然是栈那就有一个非常重要的特性先进后出, 这里也是, 他每次插入都是插入(push
)到链头, 而出栈(pop
)也是链头先出.详细可参考Treiber Stack
- 此链表用于记录所有的等待线程.
内部方法
构造方法
/**
* Creates a {@code FutureTask} that will, upon running, execute the
* given {@code Callable}.
*
* @param callable the callable task
* @throws NullPointerException if the callable is null
*/
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
/**
* Creates a {@code FutureTask} that will, upon running, execute the
* given {@code Runnable}, and arrange that {@code get} will return the
* given result on successful completion.
*
* @param runnable the runnable task
* @param result the result to return on successful completion. If
* you don't need a particular result, consider using
* constructions of the form:
* {@code Future<?> f = new FutureTask<Void>(runnable, null)}
* @throws NullPointerException if the runnable is null
*/
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
两种构造方法, 第一种直接使用Callable
方法, 然后返回其返回值.第二种方法将Runnable
封装成Callable
后执行并返回传入的result
.
其他方法
run()
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);
}
} 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)
handlePossibleCancellationInterrupt(s);
}
}
这个方法连个方法注释都没有, 真辣鸡~~
- 需要状态为
NEW
且runner
为空, 然后CAS
成功的线程才能成为其runner
执行当前任务 - 第二步中没看懂为啥还要再check一遍
state==NEW
, 先记着. - 然后就是执行
call()
方法, 拿到返回值result
, 这里分两种情况, 一种正常一种异常, 分别执行了set
/setException
方法, 也是上面唯三的修改state
值方法之二, 分别看看.
/**
* Sets the result of this future to the given value unless
* this future has already been set or has been cancelled.
*
* <p>This method is invoked internally by the {@link #run} method
* upon successful completion of the computation.
*
* @param v the value
*/
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
- 任务成功执行后,
CAS
修改NEW
-COMPLETING
, 第一个中间状态来了~ - 设置返回值, 然后
putOrderedInt
直接将state
值改成NORMAL
- 调用
finishCompletion()
方法
/**
* Causes this future to report an {@link ExecutionException}
* with the given throwable as its cause, unless this future has
* already been set or has been cancelled.
*
* <p>This method is invoked internally by the {@link #run} method
* upon failure of the computation.
*
* @param t the cause of failure
*/
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t;
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
- 任务异常执行后,
CAS
修改NEW
-COMPLETING
, 第一个中间状态来了~ - 设置返回值, 然后
putOrderedInt
直接将state
值改成EXCEPTIONAL
- 调用
finishCompletion()
方法
然后都是调用finishCompletion
方法
/**
* Removes and signals all waiting threads, invokes done(), and
* nulls out callable.
*/
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
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
}
- 注释说干了这么几件事: 移除并且唤醒所有等待线程, 调用
invoke
,callable
置null
- 第一层循环用于找到当前
FutureTask
的waiter
, 置null
- 第二层将所有等待线程唤醒
- 调用
done
方法,callable
置null
减少内存占用同时方便GC
done
方法是一个protected
方法, 默认是空实现, 需要子类自己实现
/**
* Protected method invoked when this task transitions to state
* {@code isDone} (whether normally or via cancellation). The
* default implementation does nothing. Subclasses may override
* this method to invoke completion callbacks or perform
* bookkeeping. Note that you can query status inside the
* implementation of this method to determine whether this task
* has been cancelled.
*/
protected void done() { }
回到run
方法的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)
handlePossibleCancellationInterrupt(s);
-
runner
在state
被设置之前必须为non-null
以防止并发调用run
方法, 因为run
方法的第一步就是一个只有runner
为null
才能成功的CAS
操作. - 上面解释了一句, 然后现在任务执行完毕, 可以将
runner
置null
- 任务完成后必须重新读取
state
状态防止遗漏中断(leaked interrupts
)
继续看下handlePossibleCancellationInterrupt
方法做了啥
/**
* 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) {
// 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();
}
一个自旋操作, 等待线程退出INTERRUPTING
状态
说到现在, waiters
这个链表好像没有用到诶~
接下来重点看下get()
方法
/**
* @throws CancellationException {@inheritDoc}
*/
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
当小于等于COMPLETING
时,调用awaitDone
方法.
前面我们在run
方法中有两个状态转换:
NEW -> COMPLETING -> NORMAL
NEW -> COMPLETING -> EXCEPTIONAL
而COMPLETING
这个状态只在这里出现过, 那就是当FutureTask
还没执行结束时, 执行awaitDone
/**
* Awaits completion or aborts on interrupt or timeout.
*
* @param timed true if use timed waits
* @param nanos time to wait, if timed
* @return state upon completion
*/
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
Thread.yield();
else if (q == null)
q = new WaitNode();
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
LockSupport.park(this);
}
}
类注释写了这个方法四种结果
- 等待执行完成
- 取消
- 中断
- 超时
处理逻辑
- 如果线程已经中断, 移除
waiter
, 抛出InterruptedException
- 如果线程已经完成, 置空线程位, 返回状态位
- 如果
COMPLETING
中, 则短暂yield
释放一下 - 然后就是将当前线程构建成
WaitNode
一直往waiters
链表头部(自旋 +CAS
, 还记得前面说的Treiber Stack
吗?) - 加入链表成功后, 挂起(分是否需要超时设置不同处理)
然后就是report
方法
/**
* Returns result or throws exception for completed task.
*
* @param s completed state value
*/
@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);
}
这个方法就很简单粗暴了, 正常结束就正常返回, 否则抛出对应异常.
还有几个状态目前我们没接触到, 就是异常与取消
NEW -> CANCELLED
NEW -> INTERRUPTING -> INTERRUPTED
来看看什么时候会用到他们
public boolean cancel(boolean mayInterruptIfRunning) {
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try { // in case call to interrupt throws exception
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}
-
mayInterruptIfRunning
为false
, 则状态改为CANCELLED
-
mayInterruptIfRunning
为true
, 则状态改为INTERRUPTING
, 然后中断其线程, 最后状态改为INTERRUPTED
- 最后都执行
finishCompletion
方法.
最后我们看下FutureTask
中的最后一个方法runAndReset
/**
* 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;
}
这个方法跟run
方法非常类似, 下面提几点不同
- 从方法名就看得出来,
runAndReset
不会设置返回值 -
runAndReset
方法有返回值, 如果执行成功且没有被中断或取消返回true
, 否则返回false
最后来讲讲 CAS
到底是个啥.
以FutureTask
为例来看下.
CAS
操作就是 compareAndSwap
, 比较然后交换, 但在多线程下这种操作是问题的, 所以只能从硬件层面上实现, 只要比较成功那么一定交换成功, 也就是说这是一个原子操作.
那么此方法需要什么呢?
看下源码
CAS
操作都是通过 UnSafe
类来实现, 从名字就知道这个类可以做很多坏事~ 慎用
public final native boolean compareAndSwapObject(Object var1, long var2, Object var4, Object var5);
-
var1
操作对象 -
var2
对象字段在对象中的偏移量 -
var4
期望值 -
var5
修改值
简而言之, 当对象字段
的值为期望值
时, 更新为修改值
可能就这个偏移量var2
不知道咋玩.
看下FutureTask
中的用法
// 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);
}
}
-
getDeclaredField
通过反射拿到字段对象Field
-
objectFieldOffset
UnSafe
的方法, 通过Field
对象拿到其所在偏移量
每个使用
CAS
的类, 就看他这一步拿了哪些字段的偏移量, 就表示需要使用CAS
操作哪些字段. 比如这里只用到了state
/runner
/waiters
这篇博客前后写了快一周, 不是细嚼慢咽, 也不是FutureTask
源码有多难啃, 纯粹是琐事太多与性子懒劲上来了.....
珍惜能静下心来学习的日子。