【线程池】史上最全的ThreadPoolExecutor源码详解

public class ThreadPoolExecutor extends AbstractExecutorService {
    // 线程池的状态和线程数量都是通过一个原子类型AtomicInteger的变量ctl来表示的，创建线程池的时候就会将ctl初始化为RUNNING状态和0个工作线程
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    // 表示低29位的值，用于表示线程池中工作线程的数量
    private static final int COUNT_BITS = Integer.SIZE - 3; // =29
    // 用于表示线程池中工作线程的数量的最大值，等于2^29-1
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1; // =000 11111...
    // 线程池的五种状态，高3位表示    下面的状态值依次增高，是根据他们状态流程的顺序依次增高的
    private static final int RUNNING    = -1 << COUNT_BITS; // 111 00000...
    private static final int SHUTDOWN   =  0 << COUNT_BITS; // 000 00000...
    private static final int STOP       =  1 << COUNT_BITS; // 001 00000...
    private static final int TIDYING    =  2 << COUNT_BITS; // 010 00000...
    private static final int TERMINATED =  3 << COUNT_BITS; // 011 00000...
 
    // 传入ctl，计算返回线程池的状态
    // 通过与运算，将高3位的值保留，低29位的值置为0
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    // 传入ctl，计算返回线程池中工作线程的数量
    // 通过与运算，将高3位的值置为0，低29位的值保留
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    // 计算ctl的值，等于运行状态“加上”线程数量
    // 通过或运算，将高3位和低29位的值都保留，合并为ctl并返回
    private static int ctlOf(int rs, int wc) { return rs | wc; }
 
 
    // runStateLessThan方法用于判断线程池的状态是否小于某个状态
    private static boolean runStateLessThan(int c, int s) {
        return c < s;
    }
    // runStateAtLeast方法用于判断线程池的状态是否大于等于某个状态
    private static boolean runStateAtLeast(int c, int s) {
        return c >= s;
    }
    // isRunning方法用于判断线程池的状态是否是RUNNING状态
    private static boolean isRunning(int c) {
        return c < SHUTDOWN;
    }
    // 该方法用于增加工作线程的数量，如果增加成功则返回true，否则返回false
    private boolean compareAndIncrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect + 1);
    }
    // 该方法用于减少工作线程的数量，如果减少成功则返回true，否则返回false
    private boolean compareAndDecrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect - 1);
    }
    // 该方法用于设置线程池的状态，如果设置成功则返回true，否则返回false
    private void decrementWorkerCount() {
        do {} while (! compareAndDecrementWorkerCount(ctl.get()));
    }
    // 用于保存工作线程的集合
    private final HashSet<Worker> workers = new HashSet<Worker>();
    // 线程池的状态锁，对线程池状态（比如线程池大小、runState等）的改变都要使用这个锁
    private final ReentrantLock mainLock = new ReentrantLock();
    // 线程池的条件锁，用mainLock创建的，在线程池的状态发生改变时，会使用这个条件锁来通知所有等待的线程
    private final Condition termination = mainLock.newCondition();
 
 
    // 核心参数：任务阻塞队列
    private final BlockingQueue<Runnable> workQueue;
    // 核心参数：线程池的线程工厂，用于创建线程
    private volatile ThreadFactory threadFactory;
    // 核心参数：拒绝策略，用于在线程池已经关闭或者线程数已经达到最大值的情况下，对新任务的处理策略
    private volatile RejectedExecutionHandler handler;
    // 核心参数：当线程数大于核心线程数时，多余的空闲线程存活的最长时间
    private volatile long keepAliveTime;
    // 是否允许核心线程空闲退出，默认值为false
    private volatile boolean allowCoreThreadTimeOut;
    // 核心参数：线程池的核心线程数
    private volatile int corePoolSize;
    // 核心参数：线程池的最大线程数
    private volatile int maximumPoolSize;
 
    // 这是一个动态的变量。
    // largestPoolSize表示线程池中曾经出现过的最大线程数，即在线程池的整个生命周期中，曾经有多少个线程同时处于活动状态。所以说largestPoolSize是小于等于maximumPoolSize的。
    // 它的值可以通过ThreadPoolExecutor的getLargestPoolSize()方法获取。
    private int largestPoolSize;
    // 表示已经完成的任务数量
    private long completedTaskCount;
    // 默认的拒绝策略，默认的拒绝策略就是直接抛出异常。如果构造方法中没有转入指定的拒绝策略，就会将defaultHandler赋值给handler
    private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();
    ...
}

// 线程池的状态和线程数量都是通过一个原子类型AtomicInteger的变量ctl来表示的，创建线程池的时候就会将ctl初始化为RUNNING状态和0个工作线程
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 表示低29位的值，用于表示线程池中工作线程的数量
private static final int COUNT_BITS = Integer.SIZE - 3; // =29
// 用于表示线程池中工作线程的数量的最大值，等于2^29-1
private static final int CAPACITY   = (1 << COUNT_BITS) - 1; // =000 11111...
// 线程池的五种状态，高3位表示    下面的状态值依次增高，是根据他们状态流程的顺序依次增高的
private static final int RUNNING    = -1 << COUNT_BITS; // 111 00000...
private static final int SHUTDOWN   =  0 << COUNT_BITS; // 000 00000...
private static final int STOP       =  1 << COUNT_BITS; // 001 00000...
private static final int TIDYING    =  2 << COUNT_BITS; // 010 00000...
private static final int TERMINATED =  3 << COUNT_BITS; // 011 00000...
// 传入ctl，计算返回线程池的状态
// 通过与运算，将高3位的值保留，低29位的值置为0
private static int runStateOf(int c)     { return c & ~CAPACITY; }
// 传入ctl，计算返回线程池中工作线程的数量
// 通过与运算，将高3位的值置为0，低29位的值保留
private static int workerCountOf(int c)  { return c & CAPACITY; }
// 计算ctl的值，等于运行状态“加上”线程数量
// 通过或运算，将高3位和低29位的值都保留，合并为ctl并返回
private static int ctlOf(int rs, int wc) { return rs | wc; }
 
 
// runStateLessThan方法用于判断线程池的状态是否小于某个状态
private static boolean runStateLessThan(int c, int s) {
    return c < s;
}
// runStateAtLeast方法用于判断线程池的状态是否大于等于某个状态
private static boolean runStateAtLeast(int c, int s) {
    return c >= s;
}
// isRunning方法用于判断线程池的状态是否是RUNNING状态
private static boolean isRunning(int c) {
    return c < SHUTDOWN;
}

// 初始状态为RUNNING
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

public void shutdown() {
    // 获取线程池锁
    final ReentrantLock mainLock = this.mainLock;
    // 加锁，修改线程池状态
    mainLock.lock();
    try {
		  // 检查是否有权限关闭线程池
        checkShutdownAccess();
        // 修改状态为SHUTDOWN，自旋操作，只有状态修改成功才会返回
        advanceRunState(SHUTDOWN);
        // 标记空闲线程为中断状态
        interruptIdleWorkers();
        onShutdown();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}
 
private void advanceRunState(int targetState) {
    // 自旋修改线程池状态
    for (;;) {
        // 获取ctl
        int c = ctl.get();
        // 如果状态大于SHUTDOWN（因为只有RUNNING状态才能转化为SHUTDOWN状态，而只有RUNNING状态是小于SHUTDOWN状态的），
        // 或者修改为SHUTDOWN成功了，才会break跳出自旋
        if (runStateAtLeast(c, targetState) ||
            ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
            break;
    }
}

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    // 获取线程池的锁
    final ReentrantLock mainLock = this.mainLock;
    // 加锁，进行状态修改操作
    mainLock.lock();
    try {
        // 检查是否有权限关闭线程池
        checkShutdownAccess();
        // 修改为STOP状态
        advanceRunState(STOP);
        // 标记所有线程为中断状态
        interruptWorkers();
        // 将队列中的任务全部移除，并返回
        tasks = drainQueue();
    } finally {
        // 解锁
        mainLock.unlock();
    }
    // 尝试终止线程池
    tryTerminate();
    // 返回队列中的任务
    return tasks;
}

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // 如果状态为STOP了，这里会直接退出循环，且减少工作线程数量
        // 退出循环了也就相当于这个线程的生命周期结束了
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int wc = workerCountOf(c);
        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        try {
            // 真正响应中断是在poll()方法或者take()方法中
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            // 这里捕获中断异常
            timedOut = false;
        }
    }
}

final void tryTerminate() {
    // 自旋修改状态
    for (;;) {
        int c = ctl.get();
        // 下面几种情况不会执行后续代码
        // 1. 运行中
        // 2. 状态的值比TIDYING还大，也就是TERMINATED
        // 3. SHUTDOWN状态且任务队列不为空
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        // 工作线程数量不为0，也不会执行后续代码
        if (workerCountOf(c) != 0) {
            // 尝试中断空闲的线程
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
        
        // 获取线程池的锁
        final ReentrantLock mainLock = this.mainLock;
        // 加锁
        mainLock.lock();
        try {
            // CAS修改状态为TIDYING状态
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    // 更新成功，执行terminated钩子方法
                    terminated();
                // 确保terminated钩子方法执行完毕后，再次修改状态为TERMINATED（最终的终止状态），线程池彻底关闭
                } finally {
                    // 强制更新状态为TERMINATED，这里不需要CAS了
                    ctl.set(ctlOf(TERMINATED, 0));
                    // 通知所有等待线程
                    termination.signalAll();
                }
                return;
            }
        } finally {
            // 解锁
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

// 固定线程池
ExecutorService executorService = Executors.newFixedThreadPool(2);
// 提交单个任务，无返回值
executorService.execute(new Runnable() {
    public void run() {
    }
});
// 提交单个任务，有返回值，返回Future
executorService.submit(new Callable<String>() {
    public String call() throws Exception {
        return "abc";
    }
});

/**
 * 用给定的初始参数创建一个新的ThreadPoolExecutor。
 * 所有的构造方法最终都会调用这个构造方法，都是基于这个方法实现的
 */
public ThreadPoolExecutor(  int corePoolSize, // 线程池的核心线程数量
                            int maximumPoolSize, // 线程池的最大线程数
                            long keepAliveTime, // 当线程数大于核心线程数时，多余的空闲线程存活的最长时间
                            TimeUnit unit, // 存活时间的时间单位
                            BlockingQueue<Runnable> workQueue, // 任务队列，用来储存等待执行任务的队列
                            ThreadFactory threadFactory, // 线程工厂，用来创建线程，一般默认即可
                            RejectedExecutionHandler handler // 拒绝策略，当提交的任务过多而不能及时处理时，我们可以定制策略来处理任务
                        ) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
 
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
 
    this.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

public class ThreadPoolTest01 {
    public static void main(String[] args) {
        // 通过构造方法新建一个线程池
        // 核心数量为5，最大数量为10，空闲时间为1秒，队列长度为5，拒绝策略打印一句话
        ExecutorService threadPool = new ThreadPoolExecutor(5, 10,
                1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5),
                Executors.defaultThreadFactory(), 
                new RejectedExecutionHandler() { // 自定义拒绝策略
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        System.out.println(currentThreadName() + ", discard task");
                    }
                }
        );
        // 循环提交20个任务，注意观察num
        for (int i = 0; i < 20; i++) {
            int num = i;
            threadPool.execute(()->{
                try {
                    // 打印执行任务的线程信息 和 num的大小  以及执行的时间
                    System.out.println(currentThreadName() + ", "+ num + " running, " + System.currentTimeMillis());
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
    }
    
    // 打印当前线程的名字
    private static String currentThreadName() {
        return Thread.currentThread().getName();
    }
}

// 提交任务，任务并非立即执行，所以翻译成执行任务似乎不太合适，其实是将任务提交到任务队列中，然后线程池再从任务队列中取出任务执行
public void execute(Runnable command) {
    // 任务不能为空
    if (command == null)
        throw new NullPointerException();
        
    // 获取ctl（高3位存储状态，低29位存储工作线程的数量）
    int c = ctl.get();
    // 1. 如果工作线程数量小于核心数量
    if (workerCountOf(c) < corePoolSize) {
        // 就添加一个工作线程（核心）  并将该任务作为该工作线程的第一个任务，创建线程完成后直接返回
        if (addWorker(command, true))
            return;
        // 如果创建工作线程失败，则重新获取下ctl
        c = ctl.get();
    }
    // 2. 如果达到了核心数量且线程池是运行状态，则将任务加入队列（workQueue是构造方法的核心参数之一）
    if (isRunning(c) && workQueue.offer(command)) {
        // 再次获取ctl
        int recheck = ctl.get();
        // 再次检查线程池状态，如果不是运行状态，就移除任务并执行拒绝策略
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 容错检查工作线程数量是否为0，如果为0就创建一个，此时就不用为工作线程绑定任务了，因为任务已经加入到队列中了
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 3. 任务入队列失败（例如队列满了），尝试创建非核心工作线程，将该任务和非核心工作线程绑定在一起
    else if (!addWorker(command, false))
        // 非核心工作线程创建失败，执行拒绝策略
        reject(command);
}

public void testRetry() {
    int i = 0;
    retry:  //①
    while (true) {
        i++;
        System.out.println("i=" + i);
        int j = 0;
               // retry:   //②
        for (; ; ) {
            j++;
            System.out.println("j=" + j);
            if (j == 2) {
                break retry;
            }
        }
    }
}

// 该方法用于创建新的工作线程添加到线程池中，如果添加成功则启动线程
// 参数firstTask是一个Runnable对象，表示要给该线程的第一个任务
// 参数core表示创建的是否是核心线程
// 返回值表示是否成功创建并启动了工作线程
private boolean addWorker(Runnable firstTask, boolean core) {
    // 判断有没有资格创建新的工作线程
    // 主要是一些状态/数量的检查等等
    retry:
    for (;;) {
        // 获取ctl
        int c = ctl.get();
        // 获取当前线程池运行状态
        int rs = runStateOf(c);
        // 线程池状态检查
        if (rs >= SHUTDOWN && // 大于等于SHUTDOWN说明线程池不是RUNNING状态
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            // 两种情况会直接返回false  1.如果非运行状态  2.不是【线程池是停止状态并且传入的任务是null并且workQueue不等于空】这种情况   
            return false;
        // 工作线程数量检查
        for (;;) {
            // 获取当前线程数
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                // 判断是否饱和容量了，如果已经达到最大线程数则不能再新建线程，直接返回false
                return false;
            // 完成了上面的判断后，说明现在线程池可以创建新线程，则将线程数量加1，因为后面要创建线程了，并跳出循环
            if (compareAndIncrementWorkerCount(c))
                // 因为外面的retry:已经不在循环内了，跳到retry:位置后也不会再进入到该循环了，其实就相当于跳出这两层循环并且不再进入这两层循环，执行后续的流程
                break retry;
            // 如果上面的compareAndIncrementWorkerCount(c)方法返回false，则说明有其他线程在操作线程池的线程数量，所以需要重新获取ctl
            c = ctl.get();  // Re-read ctl
            // 如果当前的运行状态已经和最开始获取的状态不一样了
            if (runStateOf(c) != rs)
                // 回到retry:，但是因为这里用的是continue，所以程序往后执行还是会再次进入到该循环继续执行上面的判断
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }
    
    // 如果上面的条件满足，则会把工作线程数量加1，然后执行下面创建线程的动作
    // 标记是否启动了工作线程
    boolean workerStarted = false;
    // 标记是否成功添加了工作线程
    boolean workerAdded = false;
    // 要创建的Worker对象
    Worker w = null;
    try {
        // 创建工作线程
        // 增加一个worker   这个firstTask就是一个线程（任务线程，每一个Worker都和一个任务线程绑定），Runable,具体可以看这Worker源码
        w = new Worker(firstTask);
        // 这个是绑定在Worker上的工作线程，并不是任务线程，工作线程是用来执行任务的，
        final Thread t = w.thread;
        // 判断t是否为null
        if (t != null) {
            // 获取线程池的锁
            final ReentrantLock mainLock = this.mainLock;
            // 在读取线程池状态的时候就应该上锁，防止有并发操作破坏程序的一致性，上下不一致
            mainLock.lock();
            try {
                // 锁定后并重新检查线程池的状态，查看是否存在线程工厂的失败或者锁定前的关闭
                // 获取线程池运行状态
                int rs = runStateOf(ctl.get());
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    
                    // 添加到工作线程队列。 workers是线程池中存放工作线程的集合
                    // 增加work  每一个线程池都持有一个workers集合，里面存储着线程池的所有worker，也就是所有线程
                    workers.add(w);
                    // 还在池子中的线程数量（只能在mainLock中使用）
                    int s = workers.size();
                    // 不能超过线程池的最大线程数量
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    
                    // 标记线程添加成功
                    workerAdded = true;
                }
            } finally {
                // 解锁
                mainLock.unlock();
            }
            if (workerAdded) {
                // 线程添加成功之后启动线程
                // 启动绑定在worker上的线程。启动了之后该工作线程就会开始从任务队列中拿任务去执行了
                t.start();
                // 标记工作线程启动成功
                workerStarted = true;
            }
        }
    } finally {
        // 线程启动失败，执行失败方法（线程数量减1，执行tryTerminate()方法等）
        if (! workerStarted)
            addWorkerFailed(w);
    }
    // 返回新创建的工作线程是否启动成功
    return workerStarted;
}

private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    // 加锁
    mainLock.lock();
    try {
        // 移除添加失败的worker
        if (w != null)
            workers.remove(w);
        // 减少worker数量
        decrementWorkerCount();
        // 尝试终止线程池
        tryTerminate();
    } finally {
        // 解锁
        mainLock.unlock();
}

public class ThreadPoolTest02 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 新建一个固定5个线程的线程池
        ExecutorService threadPool = Executors.newFixedThreadPool(5);
        List<Future<Integer>> futureList = new ArrayList<>();
        // 提交5个任务，分别返回0、1、2、3、4
        for (int i = 0; i < 5; i++) {
            int num = i;
            // 使用submit()方法来提交任务，任务执行的结果用Future包装
            Future<Integer> future = threadPool.submit(() -> {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("return: " + num);
                // 返回值
                return num;
            });
            // 把future添加到list中
            futureList.add(future);
        }
        // 任务全部提交完再从future中get返回值，并做累加
        int sum = 0;
        for (Future<Integer> future : futureList) {
            sum += future.get();
        }
        System.out.println("sum=" + sum);
    }
}

public <T> Future<T> submit(Callable<T> task) {
    // 非空检测
    if (task == null) throw new NullPointerException();
    // 将任务包装成FutureTask
    RunnableFuture<T> ftask = newTaskFor(task);
    // 交给execute()方法去执行
    execute(ftask);
    // 返回futureTask
    return ftask;
}
 
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    // 将普通任务包装成FutureTask
    return new FutureTask<T>(callable);
}

public void run() {
    // FutureTask任务的状态不为NEW，或者修改为当前线程来运行这个任务失败，则直接返回
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    
    try {
        // 真正的任务，存储在FutureTask类的成员属性callable中
        Callable<V> c = callable;
        // 任务的state必须为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是Future的成员属性，用来保存执行任务的线程 
        runner = null;
        // 获取任务状态
        int s = state;
        // 如果任务执行过程中被中断了，则处理中断
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

// 保存任务执行过程中出现的异常
// t为要保存的异常信息
protected void setException(Throwable t) {
    // 将状态从NEW置为COMPLETING
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        // 返回值置为传进来的异常（outcome就是FutureTask调用get()方法时返回的内容）
        outcome = t;
        // 最终的状态设置为EXCEPTIONAL
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        // 调用完成方法
        finishCompletion();
    }
}
 
// 保存任务执行结果
// v为任务执行完返回的结果
protected void set(V v) {
    // 将状态从NEW置为COMPLETING
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        // 返回值置为传进来的结果（outcome就是FutureTask调用get()方法时返回的内容）
        outcome = v;
        // 最终的状态设置为NORMAL
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        // 调用完成方法
        finishCompletion();
    }
}

private void finishCompletion() {
    // 如果队列不为空（waiters是FutureTask中的成员属性，也就是get方法阻塞的线程队列，这个队列实际上为调用者线程）
    for (WaitNode q; (q = waiters) != null;) {
        // 置空队列
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                // 调用者线程，例如调用FutureTask对象的get()方法的线程就是调用者线程
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    // 如果调用者线程不为空，则唤醒它。也就是当任务还没有执行玩，调用者就去执行任务FutureTask.get()，调用者就会被阻塞在get()中，
                    // 当任务执行完成后，FutureTask.run()方法就会调用finishCompletion()方法中的LockSupport.unpark(t)，唤醒调用者线程，这样调用者就能得到任务执行结果了。
                    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
}

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    // 绑定任务到FutureTask
    this.callable = callable;
    // 初始化任务状态
    this.state = NEW;       // ensure visibility of callable
}

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    // 如果状态小于等于COMPLETING，则进入队列等待
    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()) {
            removeWaiter(q);
            throw new InterruptedException();
        }
        // 4. 如果状态大于COMPLETING了，则跳出循环并返回
        // 这是自旋的出口
        int s = state;
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        // 如果状态等于COMPLETING，说明任务快完成了，就差设置状态到NORMAL或EXCEPTIONAL和设置结果了
        // 这时候就让出CPU，优先完成任务
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
        // 1. 如果队列为空
        else if (q == null)
            // 初始化队列（WaitNode中记录了调用者线程）
            q = new WaitNode();
        // 2. 未进入队列
        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);
        }
        // 3. 阻塞当前线程（调用者线程）
        else
            LockSupport.park(this);
    }
}

private V report(int s) throws ExecutionException {
    // 获取任务执行结果  outcome是FutureTask类的成员属性，用来存储任务执行结果 
    Object x = outcome;
    // 任务正常结束
    if (s == NORMAL)
        return (V)x;
    // 被取消了
    if (s >= CANCELLED)
        throw new CancellationException();
    // 执行异常，则将异常结果抛出
    throw new ExecutionException((Throwable)x);
}

public boolean cancel(boolean mayInterruptIfRunning) {
    // 如果任务已经完成或者已经被取消，直接返回
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try {    // in case call to interrupt throws exception
        if (mayInterruptIfRunning) {
            try {
                // 执行任务的线程
                Thread t = runner;
                // 取消任务是通过中断执行任务的线程实现的
                if (t != null)
                    t.interrupt();
            } finally { // final state
                // 设置任务状态为中断
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}

private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}