ThreadPoolExecutor 学习笔记

一、接口 ExecutorService 线程池服务接口

#####1. 提交任务

提交任务

	<T> Future<T> submit(Callable<T> task) 
			  提交一个返回值的任务用于执行，返回一个表示任务的未决结果的 Future。
	 Future<?>	submit(Runnable task) 
			  提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。
	<T> Future<T> submit(Runnable task, T result) 
			  提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。
			  
```			

#####2.执行所有任务或执行任一任务

``` java 执行所有任务或执行任一任务


	<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) 
			  执行给定的任务，当所有任务完成时，返回保持任务状态和结果的 Future 列表。
	<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) 
			  执行给定的任务，当所有任务完成或超时期满时（无论哪个首先发生），返回保持任务状态和结果的 Future 列表。
	<T> T invokeAny(Collection<? extends Callable<T>> tasks) 
			  执行给定的任务，如果某个任务已成功完成（也就是未抛出异常），则返回其结果。
	<T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) 
			  执行给定的任务，如果在给定的超时期满前某个任务已成功完成（也就是未抛出异常），则返回其结果。

#####3. 关闭线程池服务

关闭线程池服务

void	shutdown() 
	  启动一次顺序关闭，执行以前提交的任务，但不接受新任务。
List<Runnable>	shutdownNow() 
	  试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行的任务列表。

boolean	isShutdown() 
	  如果此执行程序已关闭，则返回 true。
boolean	isTerminated() 
	  如果关闭后所有任务都已完成，则返回 true。

boolean	awaitTermination(long timeout, TimeUnit unit) 
	  请求关闭、发生超时或者当前线程中断，无论哪一个首先发生之后，都将导致阻塞，直到所有任务完成执行。

#####关闭ExecutorService
下列方法分两个阶段关闭 ExecutorService。第一阶段调用 shutdown 拒绝传入任务，然后调用 shutdownNow（如有必要）取消所有遗留的任务：

shutdownAndAwaitTermination

void shutdownAndAwaitTermination(ExecutorService pool) {
  pool.shutdown(); // Disable new tasks from being submitted
  try {
 // Wait a while for existing tasks to terminate
 if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
   pool.shutdownNow(); // Cancel currently executing tasks
   // Wait a while for tasks to respond to being cancelled
   if (!pool.awaitTermination(60, TimeUnit.SECONDS))
	   System.err.println("Pool did not terminate");
 }
  } catch (InterruptedException ie) {
 // (Re-)Cancel if current thread also interrupted
 pool.shutdownNow();
 // Preserve interrupt status
 Thread.currentThread().interrupt();
  }
}

二、ThreadPoolExecutor 的整体介绍

核心和最大池大小
ThreadPoolExecutor 将根据 corePoolSize（参见 getCorePoolSize()）和 maximumPoolSize（参见 getMaximumPoolSize()）设置的边界自动调整池大小。

• 当新任务在方法 execute(java.lang.Runnable) 中提交时，如果运行的线程少于 corePoolSize，则创建新线程来处理请求，即使其他辅助线程是空闲的。
• 如果运行的线程多于 corePoolSize 而少于 maximumPoolSize，则仅当队列满时才创建新线程。如果设置的 corePoolSize 和 maximumPoolSize 相同，则创建了固定大小的线程池。
• 如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则允许池适应任意数量的并发任务。

按需构造
默认情况下，即使核心线程最初只是在新任务到达时才创建和启动的，也可以使用方法 prestartCoreThread() 或 prestartAllCoreThreads() 对其进行动态重写。如果构造带有非空队列的池，则可能希望预先启动线程。

保持活动时间
如果池中当前有多于 corePoolSize 的线程，则这些多出的线程在空闲时间超过 keepAliveTime 时将会终止（参见 getKeepAliveTime(java.util.concurrent.TimeUnit)）。这提供了当池处于非活动状态时减少资源消耗的方法。
如果池后来变得更为活动，则可以创建新的线程。也可以使用方法 setKeepAliveTime(long, java.util.concurrent.TimeUnit) 动态地更改此参数。
使用 Long.MAX_VALUE TimeUnit.NANOSECONDS 的值在关闭前有效地从以前的终止状态禁用空闲线程。默认情况下，保持活动策略只在有多于 corePoolSizeThreads 的线程时应用。
但是只要 keepAliveTime 值非 0，allowCoreThreadTimeOut(boolean) 方法也可将此超时策略应用于核心线程。

排队
所有 BlockingQueue 都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互：

1. 如果运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。
2. 如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。
3. 如果无法将请求加入队列，则创建新的线程，除非创建此线程超出 maximumPoolSize，在这种情况下，任务将被拒绝。

排队有三种通用策略：

1. 直接提交。工作队列的默认选项是 SynchronousQueue，它将任务直接提交给线程而不保持它们。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
2. 无界队列。使用无界队列（例如，不具有预定义容量的 LinkedBlockingQueue）将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
3. 有界队列。当使用有限的 maximumPoolSizes 时，有界队列（如 ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O 边界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。

被拒绝的任务
当 Executor 已经关闭，并且 Executor 将有限边界用于最大线程和工作队列容量，且已经饱和时，在方法 execute(java.lang.Runnable) 中提交的新任务将被拒绝。
在以上两种情况下，execute 方法都将调用其 RejectedExecutionHandler 的 RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) 方法。

下面提供了四种预定义的处理程序策略：

1. 在默认的 ThreadPoolExecutor.AbortPolicy 中，处理程序遭到拒绝将抛出运行时 RejectedExecutionException。
2. 在 ThreadPoolExecutor.CallerRunsPolicy 中，线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。
3. 在 ThreadPoolExecutor.DiscardPolicy 中，不能执行的任务将被删除。
   4。 在 ThreadPoolExecutor.DiscardOldestPolicy 中，如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）。
   定义和使用其他种类的 RejectedExecutionHandler 类也是可能的，但这样做需要非常小心，尤其是当策略仅用于特定容量或排队策略时。

钩子 (hook) 方法
此类提供 protected 可重写的 beforeExecute(java.lang.Thread, java.lang.Runnable) 和 afterExecute(java.lang.Runnable, java.lang.Throwable) 方法，这两种方法分别在执行每个任务之前和之后调用。
它们可用于操纵执行环境；例如，重新初始化 ThreadLocal、搜集统计信息或添加日志条目。此外，还可以重写方法 terminated() 来执行 Executor 完全终止后需要完成的所有特殊处理。
如果钩子 (hook) 或回调方法抛出异常，则内部辅助线程将依次失败并突然终止。

队列维护
方法 getQueue() 允许出于监控和调试目的而访问工作队列。强烈反对出于其他任何目的而使用此方法。remove(java.lang.Runnable) 和 purge() 这两种方法可用于在取消大量已排队任务时帮助进行存储回收。

线程池没有被程序里的其它对象引用，且没有剩余的线程会时，会自动 shutdown。如果希望确保回收取消引用的池（即使用户忘记调用 shutdown()），则必须安排未使用的线程最终终止：设置适当保持活动时间，使用 0 核心线程的下边界和/或设置 allowCoreThreadTimeOut(boolean)。

三、ThreadPoolExecutor内部实现

ThreadPoolExecutor

public class ThreadPoolExecutor extends AbstractExecutorService {

/**	
    * 线程池的主控制状态 ctl ，用 atomic integer 来表示两个概念的字段
    *   workerCount, 指示当前有效的线程数
    *   runState,    指示当前的运行状态 running, shutting down etc
    * 为了把两个字段打包成一个int, 限制workerCount  to
    * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
    * billion) otherwise representable. 
    *
    *
    * The runState provides the main lifecyle control, taking on values:
    *
    *   RUNNING:  Accept new tasks and process queued tasks
    *   SHUTDOWN: Don't accept new tasks, but process queued tasks
    *   STOP:     Don't accept new tasks, don't process queued tasks,
    *             and interrupt in-progress tasks
    *   TIDYING:  All tasks have terminated, workerCount is zero,
    *             the thread transitioning to state TIDYING
    *             will run the terminated() hook method
    *   TERMINATED: terminated() has completed
    *
    * 运行状态的次序很重要，可以用来进行比较。runState 随着时间单调递增的, but need not hit each state. The transitions are:
    *
    * RUNNING -> SHUTDOWN
    *    On invocation of shutdown(), perhaps implicitly in finalize()
    * (RUNNING or SHUTDOWN) -> STOP
    *    On invocation of shutdownNow()
    * SHUTDOWN -> TIDYING
    *    When both queue and pool are empty
    * STOP -> TIDYING
    *    When pool is empty
    * TIDYING -> TERMINATED
    *    When the terminated() hook method has completed
    *
    * Threads waiting in awaitTermination() will return when the
    * state reaches TERMINATED.
    *
    * 为什么从SHUTDOWN 转到 TIDYING 需要检测，因为 queue有可能从non-emtpy 转变 empty ,
    * but we can only terminate if, after seeing that it is empty, we see
    * that workerCount is 0 (which sometimes entails a recheck -- see
    * below).
    */
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; }
   
   private static boolean isRunning(int c) {
       return c < SHUTDOWN;
   }

   /**
    * Attempt to CAS-increment the workerCount field of ctl.
    */
   private boolean compareAndIncrementWorkerCount(int expect) {
       return ctl.compareAndSet(expect, expect + 1);
   }

   /**
    * Attempt to CAS-decrement the workerCount field of ctl.
    */
   private boolean compareAndDecrementWorkerCount(int expect) {
       return ctl.compareAndSet(expect, expect - 1);
   }

   /**
    * Decrements the workerCount field of ctl. This is called only on
    * abrupt termination of a thread (see processWorkerExit). Other
    * decrements are performed within getTask.
    */
   private void decrementWorkerCount() {
       do {} while (! compareAndDecrementWorkerCount(ctl.get()));
   }


// 保存任务的工作队列
   private final BlockingQueue<Runnable> workQueue;

// 保存当前正在执行线程的workers集合
private final HashSet<Worker> workers = new HashSet<Worker>();


// 主要的lock
private final ReentrantLock mainLock = new ReentrantLock();

private final Condition termination = mainLock.newCondition();



private volatile long keepAliveTime;

   private volatile boolean allowCoreThreadTimeOut;


//*************************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).
    */
   private final class Worker
       extends AbstractQueuedSynchronizer
       implements Runnable
   {

       /** 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) {
       	// 初始Worker状态为-1
           setState(-1); // inhibit interrupts until runWorker
           this.firstTask = firstTask;
           // 创建一个新的Thread线程对象
           this.thread = getThreadFactory().newThread(this);
       }

       /** Delegates main run loop to outer runWorker  */
       public void run() {
       	// 这里调用外部的runWorker!!!
           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) {
           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;
           // 当前Worker状态已启动，并且线程不为空
           if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
               try {
                   t.interrupt();
               } catch (SecurityException ignore) {
               }
           }
       }
   }


 
 private static final boolean ONLY_ONE = true;

    protected void terminated() { }


 /**
    * 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.
    */
   // 在每个 Worker 退时，会调用 tryTerminate 方法，所以这个方法会被调用多次
   final void tryTerminate() {
       for (;;) {
           int c = ctl.get();
           if (isRunning(c) ||
               runStateAtLeast(c, TIDYING) ||
               (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
               // 处理以下状态直接返回
               // 正在运行，
               // 已经>TIDYING 清理中
               // SHUTDOWN ,但 workQueue不为空
               return;  
               
            // 活动线程数不为0里，中断一个空闲Worker并return    
           if (workerCountOf(c) != 0) { // Eligible to terminate
               interruptIdleWorkers(ONLY_ONE);
               return;
           }

		// 当活动工作线程数为0时
           final ReentrantLock mainLock = this.mainLock;
           mainLock.lock();
           try {
           	// 设置为 TIDYING 状态
               if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                   try {
                       terminated();
                   } finally {
                   	 // 设置为 TERMINATED 终止状态
                       ctl.set(ctlOf(TERMINATED, 0));
                       // 发送 termination.signalAll 信号， 在 awaitTermination 方法中等待该信号
                       termination.signalAll();
                   }
                   return;
               }
           } finally {
               mainLock.unlock();
           }
           // else retry on failed CAS
       }
   }
   
   //******************************** interruptWorkers ***************************
       
   
   /**
    * Interrupts all threads, even if active. Ignores SecurityExceptions
    * (in which case some threads may remain uninterrupted).
    */
   // 强制中断所有Worker,即使Worker 正在运行中
   private void interruptWorkers() {
       final ReentrantLock mainLock = this.mainLock;
       mainLock.lock();
       try {
           for (Worker w : workers)
               w.interruptIfStarted();
       } finally {
           mainLock.unlock();
       }
   }

   /**
    * 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.
    */
   // 中断空闲Worker
   // @param onlyOne 为true，只来自 tryTerminate的调用。
       
   private void interruptIdleWorkers(boolean onlyOne) {
       final ReentrantLock mainLock = this.mainLock;
       mainLock.lock();
       try {
           for (Worker w : workers) {
               Thread t = w.thread;
               // 这里获取Worker的锁，说明当前线程处getTask 的wait状态
               // Worker 在运行时获取独占锁，在getTask释放锁，这里只中断空闲线程
               if (!t.isInterrupted() && w.tryLock()) {
                   try {
                       t.interrupt();
                   } catch (SecurityException ignore) {
                   } finally {
                       w.unlock();
                   }
               }
               
               // 只中断一个，返回
               if (onlyOne)
                   break;
           }
       } finally {
           mainLock.unlock();
       }
   }
   
   private void interruptIdleWorkers() {
       interruptIdleWorkers(false);
   }

   
   
    /**
    * Drains the task queue into a new list, normally using
    * drainTo. But if the queue is a DelayQueue or any other kind of
    * queue for which poll or drainTo may fail to remove some
    * elements, it deletes them one by one.
    */
   private List<Runnable> drainQueue() {
       BlockingQueue<Runnable> q = workQueue;
       List<Runnable> taskList = new ArrayList<Runnable>();
       q.drainTo(taskList);
       if (!q.isEmpty()) {
           for (Runnable r : q.toArray(new Runnable[0])) {
               if (q.remove(r))
                   taskList.add(r);
           }
       }
       return taskList;
   }
   
   /**
    * Executes the given task sometime in the future.  The task
    * may execute in a new thread or in an existing pooled thread.
    *
    * If the task cannot be submitted for execution, either because this
    * executor has been shutdown or because its capacity has been reached,
    * the task is handled by the current {@code RejectedExecutionHandler}.
    *
    * @param command the task to execute
    * @throws RejectedExecutionException at discretion of
    *         {@code RejectedExecutionHandler}, if the task
    *         cannot be accepted for execution
    * @throws NullPointerException if {@code command} is null
    */
   public void execute(Runnable command) {
   	// 参数校验
       if (command == null)
           throw new NullPointerException();
       /*
        * Proceed in 3 steps:
        *
        *          
        * 1. 如果当前运行的线程数比corePoolSize少, 会尝试启动一个新的线程来执行当前任务。
        * 调用addWorker会检查runState 和 workerCount，通过返回 flase 来避免 添加不必要的核心线程
        
        2. 如果一个任务可以成功放入队列，我们仍然需要double-check 来决定是否需要添加一个线程。（因为自从上次checking存在的线程，可能已经死亡）或者进行这个方法时，线程池关闭了。所以，我们需要recheck state 来判断是否需要在stopped时 roll back,或者启动一个新的线程。  
        *
        * 3. 如果任务不能放入队列，我们会尝试启动一个新的线程。如果启动失败，我们会知道我们正在关闭或saturated，来拒绝这个任务
        */
       int c = ctl.get();
       if (workerCountOf(c) < corePoolSize) {
       	// worker数小于corePoolSize启动新的Worker
           if (addWorker(command, true))
               return;// 启动新的core线程处理任务成功，直接返回
           c = ctl.get();
       }//启动core线程失败，继续后续处理
       
       if (isRunning(c) && workQueue.offer(command)) {
       	// 线程池正在运行，且成功添加到workQueue中
           int recheck = ctl.get();
           if (! isRunning(recheck) && remove(command))
           	 // recheck 当前状态不是运行，则移除并拒绝这个任务
               reject(command);
           else if (workerCountOf(recheck) == 0)
           	// 当前没有正在执行的Worker，则启动一个新Worker
               addWorker(null, false);
       }
       else if (!addWorker(command, false))
       	// 在队列放不下的情况下，启动新的Worker失败(线程数超过maxPoolSize)
           reject(command);
   }

   /*
    * Methods for creating, running and cleaning up after workers
    */

   /**
    * Checks if a new worker can be added with respect to current
    * pool state and the given bound (either core or maximum). If so,
    * the worker count is adjusted accordingly, and, if possible, a
    * new worker is created and started, running firstTask as its
    * first task. This method returns false if the pool is stopped or
    * eligible to shut down. It also returns false if the thread
    * factory fails to create a thread when asked.  If the thread
    * creation fails, either due to the thread factory returning
    * null, or due to an exception (typically OutOfMemoryError in
    * Thread#start), we roll back cleanly.
    *
    * @param firstTask the task the new thread should run first (or
    * null if none). Workers are created with an initial first task
    * (in method execute()) to bypass queuing when there are fewer
    * than corePoolSize threads (in which case we always start one),
    * or when the queue is full (in which case we must bypass queue).
    * Initially idle threads are usually created via
    * prestartCoreThread or to replace other dying workers.
    *
    * @param core if true use corePoolSize as bound, else
    * maximumPoolSize. (A boolean indicator is used here rather than a
    * value to ensure reads of fresh values after checking other pool
    * state).
    * @return true if successful
    */
   private boolean addWorker(Runnable firstTask, boolean core) {
       retry:
       for (;;) {
           int c = ctl.get();
           int rs = runStateOf(c);

           // Check if queue empty only if necessary.
           // runState >= SHUTDOWN 
           // 
           if (rs >= SHUTDOWN &&
               ! (rs == SHUTDOWN &&
                  firstTask == null &&
                  ! workQueue.isEmpty()))
               return false;

           for (;;) {
               int wc = workerCountOf(c);
               if (wc >= CAPACITY ||
                   wc >= (core ? corePoolSize : maximumPoolSize))
                   // 在于线程池的容量返回
                   // 大于corePoolSize 或 maximumPoolSize 返回 
                   return false;
               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
           }
       }

       boolean workerStarted = false;
       boolean workerAdded = false;
       Worker w = null;
       try {
           final ReentrantLock mainLock = this.mainLock;
           w = new Worker(firstTask);//new一个新的Worker
           final Thread t = w.thread;
           if (t != null) {
               mainLock.lock();
               try {
                   // Recheck while holding lock.
                   // Back out on ThreadFactory failure or if
                   // shut down before lock acquired.
                   int c = ctl.get();
                   int rs = runStateOf(c);

                   if (rs < SHUTDOWN ||
                       (rs == SHUTDOWN && firstTask == null)) {
                       if (t.isAlive()) // precheck that t is startable
                           throw new IllegalThreadStateException();
                       workers.add(w);// 添加到workers集合
                       int s = workers.size();
                       if (s > largestPoolSize)//记录线程最大时的线程数
                           largestPoolSize = s;
                       workerAdded = true;
                   }
               } finally {
                   mainLock.unlock();
               }
               if (workerAdded) {
                   t.start();//启动Worker线程
                   workerStarted = true;
               }
           }
       } finally {
       	// 添加失败，进行worker 清理工作
           if (! workerStarted)
               addWorkerFailed(w);
       }
       return workerStarted;
   }

   /**
    * Rolls back the worker thread creation.
    * - removes worker from workers, if present
    * - decrements worker count
    * - rechecks for termination, in case the existence of this
    *   worker was holding up termination
    */
   private void addWorkerFailed(Worker w) {
       final ReentrantLock mainLock = this.mainLock;
       mainLock.lock();
       try {
           if (w != null)
               workers.remove(w);
           decrementWorkerCount();
           tryTerminate();
       } finally {
           mainLock.unlock();
       }
   }

   /**
    * 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;
           workers.remove(w);
       } finally {
           mainLock.unlock();
       }

	// Worker退出时，尝试终止整个线程池
       tryTerminate();

       int c = ctl.get();
       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);
       }
   }

   /**
    * 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.
    *
    * @return task, or null if the worker must exit, in which case
    *         workerCount is decremented
    */
   private Runnable getTask() {
       boolean timedOut = false; // Did the last poll() time out?

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

		// 如果正在 SHUTDOWN 并且
           // Check if queue empty only if necessary.
           if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
               decrementWorkerCount();
               return null;
           }

		// workers 是否需要判断超时退出
           boolean timed;      // Are workers subject to culling?

           for (;;) {
           	 // 判断当前线程数是否需要超时退出
               int wc = workerCountOf(c);
               timed = allowCoreThreadTimeOut || wc > corePoolSize;
				
			 // 当前线程不需要超时break
               if (wc <= maximumPoolSize && ! (timedOut && timed))
                   break;
               
               // 递减WorkerCount计数，直接返回null    
               if (compareAndDecrementWorkerCount(c))
                   return null;
               c = ctl.get();  // Re-read ctl
               if (runStateOf(c) != rs)
                   continue retry;
               // else CAS failed due to workerCount change; retry inner loop
           }

           try {
               Runnable r = timed ?
                   workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                   workQueue.take();
               if (r != null)
                   return r;
               timedOut = true;
           } catch (InterruptedException retry) {
               timedOut = false;
           }
       }
   }

   /**
    * 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
    * clearInterruptsForTaskRun called to 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
    */
   final void runWorker(Worker w) {
       Thread wt = Thread.currentThread();// 获取当前Worker的执行线程
       Runnable task = w.firstTask;
       w.firstTask = null;
       w.unlock(); // allow interrupts 允许被中断
       boolean completedAbruptly = true;
       try {
       	// 循环获取任务, 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();//中断worker线程
                   
               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);
       }
   }
   
   

//**************************** shutdown ***************************************

   
   
   /**
    * Initiates an orderly shutdown in which previously submitted
    * tasks are executed, but no new tasks will be accepted.
    * Invocation has no additional effect if already shut down.
    *
    * <p>This method does not wait for previously submitted tasks to
    * complete execution.  Use {@link #awaitTermination awaitTermination}
    * to do that.
    *
    * @throws SecurityException {@inheritDoc}
    */
   public void shutdown() {
       final ReentrantLock mainLock = this.mainLock;
       mainLock.lock();
       try {
           checkShutdownAccess();
           // 设置运行状态为SHUTDOWN
           advanceRunState(SHUTDOWN);
           interruptIdleWorkers();//中断所有线程
           onShutdown(); // hook for ScheduledThreadPoolExecutor
       } finally {
           mainLock.unlock();
       }
       
       // shutdown后，尝试终止线程池
       tryTerminate();
   }

   /**
    * 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();
           advanceRunState(STOP);
           interruptWorkers();
           tasks = drainQueue();
       } finally {
           mainLock.unlock();
       }
       
       // shutdown后，尝试终止线程池
       tryTerminate();
       return tasks;
   }

   public boolean isShutdown() {
       return ! isRunning(ctl.get());
   }

   /**
    * Returns true if this executor is in the process of terminating
    * after {@link #shutdown} or {@link #shutdownNow} but has not
    * completely terminated.  This method may be useful for
    * debugging. A return of {@code true} reported a sufficient
    * period after shutdown may indicate that submitted tasks have
    * ignored or suppressed interruption, causing this executor not
    * to properly terminate.
    *
    * @return true if terminating but not yet terminated
    */
   public boolean isTerminating() {
       int c = ctl.get();
       return ! isRunning(c) && runStateLessThan(c, TERMINATED);
   }

   public boolean isTerminated() {
       return runStateAtLeast(ctl.get(), TERMINATED);
   }

   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;
               // 等待 termination
               nanos = termination.awaitNanos(nanos);
           }
       } finally {
           mainLock.unlock();
       }
   }

   /**
    * Invokes {@code shutdown} when this executor is no longer
    * referenced and it has no threads.
    */
   protected void finalize() {
       shutdown();
   }
   
   
   
//******************************预启动线程 *************************

   /**
    * Starts a core thread, causing it to idly wait for work. This
    * overrides the default policy of starting core threads only when
    * new tasks are executed. This method will return {@code false}
    * if all core threads have already been started.
    *
    * @return {@code true} if a thread was started
    */
   public boolean prestartCoreThread() {
       return workerCountOf(ctl.get()) < corePoolSize &&
           addWorker(null, true);
   }

   /**
    * Same as prestartCoreThread except arranges that at least one
    * thread is started even if corePoolSize is 0.
    */
   void ensurePrestart() {
       int wc = workerCountOf(ctl.get());
       if (wc < corePoolSize)
           addWorker(null, true);
       else if (wc == 0)
           addWorker(null, false);
   }

   /**
    * Starts all core threads, causing them to idly wait for work. This
    * overrides the default policy of starting core threads only when
    * new tasks are executed.
    *
    * @return the number of threads started
    */
   public int prestartAllCoreThreads() {
       int n = 0;
       while (addWorker(null, true))
           ++n;
       return n;
   }

   /**
    * Removes this task from the executor's internal queue if it is
    * present, thus causing it not to be run if it has not already
    * started.
    *
    * <p> This method may be useful as one part of a cancellation
    * scheme.  It may fail to remove tasks that have been converted
    * into other forms before being placed on the internal queue. For
    * example, a task entered using {@code submit} might be
    * converted into a form that maintains {@code Future} status.
    * However, in such cases, method {@link #purge} may be used to
    * remove those Futures that have been cancelled.
    *
    * @param task the task to remove
    * @return true if the task was removed
    */
   public boolean remove(Runnable task) {
       boolean removed = workQueue.remove(task);
       // 在 SHUTDOWN 并且队列为空时，尝试终止 tryTerminate
       tryTerminate(); // In case SHUTDOWN and now empty
       return removed;
   }

   /**
    * Tries to remove from the work queue all {@link Future}
    * tasks that have been cancelled. This method can be useful as a
    * storage reclamation operation, that has no other impact on
    * functionality. Cancelled tasks are never executed, but may
    * accumulate in work queues until worker threads can actively
    * remove them. Invoking this method instead tries to remove them now.
    * However, this method may fail to remove tasks in
    * the presence of interference by other threads.
    */
   public void purge() {
       final BlockingQueue<Runnable> q = workQueue;
       try {
           Iterator<Runnable> it = q.iterator();
           while (it.hasNext()) {
               Runnable r = it.next();
               if (r instanceof Future<?> && ((Future<?>)r).isCancelled())
                   it.remove();
           }
       } catch (ConcurrentModificationException fallThrough) {
           // Take slow path if we encounter interference during traversal.
           // Make copy for traversal and call remove for cancelled entries.
           // The slow path is more likely to be O(N*N).
           for (Object r : q.toArray())
               if (r instanceof Future<?> && ((Future<?>)r).isCancelled())
                   q.remove(r);
       }

	// 在 SHUTDOWN 并且队列为空时，尝试终止 tryTerminate
       tryTerminate(); // In case SHUTDOWN and now empty
   }

五、 ExecutorCompletionService

使用提供的 Executor 来执行任务的 CompletionService。此类将把 提交任务完成时的结果 放置 在可使用 take 访问的队列上。该类非常轻便，适合于在执行几组任务时临时使用。

(把执行任务完成的结果提交到一个队列中。)