1、介绍
Java创建线程一般用Thread，Runable。如果需要返回结果，就用Callable。Callable运行需要配合Future。
Future是一个接口，一般用FutureTask实现类去接收Callable任务返回的结果。FutureTask是同步非阻塞执行的任务，他不会主动通知你结果。
1.1 使用方式
如下图，创建一个FutureTask丢到线程里去执行，通过get获取结果，如果没有执行完成get方法会阻塞线程，一直等到执行完成。
public static void main(String[] args) throws ExecutionException, InterruptedException {
FutureTask<String> futureTask = new FutureTask<>(()->{
Thread.sleep(2000);
return "异步任务";
});
Thread t = new Thread(futureTask);
t.start();
System.out.println(futureTask.get());
}
2、FutureTask源码解析（查看注释）
2.1 FutureTask属性
/*
* 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; // Callable的结果(正常或者异常结果)正在封装给当前的FutureTask
private static final int NORMAL = 2; // 任务正常结束
private static final int EXCEPTIONAL = 3; // 执行任务时，发生了异常
private static final int CANCELLED = 4; // 任务被取消了
private static final int INTERRUPTING = 5; // 线程的中断状态被设置为了true(还在运行)
private static final int INTERRUPTED = 6; // 线程被中断了
/** The underlying callable; nulled out after running */
//当前执行的任务
private Callable<V> callable;
/** The result to return or exception to throw from get() */
// 存放返回的结果或者异常信息,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 */
//单向链表，存放通过get方法挂起等待的线程
private volatile WaitNode waiters;
// 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);
}
}
2.2 FutureTask的run方法
FutureTask的继承链。FutureTask–>RunnableFuture–>Runnable, Future。FutureTask重写了Runable的run()方法。
通过run方法执行的call方法
// run方法执行流程，最终会执行Callable的call方法
public void run() {
//保证任务状态是NEW才可以运行，通过CAS操作将当前线程设置为runner
if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
return;
try {
//要执行的任务 c
Callable<V> c = callable;
//判断任务不为空，并且状态是NEW
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设置为空
runner = null;
//拿到状态
int s = state;
//中断后续处理
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
protected void set(V v) {
//CAS操作，将state从NEW改为COMPLETING状态
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//将结果复制给outcome
outcome = v;
//状态修改为NORMAL，正常结束
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
2.3 get()获取结果
public V get() throws InterruptedException, ExecutionException {
int s = state;
//没有完成
if (s <= COMPLETING)
//尝试挂起线程，等待拿结果
s = awaitDone(false, 0L);
return report(s);
}
2.4 awaitDone等待完成
该方法判断当前线程执行状态：
- 已经结束，直接返回状态，在方法调用处会返回结果
- 正在包装结果中，让出时间片等待一会
- 还在NEW状态，使用头插法将当前线程进入等待队列（这里是读取结果的线程等待对接，如果多个线程需要读取结果，他们将进入队列等待，队列是单链表结构）
- 根据时间判断线程是否需要挂起
private int awaitDone(boolean timed, long nanos) throws InterruptedException {
//计算deadline，如果是get()，就是0 ，如果是get(time, unit)就追加系统当前时间
final long deadline = timed ? System.nanoTime() + nanos : 0L;
//构建WaitNode
WaitNode q = null;
//是否进了队列
boolean queued = false;
for (;;) {
//这个get的线程是否中断
if (Thread.interrupted()) {
//将当前节点从waiters移除
removeWaiter(q);
//并抛出异常
throw new InterruptedException();
}
int s = state;
//判断任务是否已经执行结束
if (s > COMPLETING) {
//如果设置过WaitNode，直接移除WaitNode的线程(这里是一个死循环，节点在下面创建，但是循环之后可能又回到这里)
if (q != null)
q.thread = null;
return s;
}
//COMPLETING状态，任务结果返回中...
else if (s == COMPLETING) // cannot time out yet
//COMPLETING状态持续时间非常短，只要做一下让步即可
Thread.yield();
//<COMPLETING的只有NEW状态，现在线程状态是NEW，(call方法还没有执行完成，准备挂起线程)
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
//如果无限挂起时间，并且线程还没有结束，继续挂起(get()挂起线程的方式)
LockSupport.park(this);
}
}
2.5 任务状态变成NORMAL，做后续处理
任务完成后，需要将等待队列的线程一个个唤醒，唤醒的线程将拿到任务执行的结果，并通过GC回收节点
/**
* Removes and signals all waiting threads, invokes done(), and
* nulls out callable.
*/
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
//拿到第一个节点后，直接CAS将其设置为null
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
//基于q拿到线程信息
Thread t = q.thread;
if (t != null) {
//将q的线程设置为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
}
2.6 返回结果
//任务结束
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、FutureTask存在的问题
- 问题1： FutureTask获取执行结果前，主线程需要通过get()方法一直阻塞等待子线程执行完成call方法，才可以拿到返回结果
- 问题2：如果不通过get挂起线程，通过while循环，不停的判断任务的状态是否结束，结束后，再拿结果。如果任务长时间没有执行完毕，CPU会一直调度查看任务状态的方法，浪费CPU资源。
FutureTask是一个同步非阻塞处理任务的方式。
需要一个异步非阻塞处理任务的方式。
CompletableFuture在一定程度上提高了各种异步非阻塞的方案，并且响应式变成，代码编写效果上，效率更高。CompletableFuture也实现了Future接口，可以不使用FutureTask，直接只用CompletableFuture。
