ThreadPoolExecutor 源码解析

//控制线程池中数量和状态的字段，用 AtomicInteger 保存
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//count bit 顾名思义就是 workerCount 的位数，这里是 29
private static final int COUNT_BITS = Integer.SIZE - 3;
//1<<29 -1 == 1111....1111(29 个 1) 线程数 (workerCount) 上限 大约 5 亿
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
//高三位存放状态，相应的低 29 位就是 workerCount
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

// Packing and unpacking ctl
// 拆解 ctl 获取状态和数量
private static int runStateOf(int c)     { return c & ~CAPACITY; }
private static int workerCountOf(int c)  { return c & CAPACITY; }
// 拼接状态和数量得到 ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }

//任务缓存队列，存放待执行的任务
private final BlockingQueue<Runnable> workQueue; 
//可重入锁，线程池主要的锁
private final ReentrantLock mainLock = new ReentrantLock();
//线程集合（线程池的 workers 集合）
private final HashSet<Worker> workers = new HashSet<Worker>();
//对应的条件变量
private final Condition termination = mainLock.newCondition(); 
//用来记录线程池中曾经出现过的最大线程数，和线程池容量没有关系
private int largestPoolSize;
//用来记录已经执行完毕的任务个数
private long completedTaskCount; 
//工厂方法，用来创建线程
private volatile ThreadFactory threadFactory; 
//拒绝策略
private volatile RejectedExecutionHandler handler; 
//线程闲置时候的最大存活时间
private volatile long keepAliveTime; 
//是否允许核心线程闲置的时候超时
private volatile boolean allowCoreThreadTimeOut; 
//核心线程数
private volatile int corePoolSize; 
//最大线程数
private volatile int maximumPoolSize; 
//默认的拒绝策略：AbortPolicy 直接拒绝并抛异常
private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    //获取线程池的状态和线程数量
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) { //workerCount 小于核心线程数量
        //将任务添加到 workers 中，第二个参数代表是否根据 corePoolSize 来添加线程，false 则根据 maxPoolSize
        if (addWorker(command, true))
            return;
        //添加失败，重新获取 ctl
        c = ctl.get();
    }
    
    //上面添加到 workers 中失败，有可能是核心线程不够用了或者线程池不是运行状态
    //如果线程池是 Running 状态 尝试添加任务到阻塞队列中
    if (isRunning(c) && workQueue.offer(command)) {
        //添加到等待队列成功，重新获取 ctl
        int recheck = ctl.get();
        //如果线程池不是 Running 状态就从等待队列中 remove 这个任务
        if (! isRunning(recheck) && remove(command))
            reject(command); //采用拒绝策略拒绝该任务
        else if (workerCountOf(recheck) == 0)
            //线程池 Running 但是没有线程
            //创建一个线程但是不传入 Runnable（已经在阻塞队列中了）
            addWorker(null, false);
    }
    //两种情况
    //1. 线程池不是 Running 状态并且 command 不为 null，addWorker 会直接 false 然后拒绝这个任务
    //2. 添加到 workQueue（阻塞队列）失败，也就是 queue 满了，可能是需要扩容了
    //   所以后面的参数是 false，添加失败也会直接拒绝
    else if (!addWorker(command, false))
        reject(command);
}

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // rs>=SHUTDOWN 说明不是 RUNNING 状态
        if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null &&!workQueue.isEmpty()))
            return false;
        
        for (;;) {
            int wc = workerCountOf(c);
            //如果大于可允许的最大线程数，或者大于当前的线程池的上界，直接 false
            //上面传入的第二个参数的作用体现出来了，为 true 上界则是 corePoolSize
            if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            //利用 CAS 自增 增大 workCount 线程数，成功后就跳出循环
            if (compareAndIncrementWorkerCount(c))
                break retry;
            //CAS 失败，重新获取 ctl
            c = ctl.get();  // Re-read ctl
            //判断还是不是 Running 状态（能到这里说明 rs==Running 状态），不是的话就跳出去重新来过
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
            // CAS 失败并且还是 Running 状态，继续自旋尝试自增
        }
    }
    //线程是否启动，以及线程是否添加
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        //创建一个 Worker（对线程的封装）
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                int rs = runStateOf(ctl.get());
				//是 RUNNING 状态 或者 是 SHUTDOWN 状态且没有提交任务 (SHUTDOWN 状态还可以执行阻塞队列的任务）
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    //添加到工作集中
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s; //记录最大值
                    workerAdded = true; //添加成功
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                //启动线程，执行 Worker 的 run 方法
                t.start(); 
                workerStarted = true; //启动成功
            }
        }
    } finally {
        if (! workerStarted)
            //启动失败，回滚 workers 并尝试关闭线程池
            addWorkerFailed(w);
    }
    return workerStarted;
}

private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    //获取锁
    mainLock.lock();
    try {
        if (w != null)
            //从 workers 中移除
            workers.remove(w);
        //减少 workerCount
        decrementWorkerCount();
        //尝试关闭线程池
        tryTerminate();
    } finally {
        mainLock.unlock();
    }
}

private final class Worker extends AbstractQueuedSynchronizer implements Runnable
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;

    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    Runnable firstTask; //传入的任务
    /** Per-thread task counter */
    volatile long completedTasks;

    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
    Worker(Runnable firstTask) {
        setState(-1); // 设置状态为 -1
        this.firstTask = firstTask;
        //根据工厂方法创建线程
        //将 Worker 传递进去，作为 Thread 的参数
        //new Thread(Worker worker);
        this.thread = getThreadFactory().newThread(this);
    }

    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }

    // Lock methods
    //
    // The value 0 represents the unlocked state.
    // The value 1 represents the locked state.
    // 是否获取到了锁
    protected boolean isHeldExclusively() {
        return getState() != 0; 
    }

    protected boolean tryAcquire(int unused) {
        //利用 CAS 设置 state
        //很明显这里是个不可重入的独占锁，具体可以对比 ReentrantLock 的实现方法
        if (compareAndSetState(0, 1)) {
            //继承自 AbstractOwnableSynchronizer
            //保存当前的持有锁的线程
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }
    
    //释放锁设置 state=0
    protected boolean tryRelease(int unused) {
        //设置独占锁线程为空
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock()        { acquire(1); } //最终会调用 tryAcquire(1);
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); } //最终会调用 tryRelease(1);
    public boolean isLocked() { return isHeldExclusively(); }

    //打断已经启动的线程
    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

final void runWorker(Worker w) {
    //当前执行线程
    Thread wt = Thread.currentThread();
    //拿到任务
    Runnable task = w.firstTask; 
    w.firstTask = null;
    //前面构造 worker 的时候设置了 state=-1
    //设置 state 为 0，tryRelease(1)
    //其实这样是为了在这之前不会被打断（还没有开始执行任务）
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true; //是否因为异常而退出？
    try {
        //task 为空就从阻塞队列中拿任务
        while (task != null || (task = getTask()) != null) {
            //加锁，但是这个锁不会和其他的 Worker 互斥，这个锁只是用来判断 worker 是否在工作
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                //打断当前执行线程 wt
                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++; //完成任务数++
                w.unlock();
            }
        }
        completedAbruptly = false; //下面的收尾工作会根据这个判断是否调整线程数
    } finally {
        processWorkerExit(w, completedAbruptly); //收尾工作
    }
}

private Runnable getTask() {
    //是否超时？和 keepAliveTime 相关
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        // 如果线程池状态为 SHUTDOWN 并且任务队列没任务 或者 线程状态>=STOP
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        // 是否允许线程超时 允许核心线程超时（主动超时）？ 或者 wc 大于了核心线程数（被动超时）
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
		//此时需要回收多余的线程
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c)) //cas 减少线程数
                return null; //返回 null
            continue; //cas 失败 继续循环
        }

        try {
            //根据 timed 判断是限时获取还是直接获取
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take(); //阻塞的获取
            if (r != null)
                return r;
            //没获取到，超时了
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        //如果是因为异常而退出则 getTask() 没有机会去调整线程数所以需要在这里调整
        decrementWorkerCount();  // wc--

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //完成任务总数+=该 worker 完成的任务数
        completedTaskCount += w.completedTasks;
        //移除出线程集 workers
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    
    //尝试结束线程池
    tryTerminate();
    
    //获取 ctl
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
        //非异常结束
        if (!completedAbruptly) {
            //最小值根据是否允许核心线程超时来判断
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1; //如果最小为 0 并且任务队列不为空则保证线程池中至少有一个线程执行这些任务
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        //1. 异常结束，新增加一个线程到线程池，相当于替换了异常的线程
        //2. 非异常结束，工作线程小于核心线程，增加线程，确保在不允许核心线程超时的情况下线程数不小于 corePoolSize
        addWorker(null, false);
    }
}

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        //如果是在 RUNNING 状态 直接 return
        //如果已经是 TIDYING 或者 TERMINATED 状态 直接 return
        //如果是 SHUTDOWN 状态并且阻塞队列中还有任务 直接 return
        //return 就说明线程池还不到关闭的时候
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        //到这里就说明要么是 STOP 状态，要么是 SHUTDOWN 状态阻塞队列也没任务了
        if (workerCountOf(c) != 0) { // Eligible to terminate
            //中断一个空闲的 worker 线程
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
		
        //wc==0 没有线程存活了
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //CAS 自旋 修改线程池状态为 TIDYING
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated(); //空方法，交给子类去实现的
                } finally {
                    //设置状态为 TERMINATED
                    ctl.set(ctlOf(TERMINATED, 0));
                    //唤醒 termination 上 awiat 的线程
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

public boolean awaitTermination(long timeout, TimeUnit unit)
    throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (;;) {
            if (runStateAtLeast(ctl.get(), TERMINATED))
                return true;
            if (nanos <= 0)
                return false;
            //限时等待
            nanos = termination.awaitNanos(nanos);
        }
    } finally {
        mainLock.unlock();
    }
}

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        //CAS 自旋设置线程池状态为 SHUTDOWN
        advanceRunState(SHUTDOWN);
        //打断 idle（空闲）的 worker
        interruptIdleWorkers(); 
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock(); //加锁，因为 Wokers 是 HashSet 是线程不安全的
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            //没有被打断并且没有在工作（空闲）
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    //打断 Worker 里面的线程
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            //只打断一个就直接 break
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        //CAS 自旋设置状态为 STOP
        advanceRunState(STOP);
        //打断所有的线程包括正在工作的线程
        interruptWorkers();
        //排干阻塞队列，因为已经 STOP 了不会再执行队列里面的任务了
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    //尝试关闭线程池
    tryTerminate();
    return tasks;
}

private void interruptWorkers() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers)
            //打断所有启动的线程
            w.interruptIfStarted();
    } finally {
        mainLock.unlock();
    }
}

    /**
     * Lock held on access to workers set and related bookkeeping.
     * While we could use a concurrent set of some sort, it turns out
     * to be generally preferable to use a lock. Among the reasons is
     * that this serializes interruptIdleWorkers, which avoids
     * unnecessary interrupt storms, especially during shutdown.
     * Otherwise exiting threads would concurrently interrupt those
     * that have not yet interrupted. It also simplifies some of the
     * associated statistics bookkeeping of largestPoolSize etc. We
     * also hold mainLock on shutdown and shutdownNow, for the sake of
     * ensuring workers set is stable while separately checking
     * permission to interrupt and actually interrupting.
     */
    private final ReentrantLock mainLock = new ReentrantLock();

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 static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

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>()));
}