FutureTask源码完整解读
1 简介
上一篇博客“”对futuretask做过简要介绍与分析，这里再次对futuretask做一次深入的分析（基于jdk1.8）。
futuretask同时实现了future 、runnable接口，因此它可以交给执行器executor去执行这个任务，也可以由调用线程直接执行run方法。
根据futuretask.run方法的执行状态，可将其分为以下3种状态
①未启动： run方法还未被执行，futuretask处于未启动状态。
②已启动: run方法在执行过程中，futuretask处于已启动状态
③已完成：run方法正常完成返回或被取消或执行过程中因异常抛出而非正常结束，futuretask处于已完成状态。
当futuretask处于未启动或已启动状态时，执行futuretask.get()方法将导致调用线程阻塞；当futuretask处于已完成状态时，执行futuretask.get()方法将导致调用线程立即返回结果或抛出异常。
当futuretask处于未启动状态时，执行futuretask.cancel()方法将导致此任务永远不会被执行；当futuretask处于已启动状态时，执行futuretask.cancel(true)方法将以中断执行此任务线程的方式来试图停止任务；当futuretask处于已启动状态时，执行 futuretask.cancel(false)方法将不会对正在执行此任务的线程产生影响（让正在执行的任务运行完成）；当futuretask处于已完成状态时，执行futuretask.cancel方法将返回false (已完成的任务任务无法取消)。
2 用法示例
futuretask因其自身继承于runnable接口，因此它可以交给执行器executor去执行；另外它也代表异步任务结果，它还可以通过executorservice.submit返回一个futuretask。另外futuretask也可单独使用。为了更好的理解futuretask ，下面结合concurrenthashmap演示一个任务缓存。缓存中有多个任务，使用多线程去执行这些任务，一个任务最多被一个线程消费，若多个线程试图执行这一个任务，只允许一个线程来执行，其他线程必须等待它执行完成。
import java.util.concurrent.*;
public class futuretasktest {
private final concurrentmap<string, future<string>> taskcache = new concurrenthashmap<>();
public string executiontask(final string taskname)
throws executionexception, interruptedexception {
while (true) {
future<string> future = taskcache.get(taskname);// 从缓存中获取任务
if (future == null) {//不存在此任务，新构建一个任务放入缓存，并启动这个任务
callable<string> task = () ->{
system.out.println("执行的任务名是"+taskname);
return taskname;
} ; // 1.2创建任务
futuretask<string> futuretask = new futuretask<string>(task);
future = taskcache.putifabsent(taskname, futuretask);// 尝试将任务放入缓存中
if (future == null) {
future = futuretask;
futuretask.run();//执行任务
}
}
try { //若任务在缓存中了，可以直接等待任务的完成
return future.get();// 等待任务执行完成
} catch (cancellationexception e) {
taskcache.remove(taskname, future);
}
}
}
public static void main(string[] args) {
final futuretasktest tasktest = new futuretasktest();
for (int i = 0; i < 7; i++) {
int finali = i;
new thread(()->{
try {
tasktest.executiontask("taskname" + finali);
} catch (executionexception | interruptedexception e) {
e.printstacktrace();
}
}).start();
new thread(()->{
try {
tasktest.executiontask("taskname" + finali);
tasktest.executiontask("taskname" + finali);
} catch (executionexception | interruptedexception e) {
e.printstacktrace();
}
}).start();
}
}
}
打印输出
3 实现原理
1) 成员变量
它有一个成员变量state表示状态
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;//任务被中断(中断的最终状态)
state可能有这几种状态转换
/** new -> completing -> normal 正常结束任务时的状态转换流程
* new -> completing -> exceptional 任务执行过程中抛出了异常时的状态转换流程
* new -> cancelled 任务被取消时的状态转换流程
* new -> interrupting -> interrupted 任务执行过程中出现中断时的状态转换流程
*/
其他成员变量
private callable<v> callable; private object outcome; // non-volatile, protected by state reads/writes private volatile thread runner; private volatile waitnode waiters;
成员变量callable表示要执行的任务，
成员变量outcome表示任务的结果或任务非正常结束的异常
成员变量runner表示执行此任务的线程
成员变量waiter表示等待任务执行结果的等待栈(数据结构是单向链表) 。waitnode是一个简单的静态内部，一个成员变量thread表示等待结果的线程，另一个成员变量next表示下一个等待节点（线程）。
static final class waitnode {
volatile thread thread;
volatile waitnode next;
waitnode() { thread = thread.currentthread(); }
}
2) 构造方法
futuretask的构造方法会初始化callable和state ，它有两个构造方法， 分别接受callable和runnable类型的待执行任务。但对于runnable类型参数，它会调用executors.callable将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
}
executors.callable方法也很简单，它就返回了一个callable的实现类runnableadapter类型的对象。
public static <t> callable<t> callable(runnable task, t result) {
if (task == null)
throw new nullpointerexception();
return new runnableadapter<t>(task, 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;
}
}
3) 主要api
(1) run与runandreset
run方法是funture最重要的方法，futuretask的一切都是从run方法开始的，它是执行callable任务的方法。
public void run() {
if (state != new ||
!unsafe.compareandswapobject(this, runneroffset,
null, thread.currentthread()))
//将当前线程设置为执行任务的线程，cas失败就直接返回
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已是最终状态，不再变化，将runer设为null,防止run方法被并发调用
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state; //清空运行线程runner后再重新获取state,防止遗漏掉对中断的处理
if (s >= interrupting)
handlepossiblecancellationinterrupt(s);
}
}
其主要逻辑是：
①检查状态，设置运行任务的线程
②调用callable的call方法去执行任务，并捕获运行中可能出现的异常
③如果任务正常完成，调用set设置任务的结果，将state设为normal, 将结果保存到outcome ,唤醒所有等待结果的线程
④若执行任务过程中发生了异常，调用setexception设置异常,将state设为exceptional ，将此异常也保存到outcome ，唤醒所有等待结果的线程
⑤最后将运行线程runner清空，若状态可能是任务被取消的中断还要处理此中断。
set 、setexception方法分别用来设置结果、设置异常，但这仅是它们的主要逻辑，它们还会进行其他的处理。
它们会将结果或异常设置到成员变量outcome上，还会更新状态state，最后调用finishcompletion从等待栈表中移除并唤醒所有（节点）线程(任务已完成，无需要等待，可以直接获取结果，等待栈已没有存在的意义了)。
protected void setexception(throwable t) {
if (unsafe.compareandswapint(this, stateoffset, new, completing)) {
outcome = t;
unsafe.putorderedint(this, stateoffset, exceptional); // final state
finishcompletion();
}
}
protected void set(v v) {
if (unsafe.compareandswapint(this, stateoffset, new, completing)) {
outcome = v;
unsafe.putorderedint(this, stateoffset, normal); // final state
finishcompletion();
}
}
run方法中有对中断的处理，我们来看看handlepossiblecancellationinterrupt方法怎么处理中断的.
这里就是简单地使当前线程让出时间片，让其他线程先执行任务，即线程礼让。
private void handlepossiblecancellationinterrupt(int s) {
if (s == interrupting)
while (state == interrupting)
thread.yield(); // wait out pending interrupt
}
runandreset方法是futuretask类自己添加的protected级别的方法（供子类调用）, 这个方法主要用来执行可多次执行且不需要结果的任务，只有在任务运行和重设成功时才返回true 。定时任务执行器scheduledthreadpoolexecutor的静态内部scheduledfuturetask的run方法调用了这个api.
和run方法相比，runandset方法与之逻辑大致相同，只是runandset没用调用set方法设置结果(本身不需要结果，也是出于防止state被修改的目的)
protected boolean runandreset() {
if (state != new ||
!unsafe.compareandswapobject(this, runneroffset,
null, thread.currentthread()))
//任务已启动或cas设置运行任务的的线程失败，直接返回false
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 没有调用set(v)方法，不设置结束
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; //任务成功运行且state还是new时返回true,反之返回false
}
(2) get方法
get方法用于获取任务的最终结果，它有两个版本，其中一个是超时版本。两个版本的最主要的区别在于，非超时版本可以不限时长地等待结果返回 ,另外非超时版本不会抛出timeoutexception超时异常。get方法超时版本的基本逻辑：若任务未完成就等待任务完成，最后调用report报告结果，report会根据状态返回结果或抛出异常。
public v get() throws interruptedexception, executionexception {
int s = state;
if (s <= completing)
s = awaitdone(false, 0l);//awaitdone第一个参数为false，表示可以无限时长等待
return report(s);
}
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();//到了限定时间，任务仍未完成，抛出超时异常timeoutexception
return report(s);//报告结果
}
get方法的核心实现在于调用awaitdone方法，awaitdone用于等待任务的结果，若任务未完成awaitdone会阻塞当前线程。
awaitdone方法的基本逻辑：①若执行任务时出现了中断，则抛出interruptedexception异常;②若此时任务已完成，就返回最新的state，③若任务即将完成就使当前线程让出cpu时间片，让其他线程先执行；④若任务还在执行中，就将当前线程加入到等待栈中，然后让当前线程休眠直到超出限定时间或等待任务完成时run方法调用finishcompletion唤醒线程(run方法中的set或setexception调用finishcompletion，而finishcompletion又会调用locksupport.unpark).
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)
//如果之前入栈失败，再次尝试入栈（cas更新），将当前节点设为等待栈的栈顶
queued = unsafe.compareandswapobject(this, waitersoffset,
q.next = waiters, q);
else if (timed) { //如果设置了超时时间
nanos = deadline - system.nanotime();
if (nanos <= 0l) {
//如果任务执行时长已经超出了给定的时间，从等待栈中移除当前节点（线程）
removewaiter(q);
return state;
}
//让当前线程休眠等待给定的时间(或等到run方法中的set或setexception调用finishcompletion来唤醒)
locksupport.parknanos(this, nanos);
}
else//未设置超时时间
//让当前线程无限时长休眠等待，直到任务完成时run方法中的set或setexception调用finishcompletion来唤醒此线程
locksupport.park(this);
}
}
上面的awaitdone方法中调用removewaiter来移除等待栈表的中断和超时节点。
其内部实现不容易理解，但主要逻辑还是很清楚的：从头到尾遍历链表，将链表中的中断/超时节点移除出链表，若有线程竞争就重头开始再次遍历链表检查并移除无效节点。
private void removewaiter(waitnode node) {
if (node != null) {
node.thread = null; //先将节点对应的线程清空,下面的"q.thread != null"就能判断节点是否超时或中断节点。
retry:
for (;;) { // restart on removewaiter race
//q表示当前遍历到的节点，pred表示q的前驱节点，s表示q的后继节点
for (waitnode pred = null, q = waiters, s; q != null; q = s) {//遍历完链表才能退出内循环
s = q.next;
//q.thread!=null 表示这不是超时或中断的节点，它是效节点，不能被从栈表中移除
//（removewaiter的开头将超时或中断的节点在thread赋空,可见node.thread=null代码）
if (q.thread != null)
pred = q; //得到下次循环时q的前驱节点
else if (pred != null) { //q.thread== null 且pred！=null,需要将无效节点q从栈表中移除
//将q的前驱、后继节点直接链接在一起，q本身被移除出栈表了
pred.next = s;
//这里是从前向后遍历链表的，无竞争情况下，不可能没检查到当前节点的前面还有无效节点，
//那么一定有其他线程修改了当前节点q的前驱，些时有线程竞争，需要从链表的头部重新遍历检查
if (pred.thread == null) // check for race
continue retry;
}
// pred==null且q.thread=null
//q的前驱节点为空，表明q是链表的头节点
//q.thread==null，表明q是无效节点
//无效节点不能作为链表的头节点，所以要更新头节点，将q的后继节点s作为链表新的头节点
else if (!unsafe.compareandswapobject(this, waitersoffset, //cas更新头节点
q, s))
//cas更新失败，重试
continue retry;
}
break;
}
}
}
get方法需要调用report方法来报告结果，而report方法的基本逻辑也简单：若是任务正常结束就返回这个任务的结果，若是任务被取消，就抛出cancellationexception异常，若是在执行任务过程中发生了异常就将其统一封装成executionexception并抛出。
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);
}
(3) cancel方法
cancel方法用于取消任务，我们可以看看cancel(boolean)方法如何实现的
public boolean cancel(boolean mayinterruptifrunning) {
if (!(state == new &&
unsafe.compareandswapint(this, stateoffset, new,
mayinterruptifrunning ? interrupting : cancelled)))
//①不是new状态，表示任务至少是completing(即将结束)状态，返回false
//②cas更新state为interrupting或cancelled失败，返回false
//只有state状态更新成功，才能取消任务（防止被并发调用）
return false;
try { // in case call to interrupt throws exception
if (mayinterruptifrunning) {//允许中断就设置中断标志
try {
thread t = runner;
if (t != null)
t.interrupt();//设置中断标志
} finally { // final state 设置中断的最终状态
//interrupting -> interrupted ,将state由“正在中断”更新为”已经中断“
unsafe.putorderedint(this, stateoffset, interrupted);
}
}
} finally {
//从等待栈上唤醒并移除所有的线程(节点)
finishcompletion();
}
return true;
}
其基本逻辑：
①任务已结束或被取消，返回false
②若mayinterruptifrunning为true,调用interrupt设置中断标志，将state设置为interrupted,若mayinterruptifrunning为false，将state设为cancelled.
③调用finishcompletion唤醒并移除等待栈中的所有线程
finishcompletion()主要是处理任务结束后的扫尾工作，其主要逻辑是：将等待栈waiters赋空，唤醒并移除等待栈上的所有节点(线程)，最后再将任务callable赋空。
private void finishcompletion() {
// assert state > completing;
for (waitnode q; (q = waiters) != null;) {
//任务取消后，等待栈表没有存在的意义了，将等待栈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 将节点next属性清空，方便垃圾回收
q = next;//向后移动一个节点
}
break;
}
}
done();//空方法，留给子类重写
callable = null; //赋空，减少痕迹 // to reduce footprint
}
(4) 其他辅助方法
iscancelled方法返回任务是否被取消的布尔值
isdone方法返回任务是否完成的布尔值(非正常结束也行)
iscancelled 、isdone都是直接根据state确定任务的状态。
public boolean iscancelled() {
return state >= cancelled;
}
public boolean isdone() {
return state != new;
}
上一篇： 很欠揍的爆笑熊孩子又来了
下一篇： 模糊查询