Java并发编程笔记——J.U.C之executors框架：ThreadPoolExecutor

public class ThreadPoolExecutor extends AbstractExecutorService {

public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
    }

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}

/**
 * 使用给定的参数创建ThreadPoolExecutor.
 *
 * @param corePoolSize    核心线程池中的最大线程数
 * @param maximumPoolSize 总线程池中的最大线程数
 * @param keepAliveTime   空闲线程的存活时间
 * @param unit            keepAliveTime的单位
 * @param workQueue       任务队列, 保存已经提交但尚未被执行的线程
 * @param threadFactory   线程工厂(用于指定如果创建一个线程)
 * @param handler         拒绝策略 (当任务太多导致工作队列满时的处理策略)
 */
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.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);   // 使用纳秒保存存活时间
    this.threadFactory = threadFactory;
    this.handler = handler;
}

/**
 * 保存线程池状态和工作线程数:
 * 低29位: 工作线程数
 * 高3位 : 线程池状态
 */
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
 
private static final int COUNT_BITS = Integer.SIZE - 3;
 
// 最大线程数: 2^29-1
private static final int CAPACITY = (1 << COUNT_BITS) - 1;  // 00011111 11111111 11111111 11111111
 
// 线程池状态
private static final int RUNNING = -1 << COUNT_BITS;        // 11100000 00000000 00000000 00000000
private static final int SHUTDOWN = 0 << COUNT_BITS;        // 00000000 00000000 00000000 00000000
private static final int STOP = 1 << COUNT_BITS;            // 00100000 00000000 00000000 00000000
private static final int TIDYING = 2 << COUNT_BITS;         // 01000000 00000000 00000000 00000000
private static final int TERMINATED = 3 << COUNT_BITS;      // 01100000 00000000 00000000 00000000

/**
 * 工作线程集合.
 */
private final HashSet<Worker> workers = new HashSet<Worker>();

/**
 * Worker表示线程池中的一个工作线程, 可以与任务相关联.
 * 由于实现了AQS框架, 其同步状态值的定义如下:
 * -1: 初始状态
 * 0:  无锁状态
 * 1:  加锁状态
 */
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
 
    /**
     * 与该Worker关联的线程.
     */
    final Thread thread;
    /**
     * Initial task to run.  Possibly null.
     */
    Runnable firstTask;
    /**
     * Per-thread task counter
     */
    volatile long completedTasks;
 
 
    Worker(Runnable firstTask) {
        setState(-1); // 初始的同步状态值
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
 
    /**
     * 执行任务
     */
    public void run() {
        runWorker(this);
    }
 
    /**
     * 是否加锁
     */
    protected boolean isHeldExclusively() {
        return getState() != 0;
    }
 
    /**
     * 尝试获取锁
     */
    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }
 
    /**
     * 尝试释放锁
     */
    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }
 
    public void lock() {
        acquire(1);
    }
 
    public boolean tryLock() {
        return tryAcquire(1);
    }
 
    public void unlock() {
        release(1);
    }
 
    public boolean isLocked() {
        return isHeldExclusively();
    }
 
    /**
     * 中断线程(仅任务非初始状态)
     */
    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}

/**
 * 默认的线程工厂.
 */
static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);//用于统计线程工厂个数
    private final ThreadGroup group;
    //用来统计每个线程工厂创建了多少线程
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;
 
    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
    }
 
    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

public <T> Future<T> submit(Runnable task, T result) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task, result);
    execute(ftask);
    return ftask;
}

public void execute(Runnable command) {
    //【1】如果任务为null，则抛出NPE异常
    if (command == null)
        throw new NullPointerException();

    //【2】获取当前线程池的状态+线程个数变量的组合值
    int c = ctl.get();

    【3】当前线程池中线程个数是否小于corePoolSize，小于则开启新线程运行
    if (workerCountOf(c) < corePoolSize) {              // CASE1: 工作线程数 < 核心线程池上限
        if (addWorker(command, true))             // 添加工作线程并执行
            return;
        c = ctl.get();
    }
 
    //【4】如果线程池处于RUNNING状态，则添加任务到阻塞队列
    if (isRunning(c) && workQueue.offer(command)) {     // CASE2: 插入任务至队列
 
        // 【4.1】二次检查
        int recheck = ctl.get();

        //【4.2】如果当前线程池状态不是RUNNING则从队列中删除任务，并执行拒绝策略
        if (!isRunning(recheck) && remove(command))
            reject(command);

        //【4.3】否则如果当前线程池为空，则添加一个线程
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);

    //【5】如果队列满了，则新增线程，新增失败则执行拒绝策略
    } else if (!addWorker(command, false))        // CASE3: 插入队列失败, 判断工作线程数 < 总线程池上限
        reject(command);    // 执行拒绝策略
}

/**
 * 添加工作线程并执行任务
 *
 * @param firstTask 如果指定了该参数, 表示将立即创建一个新工作线程执行该firstTask任务; 否则复用已有的工作线程，从工作队列中获取任务并执行
 * @param core      执行任务的工作线程归属于哪个线程池:  true-核心线程池  false-非核心线程池
 */
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    
    for (; ; ) { 
        int c = ctl.get();
        int rs = runStateOf(c);             // 获取线程池状态
 
        /**
         * 这个if主要是判断哪些情况下, 线程池不再接受新任务执行, 而是直接返回.总结下, 有以下几种情况：
         * 1. 线程池状态为 STOP 或 TIDYING 或 TERMINATED: 线程池状态为上述任一一种时, 都不会再接受任务，所以直接返回
         * 2. 线程池状态≥ SHUTDOWN 且 firstTask != null: 因为当线程池状态≥ SHUTDOWN时, 不再接受新任务的提交，所以直接返回
         * 3. 线程池状态≥ SHUTDOWN 且 队列为空: 队列中已经没有任务了, 所以也就不需要执行任何任务了，可以直接返回
         */
        if (rs >= SHUTDOWN &&
                !(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty()))
            return false;
 
        //循环CAS增加线程数
        for (; ; ) {
            int wc = workerCountOf(c);      // 获取工作线程数
 
            /**
             * 这个if主要是判断工作线程数是否超限, 以下任一情况属于属于超限, 直接返回:
             * 1. 工作线程数超过最大工作线程数(2^29-1)
             * 2. 工作线程数超过核心线程池上限(入参core为true, 表示归属核心线程池)
             * 3. 工作线程数超过总线程池上限(入参core为false, 表示归属非核心线程池)
             */
            if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
 
            if (compareAndIncrementWorkerCount(c))  // 工作线程数加1
                break retry;                        // 跳出最外层循环
 
            c = ctl.get();
            if (runStateOf(c) != rs)                // 线程池状态发生变化, 重新自旋判断
                continue retry;
        }
    }
  
    //到这里说明CAS自旋成功了
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        w = new Worker(firstTask);                  // 将任务包装成工作线程
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();    //加入独占锁，为了实现workers同步，因为可能多个线程调用了线程池的execute方法
            try {
                // 重新检查线程池状态，以避免在获取锁前调用了shutdown接口
                int rs = runStateOf(ctl.get());
                if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive())               
                        throw new IllegalThreadStateException();
                    workers.add(w);                 // 加入工作线程集合
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            //添加成功后，则启动任务
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (!workerStarted)     // 创建/启动工作线程失败, 需要执行回滚操作
            addWorkerFailed(w);
    }
    return workerStarted;
}

/**
 * 执行任务
 */
public void run() {
    runWorker(this);
}

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();     // 执行任务的线程
    Runnable task = w.firstTask;            // 任务, 如果是null则从队列取任务
    w.firstTask = null;
    w.unlock();                             // 允许执行线程被中断
    boolean completedAbruptly = true;       // 表示是否因为中断而导致退出
    try {
        while (task != null || (task = getTask()) != null) {    // 当task==null时会通过getTask从队列取任务
            w.lock();
 
            /**
             * 下面这个if判断的作用如下:
             * 1.保证当线程池状态为STOP/TIDYING/TERMINATED时，当前执行任务的线程wt是中断状态(因为线程池处于上述任一状态时，均不能再执行新任务)
             * 2.保证当线程池状态为RUNNING/SHUTDOWN时，当前执行任务的线程wt不是中断状态
             */
            if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                wt.interrupt();
 
            try {
                beforeExecute(wt, task);            // 钩子方法，由子类自定义实现
                Throwable thrown = null;
                try {
                    task.run();                     // 执行任务
                } catch (RuntimeException x) {
                    thrown = x;
                    throw x;
                } catch (Error x) {
                    thrown = x;
                    throw x;
                } catch (Throwable x) {
                    thrown = x;
                    throw new Error(x);
                } finally {
                    afterExecute(task, thrown);     // 钩子方法，由子类自定义实现
                }
            } finally {
                task = null;
                w.completedTasks++;     // 完成任务数+1
                w.unlock();
            }
        }
 
        // 执行到此处, 说明该工作线程自身既没有携带任务, 也没从任务队列中获取到任务
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);    // 处理工作线程的退出工作
    }

}

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly)          // 工作线程因异常情况而退出
        decrementWorkerCount();     // 工作线程数减1(如果工作线程执行时没有出现异常, 在getTask()方法中已经对线程数减1了)
 
    //统计整个线程池完成的任务个数，并从工作集中删除当前worker
    final ReentrantLock mainLock = this.mainLock; //全局锁
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks; // completedTaskCount记录线程池完成的总任务数
        workers.remove(w);                      // 从工作线程集合中移除(该工作线程会自动被GC回收)
    } finally {
        mainLock.unlock();
    }
 
    tryTerminate();                             // 根据线程池状态, 判断是否需要终止线程池
 
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {            // 如果线程池状态为RUNNING/SHUTDOWN
        if (!completedAbruptly) {               // 工作线程为正常退出
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && !workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);  // 新建一个工作线程
    }
}

private Runnable getTask() {
    boolean timedOut = false;       // 表示上次从阻塞队列中取任务时是否超时
 
    for (; ; ) {
        int c = ctl.get();
        int rs = runStateOf(c);     // 获取线程池状态
 
        /**
         * 以下IF用于判断哪些情况下不允许再从队列获取任务:
         * 1. 线程池进入停止状态（STOP/TIDYING/TERMINATED）, 此时即使队列中还有任务未执行, 也不再执行
         * 2. 线程池非RUNNING状态, 且队列为空
         */
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount(); // 工作线程数减1
            return null;
        }
 
        int wc = workerCountOf(c);  // 获取工作线程数
 
        /**
         * timed变量用于判断是否需要进行超时控制:
         * 对于核心线程池中的工作线程, 除非设置了allowCoreThreadTimeOut==true, 否则不会超时回收;
         * 对于非核心线程池中的工作线程, 都需要超时控制
         */
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
        // 这里主要是当外部通过setMaximumPoolSize方法重新设置了最大线程数时,需要回收多出的工作线程
        if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
 
        try {
            Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
            if (r != null)
                return r;
            timedOut = true;    // 超时仍未获取到任务
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

public interface RejectedExecutionHandler {
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

public static class AbortPolicy implements RejectedExecutionHandler {
    public AbortPolicy() {
    }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                " rejected from " +
                e.toString());
    }
}

public static class DiscardPolicy implements RejectedExecutionHandler {
    public DiscardPolicy() {
    }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    public DiscardOldestPolicy() {
    }
 
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {      // 线程池未关闭(RUNNING)
            e.getQueue().poll();    // 丢弃任务队列中的最近任务
            e.execute(r);           // 执行当前任务
        }
    }
}

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    public CallerRunsPolicy() {
    }
 
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {  // 线程池未关闭(RUNNING)
            r.run();            // 执行当前任务
        }
    }
}

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(SHUTDOWN);  // 如果线程池为RUNNING状态, 则切换为SHUTDOWN状态
        interruptIdleWorkers();     // 中断所有空闲线程
        onShutdown();               // 钩子方法, 由子类实现
    } finally {
        mainLock.unlock();
    }
    tryTerminate();                 
}

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(STOP);  // 如果线程池为RUNNING或SHUTDOWN状态, 则切换为STOP状态
        interruptWorkers();     // 中断所有工作线程
        tasks = drainQueue();   // 抽空任务队列中的所有任务
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}