一.使用

  1. public class Test {
  3.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  4.         Callable<Integer> myCallable = new MyCallable();
  5.         FutureTask<Integer> futureTask = new FutureTask<>(myCallable);
  6.         Thread thread = new Thread(futureTask);
  7.         thread.start();
  8.         Integer integer = futureTask.get();
  9.         System.out.println(integer);
  10.     }
  11. }
  14. class MyCallable implements Callable<Integer> {
  16.     @Override
  17.     public Integer call() throws InterruptedException {
  18.         System.out.println("call");
  19.         Thread.sleep(5000);
  20.         return 1;
  21.     }
  22. }

二.源码

Callable

  1. @FunctionalInterface
  2. public interface Callable<V> {
  3.     V call() throws Exception;
  4. }

FutureTask

  1. public class FutureTask<V> implements RunnableFuture<V>
  3. public interface RunnableFuture<V> extends Runnable, Future<V>
  5. public FutureTask(Callable<V> callable) {
  6.     this.callable = callable;
  7.     this.state = NEW;       
  8. }

FutureTask.run

  1. public void run() {
  2.     // private volatile Thread runner;  runner属性设置为当前线程
  3.     if (state != NEW ||
  4.         !UNSAFE.compareAndSwapObject(this, runnerOffset,
  5.                                      null, Thread.currentThread()))
  6.         return;
  7.     try {
  8.         Callable<V> c = callable;
  9.         if (c != null && state == NEW) {
  10.             V result;
  11.             boolean ran;
  12.             try {
  13.                 result = c.call();
  14.                 ran = true;
  15.             } catch (Throwable ex) {
  16.                 result = null;
  17.                 ran = false;
  18.                 setException(ex);
  19.             }
  20.             if (ran)
  21.                 set(result);
  22.         }
  23.     } finally {
  24.         runner = null;
  25.         int s = state;
  26.         if (s >= INTERRUPTING)
  27.             handlePossibleCancellationInterrupt(s);
  28.     }
  29. }
  31. protected void set(V v) {
  32.     if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
  33.         outcome = v;
  34.         UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
  35.         // 使用LockSupport.unpark唤醒所有等待的线程
  36.         finishCompletion();
  37.     }
  38. }
  40. protected void setException(Throwable t) {
  41.     if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
  42.         outcome = t;
  43.         UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL);
  44.         // 使用LockSupport.unpark唤醒所有等待的线程
  45.         finishCompletion();
  46.     }
  47. }

FutureTask.get

  1. public V get() throws InterruptedException, ExecutionException {
  2.     int s = state;
  3.     // 没有执行完毕或者正在设置执行结果时等待
  4.     if (s <= COMPLETING)
  5.         s = awaitDone(false, 0L);
  6.     return report(s);
  7. }
  9. private int awaitDone(boolean timed, long nanos)
  10.     throws InterruptedException {
  11.     WaitNode q = null;
  12.     boolean queued = false;
  13.     for (;;) {
  14.         if (q == null)
  15.             // 第一次循环
  16.             q = new WaitNode();
  17.         else if (!queued)
  18.             // 第二次循环入队，后进先出
  19.             queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
  20.                                                  q.next = waiters, q);
  21.         else
  22.             // 第三次循环
  23.             LockSupport.park(this);
  24.     }
  25. }
  27. static final class WaitNode {
  28.     volatile Thread thread;
  29.     volatile WaitNode next;
  30.     WaitNode() { thread = Thread.currentThread(); }
  31. }
  33. private V report(int s) throws ExecutionException {
  34.     Object x = outcome;
  35.     if (s == NORMAL)
  36.         return (V)x;
  37.     if (s >= CANCELLED)
  38.         throw new CancellationException();
  39.     throw new ExecutionException((Throwable)x);
  40. }