java 线程池（1）

问题 :

• 线程池中的 coreSize 和 maxSize 的作用分别是什么？
• 未执行的线程池存在在哪种数据类型，为什么使用这种类型的数据结构

ThreadPoolExecutor概述

        ThreadPoolExecutor 下文简称 TPE ，我们使用它都是从Executror 这个类中的方法 ：

    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 static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

    //ScheduledExecutorService
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }


    public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService

       Executror 的方法名很明显地说明了创建的对象的用途，我们也可以看到它们实际的都是走到了TLE构造函数，只是传入的参数不同。

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }


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

          由此可以推断通过配置TLE的各个参数，实现不同的功能。

ThreadPoolExecutor重要知识点

      官方文档中有详细介绍，细节请看官方文档

Core and maximum pool sizes 核心线程数 和 线程池最大线程数

        这两个在构造方法中需要指定，核心线程数和最大线程池线程数，很好理解，就像两条上限线，当来任务时没达到核心线程数，那么就开启一条新线程去执行，要是达到核心数量了，怎么办，任务入列，要是超过了我设定的最大线程数量，那么不再接受任务。这两个值可以动态设置：   setCorePoolSize(int) 和setMaximumPoolSize(int).

        core 线程数，当空闲的时候依然会存活，除非设置了存活时间。

On-demand construction

         默认情况下，甚至核心线程最初只在新任务到达时创建并启动，但可以使用方法prestartCoreThread（）或prestartAllCoreThreads（）动态覆盖。 如果使用非空队列构造池，则可能需要预启动线程。

Creating new threads 创建新线程

         使用ThreadFactory创建新线程。 如果没有另外指定，则使用Executors.defaultThreadFactory（），它将所有线程创建在同一个ThreadGroup中，并具有相同的NORM_PRIORITY优先级和非守护进程状态。 通过提供不同的ThreadFactory，您可以更改线程的名称，线程组，优先级，守护程序状态等。如果ThreadFactory在通过从newThread返回null请求时无法创建线程，则执行程序将继续，但可能无法 执行任何任务。

       线程应该拥有“modifyThread”RuntimePermission。 如果使用池的工作线程或其他线程不具有此权限，则服务可能会降级：配置更改可能不会及时生效，并且关闭池可能保持可以终止但未完成的状态。（取至官方文档谷歌翻译）

 

Keep-alive times 存活时间

          前面说到  Core and maximum pool sizes  就像两个上限线，当超过了核心线程数后，任务开始执行完成，那么线程就空闲了，此时要是空闲达到了 Keep-alive times 这个设定值，那么线程就会被回收。使用Long.MAX_VALUE类型的TimeUnit.NANOSECONDS有效地禁止空闲线程在关闭之前终止。

      the keep-alive policy applies only when there are more than corePoolSize threads. But method allowCoreThreadTimeOut(boolean) can be used to apply this time-out policy to core threads as well, so long as the keepAliveTime value is non-zero

任务处理原则      

Queuing

         队列的使用和线程池的线程数有关。

• 如果现有线程少于 corePoolSize 线程数量，尝试开启一条新的线程执行（能执行任务就不排队解决的原则）
• 达到或超过 corePoolSize ,任务入列
• 任务队列满了，开新线程直到 maximumPoolSize,当线程数达到 maximumPoolSize,拒绝请求。

         有以下几种队列策略 ：

• Direct handoffs.（直接交付） A good default choice for a work queue is a SynchronousQueue that hands off tasks to threads without otherwise holding them. Here, an attempt to queue a task will fail if no threads are immediately available to run it, so a new thread will be constructed. This policy avoids lockups when handling sets of requests that might have internal dependencies. Direct handoffs generally require unbounded maximumPoolSizes to avoid rejection of new submitted tasks. This in turn admits the possibility of unbounded thread growth when commands continue to arrive on average faster than they can be processed.

• Unbounded queues.（无界队列） Using an unbounded queue (for example a LinkedBlockingQueue without a predefined capacity) will cause new tasks to wait in the queue when all corePoolSize threads are busy. Thus, no more than corePoolSize threads will ever be created. (And the value of the maximumPoolSize therefore doesn't have any effect.) This may be appropriate when each task is completely independent of others, so tasks cannot affect each others execution; for example, in a web page server. While this style of queuing can be useful in smoothing out transient bursts of requests, it admits the possibility of unbounded work queue growth when commands continue to arrive on average faster than they can be processed.

• Bounded queues.（有界队列） A bounded queue (for example, an ArrayBlockingQueue) helps prevent resource exhaustion when used with finite maximumPoolSizes, but can be more difficult to tune and control. Queue sizes and maximum pool sizes may be traded off for each other: Using large queues and small pools minimizes CPU usage, OS resources, and context-switching overhead, but can lead to artificially low throughput. If tasks frequently block (for example if they are I/O bound), a system may be able to schedule time for more threads than you otherwise allow. Use of small queues generally requires larger pool sizes, which keeps CPUs busier but may encounter unacceptable scheduling overhead, which also decreases throughput.

         实际上不同应用场景的线程池为适应不同需求，使用了不同的 queue 策略。

状态标识

          状态控制有一个变量 ctl ，它又两部分组成，

 

• （后29位）workCount : 指示当前有效的线程数量，但是并不能代表当前存活的活跃的线程数。
• （高3位）runStatus : 指示线程池本身的状态

          下面是状态变量的定义

    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private static final int COUNT_BITS = Integer.SIZE - 3;
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

    // runState is stored in the high-order bits
    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
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    private static int ctlOf(int rs, int wc) { return rs | wc; }

    /*
     * Bit field accessors that don't require unpacking ctl.
     * These depend on the bit layout and on workerCount being never negative.
     */

    private static boolean runStateLessThan(int c, int s) {
        return c < s;
    }

    private static boolean runStateAtLeast(int c, int s) {
        return c >= s;
    }

    private static boolean isRunning(int c) {
        return c < SHUTDOWN;
    }

          runStatus 提供这几种状态 ：

• RUNNING : 运行
• SHUTDOWN : 停止接收新任务，执行已经入队的任务。
• STOP :  不再接受新任务，不执行已经入列的任务，中断正在执行的任务。
• TIDING : 停止接收新任务，所有任务被暂停，workCount 变为 0 ，线程状态变为 TIDYING.将执行hook 方法 terminated.
• TERMINATED : terminated方法完成。

拒绝任务策略

    当线程池关闭或是达到了消费队列中满了的情况下，拒绝任务，线程池提供了这几种方式的拒绝策略

• 中断策略AbortPolicy (默认) ： 丢弃并抛出异常
• 调用者调用策略CallerRunsPolicy : 当前线程执行任务，并告知线程以低速率接收新的任务
• 抛弃策略DiscardPolicy  :  丢弃策略
• 抛弃最老策略DiscardOldestPolicy : 丢弃进消费队列最长时间的任务，当线程池没有关闭，则重新尝试执行任务
  

ThreadPoolExecutor源码分析

         看 execute 方法。

    　//可以创建线程处理就处理，不行就入列，入列也失败，拒绝！
     public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        int c = ctl.get();
        //未达到 corePoolSize 增加新线程执行
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //增加线程失败，或者有可能 worker的数量大于等于 core ，或是大于 maxSize ,任务入列 
        if (isRunning(c) && workQueue.offer(command)) {
            //注意 ：此时任务已成功入列！！！
            int recheck = ctl.get();
            //再次检查，要是 此时是 SHUTDOWN 状态（线程池关闭），那么移除这个任务，同时拒绝这个请求
            if (! isRunning(recheck) && remove(command))
                reject(command);
            //非running 状态 ，同时 workerCounter 为 0 （可能线程池里的任务都执行完了），那么新建一个线程，而不去处理，为什么要这样呢？
            //因为任务此时在队列中了，创建线程后，自动会去获取任务并处理
            //要是都不是就退出这个方法，此时任务在队列中等待被处理
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //线程池 shut down 或是队列满了，再次新建一个线程执行，但是这次的线程数的判断边界是　maxSize ,即是
        // addWorker的第二个参数来指定 
        else if (!addWorker(command, false))
            reject(command);
    }

         假设目前还未达到 core 的数量，那么进入 addWorker 方法 。

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

            // Check if queue empty only if necessary.  
            // 线程池满足如下条件中的任意一种时, 就会直接结束该方法, 并且返回 false
           // 表示没有创建新线程, 新提交的任务也没有被执行.
           // 1 .处于 STOP, TYDING 或 TERMINATD 状态
           // 2 .处于 SHUTDOWN 状态, 并且参数 firstTask != null
          // 3 .处于 SHUTDOWN 状态, firstTask == null 且阻塞队列 workQueue为空
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
            for (;;) {
                int wc = workerCountOf(c);
                //此处可以看到 第二个参数，core 是用来 选定线程数边界的
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //自旋增加 c , ctl的值 ,成功就 break 退出
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    //说明有人抢了，
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
                // 或者 CAS 失败是因为 workerCount 改变，继续loop 
            }
        }

        //下面是增加一个线程的操作，创建 worker ，上锁，再次判断,创建成功后线程开始执行。
        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();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    // recheck 当获得锁的时候 ，退出因为 ThreadFactory 失败或是 在获得锁之前 线程池 shut down 
                    int rs = runStateOf(ctl.get());

                    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) {
                    //走到这里线程肯定是创建了，并且线程池一定是正常的。
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

          addWoker 中创建了一个Worker，我们先看一下构造方法，再慢慢分析它。

        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

public interface ThreadFactory {

    /**
     * Constructs a new {@code Thread}.  Implementations may also initialize
     * priority, name, daemon status, {@code ThreadGroup}, etc.
     *
     * @param r a runnable to be executed by new thread instance
     * @return constructed thread, or {@code null} if the request to
     *         create a thread is rejected
     */
    Thread newThread(Runnable r);
}

          可以看到 ThreadFactory 实际就是创建线程的方法，同时传入一个 Runnable ， worker 里面传入了一个this ，我们赶紧看一下worker的定义。

    private final class Worker  extends AbstractQueuedSynchronizer  implements Runnable

        意图很明显，就是worker 本身带有一个任务（可以为NULL），让刚创建的线程去执行这个任务。下面看一下它到底执行了什么？

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



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


     /**
     * Main worker run loop.  Repeatedly gets tasks from queue and
     * executes them, while coping with a number of issues:
     *
     * 1. We may start out with an initial task, in which case we
     * don't need to get the first one. Otherwise, as long as pool is
     * running, we get tasks from getTask. If it returns null then the
     * worker exits due to changed pool state or configuration
     * parameters.  Other exits result from exception throws in
     * external code, in which case completedAbruptly holds, which
     * usually leads processWorkerExit to replace this thread.
     *
     * 2. Before running any task, the lock is acquired to prevent
     * other pool interrupts while the task is executing, and then we
     * ensure that unless pool is stopping, this thread does not have
     * its interrupt set.
     *
     * 3. Each task run is preceded by a call to beforeExecute, which
     * might throw an exception, in which case we cause thread to die
     * (breaking loop with completedAbruptly true) without processing
     * the task.
     *
     * 4. Assuming beforeExecute completes normally, we run the task,
     * gathering any of its thrown exceptions to send to afterExecute.
     * We separately handle RuntimeException, Error (both of which the
     * specs guarantee that we trap) and arbitrary Throwables.
     * Because we cannot rethrow Throwables within Runnable.run, we
     * wrap them within Errors on the way out (to the thread's
     * UncaughtExceptionHandler).  Any thrown exception also
     * conservatively causes thread to die.
     *
     * 5. After task.run completes, we call afterExecute, which may
     * also throw an exception, which will also cause thread to
     * die. According to JLS Sec 14.20, this exception is the one that
     * will be in effect even if task.run throws.
     *
     * The net effect of the exception mechanics is that afterExecute
     * and the thread's UncaughtExceptionHandler have as accurate
     * information as we can provide about any problems encountered by
     * user code.
     *
     * @param w the worker
     *
     *
     *  这里可以看到执行完任务后，就会阻塞在 getTask ，而线程没有被回收
     *  除非getTask 返回 null ，所有我们利用 一些调用满足 getTask 返回 null
     *  例如 ： 超时设置
     *
     *
     **/
    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts  允许中断，释放锁（为了下面抢任务）
        boolean completedAbruptly = true;
        try {
            //抢任务，抢到就加锁，即是 firstTask 为null 时才会去 getTask 
            while (task != null || (task = getTask()) != null) {
                // 加锁，防止线程被其他线程中断 
                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.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();
                }
            }
            //来到这里，1.task == null  
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }





    /**
     * Performs blocking or timed wait for a task, depending on
     * current configuration settings, or returns null if this worker
     * must exit because of any of:
     * 1. There are more than maximumPoolSize workers (due to
     *    a call to setMaximumPoolSize).
     * 2. The pool is stopped.
     * 3. The pool is shutdown and the queue is empty.
     * 4. This worker timed out waiting for a task, and timed-out
     *    workers are subject to termination (that is,
     *    {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
     *    both before and after the timed wait, and if the queue is
     *    non-empty, this worker is not the last thread in the pool.
     *
     * @return task, or null if the worker must exit, in which case
     *         workerCount is decremented
     *
     *
     *
     * 上面的注解是 4 种返回 null 的情况， 其中第一种的原因有没有可能是因为其他线程创建线程导致的呢？
     *  worker 必须退出，顺便数量在这里减少一
     *
     */
    private Runnable getTask() {
        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.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            int wc = workerCountOf(c);

            // Are workers subject to culling?
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

            //自旋失败后继续 loop 直到 成功 
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                //这里自旋
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }

            try {
                //前面  boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
                //可以知道 keepAliveTime 区别于 上面两个条件  1. allowCoreThreadTimeOut  2.wc > corePoolSize 
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            // 在这里会捕获中断异常！！这里很重要，提供了假如是调用了 shutDown() 方法，线程可以退出的出口
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

        getTask 里的 timed 这个变量判断条件一个是 allowCoreThreadTimeOut  和 wc > corePoolSize ，下面有个例子，corePoolSize 为 0 ，而MaxSize 为 Integer.MAX_VALUE，keepAliveTime 为60L，我们由此可以知道当一分钟后，线程没有获取到任务，那么getTask 就会返回null ，退出后该线程就会被回收。

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

        思路就是从任务队列中不断拿取任务（无任务状态下，线程阻塞），然后执行该任务。有个很奇怪的地方，runworker 中在获取任务前释放了锁，在获取到任务后再次获取锁。为什么呢？在 worker 这个类前有注解。

   /**
     * Class Worker mainly maintains interrupt control state for
     * threads running tasks, along with other minor bookkeeping.
     * This class opportunistically extends AbstractQueuedSynchronizer
     * to simplify acquiring and releasing a lock surrounding each
     * task execution.  This protects against interrupts that are
     * intended to wake up a worker thread waiting for a task from
     * instead interrupting a task being run.  We implement a simple
     * non-reentrant mutual exclusion lock rather than use
     * ReentrantLock because we do not want worker tasks to be able to
     * reacquire the lock when they invoke pool control methods like
     * setCorePoolSize.  Additionally, to suppress interrupts until
     * the thread actually starts running tasks, we initialize lock
     * state to a negative value, and clear it upon start (in
     * runWorker).
     */

         worker 使用了“中断控制状态”来维护线程运行，该类继承 AbstractQueueSynchronizer ，它的作用是当这个线程在执行任务时不被其他线程中断，而是让其他线程等待被唤醒。同时，该类使用无重入的独占互斥锁而不是 ReentrantLock ，因为我们不想在调用setCorePoolSize 重入该锁。

         现在线程池就在愉快地执行任务了，假如我这时候停止线程池。

    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //检查是否可以 shutdown 
            checkShutdownAccess();
            //设置为 SHUTDOWN 
            advanceRunState(SHUTDOWN);
            //中断所有worker 
            interruptIdleWorkers();
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        //终结这个线程池
        tryTerminate();
    }



    /**
     * If there is a security manager, makes sure caller has
     * permission to shut down threads in general (see shutdownPerm).
     * If this passes, additionally makes sure the caller is allowed
     * to interrupt each worker thread. This might not be true even if
     * first check passed, if the SecurityManager treats some threads
     * specially.
     */
    private void checkShutdownAccess() {
        SecurityManager security = System.getSecurityManager();
        if (security != null) {
            security.checkPermission(shutdownPerm);
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                for (Worker w : workers)
                    security.checkAccess(w.thread);
            } finally {
                mainLock.unlock();
            }
        }
    }



    /**
     * Transitions runState to given target, or leaves it alone if
     * already at least the given target.
     *
     * @param targetState the desired state, either SHUTDOWN or STOP
     *        (but not TIDYING or TERMINATED -- use tryTerminate for that)
     */
    private void advanceRunState(int targetState) {
        for (;;) {
            int c = ctl.get();
            if (runStateAtLeast(c, targetState) ||
                ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
                break;
        }
    }


    private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
    }



    /**
     * Interrupts threads that might be waiting for tasks (as
     * indicated by not being locked) so they can check for
     * termination or configuration changes. Ignores
     * SecurityExceptions (in which case some threads may remain
     * uninterrupted).
     *
     * @param onlyOne If true, interrupt at most one worker. This is
     * called only from tryTerminate when termination is otherwise
     * enabled but there are still other workers.  In this case, at
     * most one waiting worker is interrupted to propagate shutdown
     * signals in case all threads are currently waiting.
     * Interrupting any arbitrary thread ensures that newly arriving
     * workers since shutdown began will also eventually exit.
     * To guarantee eventual termination, it suffices to always
     * interrupt only one idle worker, but shutdown() interrupts all
     * idle workers so that redundant workers exit promptly, not
     * waiting for a straggler task to finish.
     *
     *
     *  从方法名可以看出 ： 中断空闲 workers
     *  从下面的代码也可以看到要是传过来的参数是 false
     *  那么所有线程将被中断  （shutDown()方法有运用到）
     *
     */
    private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {

            for (Worker w : workers) {
                Thread t = w.thread;
                //这里 worker 的 tryLock 需要注意一下，这里要是 worker 正在执行任务(这就解释了为什么在runWorker 方法中，worker要加锁了)，
                // 那么 tryLock 返回 false,
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {
                        //注意这里忽略了这个 exception 
                        //所以文档中指出了有可能某些线程依旧会保持为非中断状态 
                    } finally {
                        w.unlock();
                    }
                }
                if (onlyOne)
                    break;
            }
        } finally {
            mainLock.unlock();
        }
    }


    /**
     * Performs any further cleanup following run state transition on
     * invocation of shutdown.  A no-op here, but used by
     * ScheduledThreadPoolExecutor to cancel delayed tasks.
     */
    void onShutdown() {
    }




    /**
     * Attempts to stop all actively executing tasks, halts the
     * processing of waiting tasks, and returns a list of the tasks
     * that were awaiting execution. These tasks are drained (removed)
     * from the task queue upon return from this method.
     *
     * <p>This method does not wait for actively executing tasks to
     * terminate.  Use {@link #awaitTermination awaitTermination} to
     * do that.
     *
     * <p>There are no guarantees beyond best-effort attempts to stop
     * processing actively executing tasks.  This implementation
     * cancels tasks via {@link Thread#interrupt}, so any task that
     * fails to respond to interrupts may never terminate.
     *
     * @throws SecurityException {@inheritDoc}
     */
    public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            // STOP 终于在这里发挥了作用！！S:D
            advanceRunState(STOP);
            interruptWorkers();
            //remove task 从队列中
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
        return tasks;
    }

         看一下 tryTerminate 方法。

    /**
     * Transitions to TERMINATED state if either (SHUTDOWN and pool
     * and queue empty) or (STOP and pool empty).  If otherwise
     * eligible to terminate but workerCount is nonzero, interrupts an
     * idle worker to ensure that shutdown signals propagate. This
     * method must be called following any action that might make
     * termination possible -- reducing worker count or removing tasks
     * from the queue during shutdown. The method is non-private to
     * allow access from ScheduledThreadPoolExecutor.
     *
     *
     * 如果满足其中一个条件 ：
     *    1. SHUTDOWN 并且 workerCount为 0 并且 队列为空
     *    2. STOP 并且 workerCount为 0
     *    那么将状态转化为 TERMINATED ；
     *
     *  如果  workerCount(c)!=0 ,那么调用  interruptIdleWorkers(true); 然后就return
     *
     *
     *  所以我们可以知道在线程池正常的状态的下必定直接 return
     *
     */
    final void tryTerminate() {
        for (;;) {
            int c = ctl.get();
            if (isRunning(c) ||
                runStateAtLeast(c, TIDYING) ||
                (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                return;
            //具备条件时执行下面
            if (workerCountOf(c) != 0) { // Eligible to terminate
                // 传入一个参数  true ，表示只中断一个，这是因为，每个线程当自己没任务时，肯定
                interruptIdleWorkers(ONLY_ONE);
                return;
            }

            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
                        //看到了吗，介绍线程池状态时，讲到要是是 TIDYING 状态时，会调用这个钩子方法
                        terminated();
                    } finally {
                        ctl.set(ctlOf(TERMINATED, 0));
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                mainLock.unlock();
            }
            // else retry on failed CAS
        }
    }

       从上面我们看到要是调用了 shutDown() 或是 shutDownNow ()那么我们正阻塞在 getTask()方法的线程就会收到中断异常，于是就会getTask就会返回 null 。那么我们继续看一下线程继续向下执行的逻辑。

    /**
     * Performs cleanup and bookkeeping for a dying worker. Called
     * only from worker threads. Unless completedAbruptly is set,
     * assumes that workerCount has already been adjusted to account
     * for exit.  This method removes thread from worker set, and
     * possibly terminates the pool or replaces the worker if either
     * it exited due to user task exception or if fewer than
     * corePoolSize workers are running or queue is non-empty but
     * there are no workers.
     *
     * @param w the worker
     * @param completedAbruptly if the worker died due to user exception
     */
    private void processWorkerExit(Worker w, boolean completedAbruptly) {
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
            decrementWorkerCount();

        //加锁执行
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            completedTaskCount += w.completedTasks;
            //移除 worker 
            workers.remove(w);
        } finally {
            mainLock.unlock();
        }

        tryTerminate();

        int c = ctl.get();
        // replacement 的意思在这里，是上面已经移除了一个 worker , 这里调用 addWorker 再补充 
        if (runStateLessThan(c, STOP)) {
            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);
        }
    }

线程池线程数量的选定

• 如果是CPU密集型应用，则线程池大小设置为N+1
• 如果是IO密集型应用，则线程池大小设置为2N+1

CPU密集型

       一些进程绝大多数时间在计算上，称为计算密集型（CPU密集型）computer-bound。一些大量循环的代码（例如：图片处理、视频编码、人工智能等）就是CPU密集型。

I/O密集型

       有一些进程则在input 和output上花费了大多时间，称为I/O密集型，I/O-bound。比如搜索引擎蜘蛛大多时间是在等待相应这种就属于I/O密集型。

          当然这也是大家常用的估算方法，阅读以下文章帮助了解 。

总结

•   线程池主要的使用到了一个 workQueue ，这是一个BlockQueue (阻塞队列) ，用于存放还没执行的任务。
•   参数里面 coreSize 主要的作用是控制创建线程的数量，maxSize 主要的作用是是否还存放还没执行的任务，还是执行拒绝策略

参考资料 ：

• Java 线程池 ThreadPoolExecutor 源码分析
• CPU操作类型