在java中，一个线程是无法获取另外一个线程的执行结果的，那么为了解决这个问题，我们想到的可能解决方式为：使用各个线程的共用存储区域，进行数据的交互访问，但是其中将会涉及到很多需要解决的问题，例如要等待数据的线程需要等待产生数据线程的执行完毕，以及产生数据线程发生异常时，等待数据的线程的感知，那么为了更好的处理这些问题，java在1.5的时候提供了一个类FutureTask为获取线程结果提供了便捷的桥梁。
首先看一个简单应用的例子
private String executionTask(final String taskName)throws ExecutionException, InterruptedException {
while (true) {
Future<String> future = taskCache.get(taskName); // 1.1,2.1
if (future == null) {
//Callable接口线程
Callable<String> task = () -> taskName;
FutureTask<String> futureTask = new FutureTask<>(task);
future = taskCache.putIfAbsent(taskName, futureTask); // 1.3
if (future == null) {
future = futureTask;
futureTask.run(); // 1.4执行任务
}
}
try {
//获取Callable接口线程的返回值
return future.get(); // 1.5,
} catch (CancellationException e) {
taskCache.remove(taskName, future);
}
}
在上面简单的例子。我们可以看到三个重要的类
- Callable， 构造有返回值线程的方法接口
- Future接口， 构造有线程返回值的接口
-
FutureTask，
Future接口的实现类
1.Callable
Callable接口类似于Runnable接口，因为它们都是为线程执行而设计的接口。但是Runnable不返回结果，也不能抛出异常。
@FunctionalInterface
public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
* 构造一个返回值，或者抛出一个异常
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}
可以看到如果需要构造有返回值的线程那么就需要使用到Callable接口进行构造并重写其中的call()方法，使用平常的Runnable是无法返回结果的
那么构造完Callable后，应该如何执行呢，普通的Thread类是无法执行Callable的，这个时候就需要另外一个类去封装执行Callable，那么就是FutureTask了，但是要先讲讲FutureTask实现的接口Future
2.Future
Future一个表示一个异步的结果计算。提供了方法提供检查计算是否完成，取消任务的执行，以及获取执行完毕后的结果
public interface Future<V> {
/**
* 试图取消此任务的执行。这种尝试将如果任务已经完成，已经被取
*消，则失败，或因其他原因不能取消。如果成功,且调用时该任务尚未
*启动，不应该运行此任务。如果任务已经开始，然后参数
*mayInterruptIfRunning确定执行此
*任务的线程是否应该被中断试图停止任务
*/
boolean cancel(boolean mayInterruptIfRunning);
/**
*如果该任务在完成前被取消，则返回true
*/
boolean isCancelled();
/**
*判断任务是否完成
*/
boolean isDone();
/**
* 等待执行完毕的结果，并获取结果
*/
V get() throws InterruptedException, ExecutionException;
/**
* 等待执行完毕的结果，并获取结果，如果在timeout时间访问内没有处
* 理完毕，则抛出TimeoutException异常
*/
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
可以看到Future接口提供了一系列的访问去查看任务的执行状态，以及结果的返回。
3.FutureTask
作为线程的返回类，FutureTask维护着很多机制，例如等待机制，异常机制等，状态机制等。现在来一步步进行深究，看FutureTask是如何让一个线程获取另外一个线程的返回值
3.1 类继承结构
image.png
FutureTask继承了Future接口和Runnable接口，使得FutureTask具有线程执行的能力以及任务调度的能力
3.1 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; //异常
private static final int CANCELLED = 4; //取消
private static final int INTERRUPTING = 5; //中断中
private static final int INTERRUPTED = 6; //已中断
FutureTask内部维护着一个全局变量state作为任务的一个状态标记，状态的转换源码注释上给出了详细的转换流程
Possible state transitions:
- NEW -> COMPLETING -> NORMAL
- NEW -> COMPLETING -> EXCEPTIONAL
- NEW -> CANCELLED
- NEW -> INTERRUPTING -> INTERRUPTED
/** The underlying callable; nulled out after running */
/** 存放callable */
private Callable<V> callable;
/** The result to return or exception to throw from get() */
/** callable执行完毕后的返回值*/
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 */
/** 线程等待节点，当其他线程使用get方法并且future还没执行完毕时，将记录在这个节点上并且使这些线程进入等待状态 */
private volatile WaitNode waiters;
3.2 FutureTask的构造器
/** 使用callable初始化执行方法，并且将状态置为NEW*/
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
-------------------------------------------------------------------
/** 使用callable初始化执行方法，并且将状态置为NEW，以及添加指定返回值result */
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
3.3 FutureTask的重点方法
FutureTask的run()方法
public void run() {
//如果当前FutureTask状态不为NEW ，或者cas runner变量的时候失败（预防并发调用的情况），直接返回
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 {
//调用Callable的方法
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设置为空
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
//如果FutureTask状态为INTERRUPTING中断或者INTERRUPTED中断
if (s >= INTERRUPTING)
//进行处理，确保cancel方法能先执行完毕，只有cancel能将状态变为INTERRUPTING以及INTERRUPTED
handlePossibleCancellationInterrupt(s);
}
}
set(result)，FutureTask设置处理结果的方法
protected void set(V v) {
//设置COMPLETING状态，这里主要预防当前线程是否被取消
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//赋值处理结果
outcome = v;
//设置完成态
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
//唤醒等待线程
finishCompletion();
}
}
setException(ex)，FutureTask处理异常方法
protected void setException(Throwable t) {
//设置COMPLETING状态，这里主要预防当前线程是否被取消
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//设置处理结果为异常
outcome = t;
//设置异常态
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
//唤醒等待线程
finishCompletion();
}
}
从FutureTask的run中可得，FutureTask处理并发的判断主要在于runner是否为空，执行的判断主要在于FutureTask状态的改变。
get()方法，获取FutureTask的返回值，如果当FutureTask还没执行完毕，那么线程将进入等待状态
public V get() throws InterruptedException, ExecutionException {
int s = state;
//如果任务还在执行，使当前线程等待
if (s <= COMPLETING)
s = awaitDone(false, 0L);
//返回FutureTask的执行结果
return report(s);
}
那么是如何使线程进入等待状态呢awaitDone(false, 0L)
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;
}
//如果当前状态正在处理，那么使当前线程让步，让FutureTask执行完毕
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);
}
}
report(int s)方法，返回处理结果或者异常
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);
}
取消FutureTask执行,cancel(boolean mayInterruptIfRunning)
public boolean cancel(boolean mayInterruptIfRunning) {
//当FutureTask状态为NEW时才能进行取消或者中断
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try { // in case call to interrupt throws exception
//如果设置中断参数为true，那么将当前执行线程进行中断处理
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
//唤醒等待节点
finishCompletion();
}
return true;
}
从cancel方法中可以看到，cancel的结果会取决于参数mayInterruptIfRunning，当mayInterruptIfRunning为false时，只能取消未开始执行的任务，因为在run()方法中是不会对CANCELLED进行任何的处理的，但是如果mayInterruptIfRunning为true时，当run()方法正在执行，并且状态还没进行改变时，可以让FutureTask中断。
但是执行cannel（false）时，如果run方法已经开始执行并且已经走过了状态NEW的判断那么，此时Callable中的方法将不会停止，同时setResult方法将无法赋值，如果执行cannel（true）时，会将执行FutureTask的线程中断，需要在执行程序中去判断执行线程是否中断，以及采用对应的处理方式，同时setResult方法也将无法赋值
runAndReset()方法，可以使FutureTask执行多次，但是不会处理处理结果
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;
}
查询状态方法：
isCancelled()任务是否被取消或者中断
isDone()任务是否完成
public boolean isCancelled() {
return state >= CANCELLED;
}
public boolean isDone() {
return state != NEW;
}
