Runnable Callable FutureTask
Runnable
Runnable接口只有一个run函数，该函数没有返回值。Thread类在调用start（）函数后就是执行的Runnable的run（）方法。其声明如下：
public interface Runnable {
public abstract void run();
}
Callable
Callable与Runnable的功能大致相似，它有一个call方法，该方法有返回值，其申明如下：
public interface Callable<V> {
V call() throws Exception;
}
Future
Future就是对于具体的Runnable和Callable任务的执行结果进行取消、查询是否完成、获取结果、设置结果等。其中get方法会阻塞知道任务返回结果。其声明如下：
public interface Future<V> {
/**
* 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 <tt>cancel</tt> is called,
* this task should never run. If the task has already started,
* then the <tt>mayInterruptIfRunning</tt> parameter determines
* whether the thread executing this task should be interrupted in
* an attempt to stop the task. *
*/
boolean cancel(boolean mayInterruptIfRunning);
/**
* Returns <tt>true</tt> if this task was cancelled before it completed
* normally.
*/
boolean isCancelled();
/**
* Returns <tt>true</tt> if this task completed.
*
*/
boolean isDone();
/**
* Waits if necessary for the computation to complete, and then
* retrieves its result.
*
* @return the computed result
*/
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
*/
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
在Future接口中声明了5个方法，下面依次解释每个方法的作用：
-
cancel方法用来取消任务，如果取消任务成功则返回true，取消任务失败则返回false。参数mayInterruptIfRunning表示是否允许取消正在执行却没有执行完毕的任务，如果设置true，则表示可以取消正在执行过程中的任务。如果任务已经完成，则无论mayInterruptIfRunning为true还是false，此方法肯定返回false，如果任务正在执行，若mayInterruptIfRunning设置为true，则返回true，若mayInterruptIfRunning设置为false，则返回false；如果任务还没有执行，则无论mayInterruptIfRunning为true还是false，肯定返回true。
-
isCancelled方法表示任务是否被取消成功，如果在任务正常完成前被取消成功，则返回 true。
-
isDone方法表示任务是否已经完成，若任务完成，则返回true；
-
get()方法用来获取执行结果，这个方法会产生阻塞，会一直等到任务执行完毕才返回；
-
get(long timeout, TimeUnit unit)用来获取执行结果，如果在指定时间内，还没获取到结果，就直接返回null。
实际上Future提供了三种功能:
- 判断任务是否完成；
- 中断任务
3 获取任务的执行结果
FutureTask
Future只是一个接口，无法直接创建对象，所以有了FutureTask，FutureTask实现了RunnableFuture<V>,而RunnableFuture又实现了Runnable和Future这两个接口，
public class FutureTask<V> implements RunnableFuture<V>
public interface RunnableFuture<V> extends Runnable, Future<V> {
/**
* Sets this Future to the result of its computation
* unless it has been cancelled.
*/
void run();
}
FutureTask还可以包装Runnable和Callable，由构造函数注入依赖。
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
}
Runnable注入会被Executors.callable()函数转换为Callable类型，也就是说FutureTask最终都是执行Callable类型的任务，该适配函数实现如下：
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;
}
}
FutureTask实现了Runnable，因此它既可以通过Thread包装来直接执行，也可以提交给ExecutorService来执行。
原理讲解
Runnable和Callable描述的都是抽象的计算任务，这些任务通常是有生命周期的，由于有些任务可能要执行很长的时间，因此通常希望可以取消这些任务。而Future用来表示一个任务的生命周期，并提供方法来判断任务是否已经完成或取消，以及获取任务的结果等。Future是接口，无法直接创建对象，所以才有了FutureTask，而FutureTask之所以能支持cancel操作，是因为FutureTask有两个很重要的属性state和runner。
private volatile int state; // 注意volatile关键字
/**
* 在构建FutureTask时设置，同时也表示内部成员callable已成功赋值，
* 一直到worker thread完成FutureTask中的run();
*/
private static final int NEW = 0;
/**
* woker thread在处理task时设定的中间状态，处于该状态时，
* 说明worker thread正准备设置result.
*/
private static final int COMPLETING = 1;
/**
* 当设置result结果完成后，FutureTask处于该状态，代表过程结果，
* 该状态为最终状态final state,(正确完成的最终状态)
*/
private static final int NORMAL = 2;
/**
* 同上，只不过task执行过程出现异常，此时结果设值为exception,
* 也是final state
*/
private static final int EXCEPTIONAL = 3;
/**
* final state, 表明task被cancel（task还没有执行就被cancel的状态）.
*/
private static final int CANCELLED = 4;
/**
* 中间状态，task运行过程中被interrupt时，设置的中间状态
*/
private static final int INTERRUPTING = 5;
/**
* final state, 中断完毕的最终状态，几种情况，下面具体分析
*/
private static final int INTERRUPTED = 6;
state有四种可能的状态转换：
NEW -> COMPLETING -> NORMAL
NEW -> COMPLETING -> EXCEPTIONAL
NEW -> CANCELLED
NEW -> INTERRUPTING -> INTERRUPTED
创建一个FutureTask首先调用构造方法, 这是state设置为初始态NEW, 当创建完一个Task通常会提交给Executors或者Thread来执行，最终会调用Task的run方法，
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
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);
}
}
FutureTask的run方法首先会判断任务的状态，如果任务状态不是NEW,说明任务状态已经改变，说明已经走了上面4种可能变化的一种，比如调用了cancel，此时状态为Interrupting。
如果状态是NEW,判断runner是否为null，如果为null，则把当前执行任务的线程赋值给runner，如果runner不为null，说明已经有线程在执行，直接返回。这里使用cas来赋值worker thread是保证多个线程同时提交同一个FutureTask时，确保该FutureTask的run方法只被调用一次。
!UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())
语义相当于
if (this.runner == null ){
this.runner = Thread.currentThread();
}
接着开始执行任务，如果要执行的任务不为空，并且state为New就执行，调用Callable的call方法，如果执行成功则set结果，如果出现异常则setException。最后把runner设为null。
set方法：如果现在的状态是NEW就把状态设置成cCOMPLETING,然后再设置成NORMAL,state的状态变化就是：NEW->COMPLETING->NORMAL.最后执行finishCompletion()方法。
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
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
}
finishCompletion()会解除所有阻塞的worker thread，调用done方法，将成员变量callable设为null。
接下来分析FutureTask非常重要的get方法：
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
}
首先判断FutureTask的状态是否为完成状态，如果是完成状态，说明已经执行过set或setException方法，返回report(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的状态是Normal，即正确执行了set方法，get方法直接返回处理的结果，如果是取消状态，即执行了setException,则抛出CancellationException异常。
如果get时，FutureTask的状态为未完成状态，则调用awaitDone方法进行阻塞。awaitDone():
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);
}
}
awaitDone方法可以看成是不断轮询查看FutureTask的状态。在get阻塞期间：
- 如果执行get的线程被中断，则移除FutureTask所有阻塞队列中的线程，并抛出中断异常
- 如果FutrueTask的状态转换为完成（正常完成或取消），返回完成状态。
- 如果FutureTask的状态为COMPLETING,说明正在set结果，此时让线程等一等。
- 如果FutureTask的状态为初始态NEW, 则将当前线程加入到FutureTask的阻塞队列中去；
- 如果get方法没有设置超时时间，则阻塞当前调用get线程，如果设置了超时时间，则判断是否达到超时时间，，如果到达，则移除FutureTask的所有阻塞队列中的线程，并返回此时FutureTask的状态，如果未到达时间，则在剩下的时间内继续阻塞当前线程。
再来看看Future的cancel方法
// 这个方法有一个参数 是否中断running
public boolean cancel(boolean mayInterruptIfRunning) {
/*
* 这个意思是 如果state不是new 那么就退出方法，这时的任务任务坑是已经完成了 或是被取消了 或是被中断了
* 如果state 是new 就设置state 为中断状态 或是取消状态
*
**/
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
try { // in case call to interrupt throws exception
//如果是可中断 那么就 调用系统中断方法 然后把状态设置成INTERRUPTED
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}
FutureTask的使用场景
FutureTask可用于异步获取执行结果或者取消执行任务的场景，通过传入Runnable或者Callable的任务给FutureTask，直接调用其run方法或者放入线程池执行，之后通过FutureTask的get方法异步获取执行结果，它非常适合用于耗时的计算，主线程可以在完成自己的任务后，再去获取结果，并且FutureTask还可以确保即使调动了多次run方法，它都只执行一次Runnable或者Callable任务，或者通过cancel取消FuturTask的执行。
public class FutureTest {
public static void main(String[] args) throws InterruptedException, ExecutionException {
System.out.println("====进入主线程执行任务");
//通过线程池管理多线程
ExecutorService threadPool = Executors.newCachedThreadPool();
//线程池提交一个异步任务
System.out.println("====提交异步任务");
Future<HashMap<String,String>> future = threadPool.submit(new Callable<HashMap<String,String>>() {
@Override
public HashMap<String,String> call() throws Exception {
System.out.println("异步任务开始执行....");
Thread.sleep(2000);
System.out.println("异步任务执行完毕，返回执行结果!!!!");
return new HashMap<String,String>(){
{this.put("futureKey", "成功获取future异步任务结果");}
};
}
});
System.out.println("====提交异步任务之后，立马返回到主线程继续往下执行");
Thread.sleep(1000);
System.out.println("====此时需要获取上面异步任务的执行结果");
boolean flag = true;
while(flag){
//异步任务完成并且未被取消，则获取返回的结果
if(future.isDone() && !future.isCancelled()){
HashMap<String,String> futureResult = future.get();
System.out.println("====异步任务返回的结果是："+futureResult.get("futureKey"));
flag = false;
}
}
//关闭线程池
if(!threadPool.isShutdown()){
threadPool.shutdown();
}
}
}
====进入主线程执行任务
====提交异步任务
====提交异步任务之后，立马返回到主线程继续往下执行
异步任务开始执行....
====此时需要获取上面异步任务的执行结果
异步任务执行完毕，返回执行结果!!!!
====异步任务返回的结果是：成功获取future异步任务结果
