Executor框架

public static ExecutorService newFixedThreadPool(int nThreads) {
  return new ThreadPoolExecutor(nThreads, nThreads,
                  0L, TimeUnit.MILLISECONDS,
                  new LinkedBlockingQueue<Runnable>());
}

public static ExecutorService newSingleThreadExecutor() {
  return new FinalizableDelegatedExecutorService
    (new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>()));
}

public static ExecutorService newCachedThreadPool() {
  return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                  60L, TimeUnit.SECONDS,
                  new SynchronousQueue<Runnable>());
}

public class CallableDemo implements Callable<String> {
  @Override
  public String call() throws Exception {
    return "Hello";
  }
  public static void main(String[] args) throws InterruptedException, ExecutionException {
    CallableDemo callableDemo = new CallableDemo();
    FutureTask futureTask = new FutureTask(callableDemo);
    new Thread(futureTask).start();
    System.out.println(futureTask.get());
  }
}

@FunctionalInterface
public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}

public interface Future<V> {


    boolean cancel(boolean mayInterruptIfRunning);


    boolean isCancelled();


    boolean isDone();
    
    V get() throws InterruptedException, ExecutionException;
    
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

public class FutureTask<V> implements RunnableFuture<V> {


    private volatile int state;
    private static final int NEW          = 0; // 新建状态，表示这个FutureTask还没有开始运行
    private static final int COMPLETING   = 1; // 表示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; // 任务结束，也是由于取消了任务，但是发起了中断运行任务线程的中断请求
    private static final int INTERRUPTED  = 6; // 任务结束，也是由于取消了任务，完成了中断运行任务线程的中断请求
  .....// 省略部分代码
}

public void run() {
  // 如果状态不是New，或者设置runner的值失败，说明有其它线程已经调用了run方法并成功设置了runner的值，直接return
  // 确保了只有一个线程可以运行try代码块中的代码
  if (state != NEW ||
    !UNSAFE.compareAndSwapObject(this, runnerOffset,
                   null, Thread.currentThread()))
    return;
  try {
    Callable<V> c = callable;
    if (c != null && state == NEW) { // 只有c不为null且状态为New的时候，才可以执行
      V result;
      boolean ran;
      try {
        result = c.call(); // 调用Callable的call方法，并得到返回值
        ran = true; // 设置运行成功标识
      } catch (Throwable ex) {
        result = null;
        ran = false; // 如果发生异常，设置运行成功标识位false
        setException(ex); // 设置异常结果
      }
      if (ran)
        set(result);  // 如果成功，把结果赋值给outCome变量
    }
  } finally {
    runner = null;
    int s = state;
    if (s >= INTERRUPTING)
      handlePossibleCancellationInterrupt(s);
  }
}


protected void set(V v) {
  if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
    outcome = v;
    UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
    finishCompletion();
  }
}

public V get() throws InterruptedException, ExecutionException {
  int s = state;
  if (s <= COMPLETING)
    s = awaitDone(false, 0L);
  return report(s);
}

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()) {
      // 如果当前线程中断标志位为true，则从列表中移除节点，并且抛出exception
      removeWaiter(q);
      throw new InterruptedException();
    }


    int s = state;
    if (s > COMPLETING) {  // 状态大于COMPLETING，标识FutureTask已经结束
      if (q != null)
        q.thread = null; // 将节点线程设置成null， 因为线程没有阻塞等待
      return s;
    }
    else if (s == COMPLETING) //  任务已经完成，但是还有一些后续操作没有完成，当前线程让出执行权
      Thread.yield();
    else if (q == null) // 标识state为New，需要将当前线程阻塞等待， 创建waitNode节点
      q = new WaitNode();
    else if (!queued)
      // 通过CAS将q添加到等到队列，如果失败则queued为false，下次循环会继续添加，知道成功
      queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                         q.next = waiters, q);
    else if (timed) { // timed为true表示要设置超时
      nanos = deadline - System.nanoTime();
      if (nanos <= 0L) {
        removeWaiter(q);
        return state;
      }
      LockSupport.parkNanos(this, nanos); // 让当前线程等待nanos时间
    }
    else
      LockSupport.park(this);
  }
}

private V report(int s) throws ExecutionException {
  Object x = outcome;
  if (s == NORMAL)
    return (V)x;
  if (s >= CANCELLED) // 大于等于CANCELLED，表示手动去掉了FutureTask
    throw new CancellationException();
  throw new ExecutionException((Throwable)x);
}