在文章FutureTask源码分析中简单说明了FutureTask中使用Treiber堆栈来保存等待结果的线程，本文将详细分析其原理。

Treiber堆使用CAS操作来实现节点的入栈和出栈，由于CAS操作只是保证操作的原子性，多线程并发时，其并不能保证可见性，因此必须依赖volatile来保证写入的可见性。这样就可以不必使用加锁来实现对共享数据结构的访问。下面先看一个实现Treiber堆的例子：

public class TreiberStack<E> {AtomicReference<Node<E>> head = new AtomicReference<Node<E>>();public void push(E item) {//总是在堆顶加入新节点Node<E> node= new Node<E>(item);Node<E> old;do {old = head.get();node.next = old;} while (!head.compareAndSet(old, node));}public E pop() {//总是从堆顶移除Node<E> old;Node<E> node;do {old = head.get();if (old == null) return null;node = old.next;} while (!head.compareAndSet(old,node));return old.item;}private static class Node<E>{public final  E  item;public Node<E>   next;public Node(E item){this.item=item;}}

FutureTask中WaitNode 作为节点，并将当前线程保存在其中，而且将栈顶元素保存在waiters中。

入栈

首先看入栈操作，当调用FutureTask的get方法时，若任务未完成则会将当前等待结果的线程加入到等待队列，并挂起当前线程。

    private int awaitDone(boolean timed, long nanos)throws InterruptedException {final long deadline = timed ? System.nanoTime() + nanos : 0L;WaitNode q = null;boolean queued = false;for (;;) {//当前线程是否已中断，该方法会清除线程状态，也就是说第一次调用返回true,//并且没有再次被中断时，第二次调用将返回falseif (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) //表示任务马上完成，不必进入等待队列Thread.yield();else if (q == null)//此时s只可能为NEWq = 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);//等待指定时间间隔后挂起}elseLockSupport.park(this);//挂起线程}}

由源码可知，当前线程未被中断时，使用CAS操作加入当前等待节点q，通过将q设为新的栈顶元素，即waiters，同时修改q.next指针指向上一次的waiters。这里使用自旋操作来保证操作一定成功。

看下面例子：

public class FutureTaskStackTest {private static FutureTask<Integer> ft=new FutureTask<Integer>(new ComputeTask(0,"task"));public static void main(String[] args) throws InterruptedException,ExecutionException{ExecutorService   executor=Executors.newSingleThreadExecutor();executor.submit(ft);executor.shutdown();for(int i=0;i<5;i++){System.out.println("for loop "+i);System.out.println(Thread.currentThread()+":"+ ft.get());}}private static class ComputeTask implements Callable<Integer>{private Integer result ;  private String  taskName ; public ComputeTask(Integer init, String taskName){  result = init;  this.taskName = taskName;  }  @Overridepublic Integer call() throws Exception {for (int i = 0; i < 100; i++) {  result =+ i;  } Thread.sleep(5000);System.out.println(taskName+" finish!");  return result;}}
}

输出：

for loop 0
task finish!
Thread[main,5,main]:99
for loop 1
Thread[main,5,main]:99
for loop 2
Thread[main,5,main]:99
for loop 3
Thread[main,5,main]:99
for loop 4
Thread[main,5,main]:99

ComputeTask将sleep 5000ms才会完成任务，主线程中循环调用5次future.get()。那么等待队列中会加入5个节点吗？实际不是，只会加入一个，当加入一个时，当前main线程会被挂起，即输出“for loop 0”之后被挂起，直到任务完成被唤醒。这就说明同一个线程中调用多次future.get()和调用一次在FutureTask中都将只会加入一个节点，当线程被唤醒时，future.get()将不会被阻塞。

那么使用如下例子：

public class FutureTaskStackTset {private static FutureTask<Integer> ft=new FutureTask<Integer>(new ComputeTask(0,"task"));public static void main(String[] args) throws InterruptedException,ExecutionException{ExecutorService   executor=Executors.newSingleThreadExecutor();executor.submit(ft);executor.shutdown();MyThread thread1=new MyThread();thread1.setName("thread1");MyThread thread2=new MyThread();thread2.setName("thread2");thread1.start();thread2.start();System.out.println(Thread.currentThread().getName()+" : "+ ft.get());}private static class ComputeTask implements Callable<Integer>{private Integer result ;  private String  taskName ; public ComputeTask(Integer init, String taskName){  result = init;  this.taskName = taskName;  }  @Overridepublic Integer call() throws Exception {for (int i = 0; i < 100; i++) {  result =+ i;  } Thread.sleep(5000);System.out.println(taskName+" finish!");  return result;}}private static class MyThread extends Thread{public void run() {try {System.out.println(this.getName()+" : "+ ft.get());} catch (InterruptedException | ExecutionException e) {e.printStackTrace();}}}
}

输出：

task finish!
main : 99
thread1 : 99
thread2 : 99

在这个例子中，在不同的线程中调用future.get()，因此Treiber堆中会加入3个等待节点。这里main线程中的future.get()必须在启动其他线程之后调用，否则，main线程被阻塞，那么子线程就不会被启动，自然也不会加入等待队列。

移除

awaitDone实现可知，若当前线程被中断，FutureTask将会清理等待队列，移除已经被中断线程的节点。

private void removeWaiter(WaitNode node) {if (node != null) {node.thread = null;retry:for (;;) {for (WaitNode pred = null, q = waiters, s; q != null; q = s) {s = q.next;if (q.thread != null)pred = q;else if (pred != null) {//移除thread为null的节点pred.next = s;if (pred.thread == null) // check for racecontinue retry;}//q和pred的thread均为null,将s设为新的栈顶元素，即waiterselse if (!UNSAFE.compareAndSwapObject(this, waitersOffset,q, s))continue retry;//失败则重新进入循环}break;}}}

removeWaiter首先将要移除节点的thread变量置为null，这一步很关键，因为后续移除等待节点就是根据thread是否为null来实现。下面分别分析几种可能：

1 waiters即为要移除的元素

从代码分析，q指向waiters的节点，s=q->next，则s指向T1，此时q.thread为null，pred也为null，进入else if (!UNSAFE.compareAndSwapObject(this, waitersOffset, q, s))，将s设为waiters，成功移除为null的节点。

2 thread为null元素位于堆栈中间

当找到q.thread为null的元素时，pred将指向T1,而S则指向T2，那么程序将会进入else if (pred != null)的代码块，将pred的next指针指向S，则成功将q移除。

完全有可能出现thread为null的元素位于中间甚至末尾的情况，当多线程调用get（）方法时，CAS保证同时只能有一个线程将节点加入等待队列。失败的线程将继续进行自旋操作，直到成功。同时，上文已提过，相同线程多次调用get也只会加入一个节点到等待队列，因此removeWaiter一次调用实际只会移除一个节点。

removeWaiter操作的作用在于移除无效节点，避免造成垃圾累积，当堆栈中节点较多，removeWaiter操作会很慢。通常情况下，不会有太多线程同时等待一个任务的结果。

出栈

当任务执行完成后，将在 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 gcq = next;}break;}}done();callable = null;        // to reduce footprint}

finishCompletion()实现即可知，首先使用CAS操作将waiters置为null，然后从栈顶到栈底唤醒所有等待节点，并将节点的thread和next置空，这有助于GC回收内存。

欢迎指出本文有误的地方，转载请注明原文出处https://my.oschina.net/7001/blog/875714