详解Java并发包中线程池ThreadPoolExecutor

//（高3位）表示线程池状态，（低29位）表示线程池中线程的个数；
// 默认状态是RUNNING，线程池中线程个数为0
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

//表示具体平台下Integer的二进制位数-3后的剩余位数表示的数才是线程的个数;
//其中Integer.SIZE=32,-3之后的低29位表示的就是线程的个数了
private static final int COUNT_BITS = Integer.SIZE - 3;

//线程最大个数（低29位）00011111111111111111111111111111（1<<29-1）
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

//线程池状态（高3位表示线程池状态）
//111 00000000000000000000000000000
private static final int RUNNING    = -1 << COUNT_BITS;

//000 00000000000000000000000000000
private static final int SHUTDOWN   =  0 << COUNT_BITS;

//001 00000000000000000000000000000
private static final int STOP       =  1 << COUNT_BITS;

//010 00000000000000000000000000000
private static final int TIDYING    =  2 << COUNT_BITS;

//011 00000000000000000000000000000
private static final int TERMINATED =  3 << COUNT_BITS;

//获取高3位（运行状态）==> c & 11100000000000000000000000000000
private static int runStateOf(int c)     { return c & ~CAPACITY; }

//获取低29位（线程个数）==> c &  00011111111111111111111111111111
private static int workerCountOf(int c)  { return c & CAPACITY; }

//计算原子变量ctl新值（运行状态和线程个数）
private static int ctlOf(int rs, int wc) { return rs | wc; }

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

//独占锁，用来控制新增工作线程Worker操作的原子性
private final ReentrantLock mainLock = new ReentrantLock();

//工作线程集合，Worker继承了AQS接口和Runnable接口，是具体处理任务的线程对象
//Worker实现AQS，并自己实现了简单不可重入独占锁，其中state=0表示当前锁未被获取状态，state=1表示锁被获取，
//state=-1表示Work创建时候的默认状态，创建时候设置state=-1是为了防止runWorker方法运行前被中断
private final HashSet<Worker> workers = new HashSet<Worker>();

//termination是该锁对应的条件队列，在线程调用awaitTermination时候用来存放阻塞的线程
private final Condition termination = mainLock.newCondition();

public void execute(Runnable command) {
    //（1）首先检查任务是否为null，为null抛出异常，否则进行下面的步骤
    if (command == null)
        throw new NullPointerException();
    //（2）ctl值中包含了当前线程池的状态和线程池中的线程数量
    int c = ctl.get();
    //（3）workerCountOf方法是获取低29位，即获取当前线程池中的线程个数，如果小于corePoolSize，就开启新的线程运行
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //（4）如果线程池处理RUNNING状态，就添加任务到阻塞队列中
    if (isRunning(c) && workQueue.offer(command)) {
        //（4-1）二次检查，获取ctl值
        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))
        reject(command);
}

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

        //(6)检查队列是否只在必要时候为空
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        //(7)使用CAS增加线程个数
        for (;;) {
            //根据ctl值获得当前线程池中的线程数量
            int wc = workerCountOf(c);
            //(7-1)如果线程数量超出限制，返回false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            //(7-2)CAS增加线程数量，同时只有一个线程可以成功
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // 重新读取ctl值
            //(7-3)CAS失败了，需要查看当前线程池状态是否发生变化，如果发生变化需要跳转到外层循环尝试重新获取线程池状态，否则内层循环重新进行CAS增加线程数量
            if (runStateOf(c) != rs)
                continue retry;
        }
    }

    //(8)执行到这里说明CAS增加新线程个数成功了，我们需要开始创建新的工作线程Worker
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        //(8-1)创建新的worker
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            //(8-2)加独占锁，保证workers的同步，可能线程池中的多个线程调用了线程池的execute方法
            mainLock.lock();
            try {
                // (8-3)重新检查线程池状态，以免在获取锁之前调用shutdown方法改变线程池状态
                int rs = runStateOf(ctl.get());

                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    //(8-4)添加新任务
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            //(8-6)添加新任务成功之后，启动任务
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    //获取锁
    mainLock.lock();
    try {
      //如果worker不为null
      if (w != null)
          //workers移除worker
          workers.remove(w);
      //通过CAS操作，workerCount-1
      decrementWorkerCount();
      tryTerminate();
    } finally {
      //释放锁
      mainLock.unlock();
    }
}

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{

    /** 使用线程工厂创建的线程，执行任务 */
    final Thread thread;
    /** 初始化执行任务 */
    Runnable firstTask;
    /** 计数 */
    volatile long completedTasks;

    /**
     * 给出初始firstTask，线程创建工厂创建新的线程
     */
    Worker(Runnable firstTask) {
        setState(-1); // 防止在调用runWorker之前被中断
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this); //使用threadFactory创建线程
    }

    /** run方法实际上执行的是runWorker方法  */
    public void run() {
        runWorker(this);
    }

    // 关于同步状态（锁）
    //
    // 同步状态state=0表示锁未被获取
    // 同步状态state=1表示锁被获取

    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    //下面都是重写AQS的方法，Worker为自定义的同步组件
    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) {
            }
        }
    }
}

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // 这个时候调用unlock方法，将state置为0，就可以被中断了
    boolean completedAbruptly = true;
    try {
        //(10)如果当前任务为null，或者从任务队列中获取到的任务为null，就跳转到(11)处执行清理工作
        while (task != null || (task = getTask()) != null) {
            //task不为null，就需要线程执行任务，这个时候，需要获取工作线程内部持有的独占锁
            w.lock();
            /**如果线程池已被停止(STOP)(至少大于STOP状态)，要确保线程都被中断
             * 如果状态不对，检查当前线程是否中断并清除中断状态，并且再次检查线程池状态是否大于STOP
             * 如果上述满足，检查该对象是否处于中断状态，不清除中断标记
             */
            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++;
                //释放锁
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        //执行清理工作：处理并退出当前worker
        processWorkerExit(w, completedAbruptly);
    }
}

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    /*
    *completedAbruptly:是由runWorker传过来的参数，表示是否突然完成的意思
    *当在就是在执行任务过程当中出现异常，就会突然完成，传true
    *
    *如果是突然完成，需要通过CAS操作，更新workerCount(-1操作)
    *不是突然完成，则不需要-1，因为getTask方法当中已经-1(getTask方法中执行了decrementWorkerCount()方法)
    */
    if (completedAbruptly)
        decrementWorkerCount();
    //(1-1)在统计完成任务个数之前加上全局锁，然后统计线程池中完成的任务个数并更新全局计数器，并从工作集中删除当前worker
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock(); //获得全局锁
    try {
        completedTaskCount += w.completedTasks; //更新已完成的任务数量
        workers.remove(w); //将完成该任务的线程worker从工作线程集合中移除
    } finally {
        mainLock.unlock(); //释放锁
    }
    /**(1-2)
     * 这一个方法调用完成了下面的事情：
     * 判断如果当前线程池状态是SHUTDOWN并且工作队列为空，
     * 或者当前线程池状态是STOP并且当前线程池里面没有活动线程，
     * 则设置当前线程池状态为TERMINATED，如果设置成了TERMINATED状态，
     * 还需要调用条件变量termination的signAll方法激活所有因为调用线程池的awaitTermination方法而被阻塞的线程
     */
    tryTerminate();

    //（1-3）如果当前线程池中线程数小于核心线程，则增加核心线程数
    int c = ctl.get();
    //判断当前线程池的状态是否小于STOP（RUNNING或者SHUTDOWN）
    if (runStateLessThan(c, STOP)) {
        //如果任务忽然完成，执行后续的代码
        if (!completedAbruptly) {
            //allowCoreThreadTimeOut表示是否允许核心线程超时，默认为false
            //min这里当默认为allowCoreThreadTimeOut默认为false的时候，min置为coorPoolSize
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            //这里说明：如果允许核心线程超时，那么allowCoreThreadTimeOut可为true，那么min值为0，不需要维护核心线程了
            //如果min为0并且任务队列不为空
            if (min == 0 && ! workQueue.isEmpty())
                min = 1; //这里表示如果min为0，且队列不为空，那么至少需要一个核心线程存活来保证任务的执行
            //如果工作线程数大于min,表示当前线程数满足，直接返回
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

final void tryTerminate() {
    for (;;) {
        //获取线程池状态
        int c = ctl.get();
        //如果线程池状态为RUNNING
        //或者状态大于TIDYING
        //或者状态==SHUTDOWN并未任务队列不为空
        //直接返回，不能调用terminated方法
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        //如果线程池中工作线程数不为0，需要中断线程
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
        //获得线程池的全局锁
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //通过CAS操作，将线程池状态设置为TIDYING
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { //private static int ctlOf(int rs, int wc) { return rs | wc; }
                try {
                    //调用terminated方法
                    terminated();
                } finally {
                    //最终将线程状态设置为TERMINATED
                    ctl.set(ctlOf(TERMINATED, 0));
                    //调用条件变量termination的signaAll方法唤醒所有因为
                    //调用线程池的awaitTermination方法而被阻塞的线程
                    //private final Condition termination = mainLock.newCondition();
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

public void shutdown() {
    //获得线程池的全局锁
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //进行权限检查
        checkShutdownAccess();

        //设置当前线程池的状态的SHUTDOWN，如果线程池状态已经是该状态就会直接返回，下面我们会分析这个方法的源码
        advanceRunState(SHUTDOWN);

        //设置中断 标志
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    //尝试将状态变为TERMINATED，上面已经分析过该方法的源码
    tryTerminate();
}

private void advanceRunState(int targetState) {
    for (;;) {
        //下面的方法执行的就是：
        //首先获取线程的ctl值，然后判断当前线程池的状态如果已经是SHUTDOWN，那么if条件第一个为真就直接返回
        //如果不是SHUTDOWN状态，就需要CAS的设置当前状态为SHUTDOWN
        int c = ctl.get();
        if (runStateAtLeast(c, targetState) ||
            //private static int ctlOf(int rs, int wc) { return rs | wc; }
            ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
            break;
    }
}

private void interruptIdleWorkers() {
    interruptIdleWorkers(false);
}
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();
    }
}

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess(); //还是进行权限检查
        advanceRunState(STOP); //设置线程池状态台STOP
        interruptWorkers(); //中断所有线程
        tasks = drainQueue(); //将任务队列中的任务移动到task中
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks; //返回tasks
}

private void interruptWorkers() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) //不需要判断当前线程是否在执行任务（即不需要调用w.tryLock方法），中断所有线程
            w.interruptIfStarted();
    } finally {
        mainLock.unlock();
    }
}

//调用该方法之后，会阻塞调用者线程，直到线程池状态为TERMINATED状态才会返回，或者等到超时时间到之后会返回
public boolean awaitTermination(long timeout, TimeUnit unit)
    throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //阻塞当前线程，（获取了Worker自己的锁），那么当前线程就不会再执行任务（因为获取不到锁）
        for (;;) {
            //当前线程池状态为TERMINATED状态，会返回true
            if (runStateAtLeast(ctl.get(), TERMINATED))
                return true;
            //超时时间到返回false
            if (nanos <= 0)
                return false;
            nanos = termination.awaitNanos(nanos);
        }
    } finally {
        mainLock.unlock();
    }
}