核心接口

Executor

Executor只有一个execute方法抽象为对任务（Runnable接口）的执行

void execute(Runnable command);

ExecutorService

ExecutorService接口在Executor的基础上提供了对任务执行的生命周期的管理，主要是submit和shutdown方法

AbstractExecutorService

AbstractExecutorService对ExecutorService一些方法做了默认的实现，主要是submit和invoke方法。

真正的任务执行 的Executor接口execute方法是由子类实现，就是ThreadPoolExecutor，它实现了基于线程池的任务执行框架。

源码分析

ThreadPoolExecutor有5 个内部类
Worker、CallerRunsPolicy、AbortPolicy、DiscardPolicy、DiscardOldestPolicy
分两种类型：policy（策略） worker（工作）
policy ：超出线程范围和队列容量时添加（task）任务执行的策略。
四种策略 ：
AbortPolicy ：抛出异常，默认
CallerRunsPolicy ：不使用线程池执行
DiscardPolicy ：直接丢弃任务
DiscardOldestPolicy ：丢弃队列中最旧的任务
设置策略有两种方式：

RejectedExecutionHandler handler = new 
ThreadPoolExecutor.CallerRunsPolicy();

RejectedExecutionHandler handler = new 
ThreadPoolExecutor(2, 5, 60, TimeUnit.SECONDS, queue,handler);

worker ：工作类（重点）。 后面详细分析分析

内部工作原理

构造方法赋值

	public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);}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,RejectedExecutionHandler handler) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), handler);}public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;} 

•corePoolSize ：线程池核心线程数量，包括空闲线程。
• maximumPoolSize：池中允许的最大线程数
• keepAliveTime： 当线程数大于核心时，此为终止前多余的空闲线程等待新任务的最长时间
• unit：keepAliveTime 参数的时间单位
• workQueue ：执行前用于保持任务的队列。此队列仅保持由 execute 方法提交的 Runnable 任务
• threadFactory：执行程序创建新线程时使用的工厂
• handler ：由于超出线程范围和队列容量而使执行
分析源码：构造函数都是调用参数多的构造函数。结论：我们使用构造函数创建ThreadPoolExecutor对象时，threadFactory可以不传，默认使用Executors.defaultThreadFactory()；handler可以不传，默认new AbortPolicy()策略，即抛出异常。

属性分析

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 bitsprivate 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;//得到线程池运行状态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; }

ctl：这个变量是整个类的核心，AtomicInteger(原子操作类，理解为一个多线程同步操作时，可以保证原子性操作的Integer)类型保证了对这个变量的操作是原子的。

Integer.SIZE：为32，Integer的二进制32中最后一位表示正负数1：负数0：正数。所以经过按位运算ctl初始值为-536870912（二进制：101后面29个0）
这一个变量保存了两个内容：
1、所有有效线程的数量
2、各个线程的状态（runState）
低29位存线程数，高3位存runState,这样runState有5个值：
RUNNING:-536870912（二进制：101后面29个0）
SHUTDOWN:0（二进制：32个0）
STOP:536870912（二进制：001后面29个0）
TIDYING:1073741824： 所有的任务都已经停止，workerCount为0，线程变成TIDYING状态会运行terminate()（二进制：010后面29个0）
TERMINATED:1610612736：terminated（）已经完成（二进制：011后面29个0）

线程池中各个状态间的转换比较复杂：
RUNNING状态：线程池正常运行，可以接受新的任务并处理队列中的任务；
SHUTDOWN状态：不再接受新的任务，但是会执行队列中的任务；
STOP状态：不再接受新任务，不处理队列中的任务

内部工作类Worker

private final class Workerextends AbstractQueuedSynchronizerimplements 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); // inhibit interrupts until runWorkerthis.firstTask = firstTask;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) {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) {}}}}

Worker类继承于AbstractQueuedSynchronizer类，任务执行时获取释放锁，这样可以防止中断，目的在于唤醒线程，而不是中断正在执行的任务。

用AQS框架实现了一个简单的非重入的互斥锁， 实现互斥锁主要目的是为了中断的时候判断线程是在空闲还是运行

之所以不用ReentrantLock是为了避免任务执行的代码中修改线程池的变量，如setCorePoolSize，因为ReentrantLock是可重入的。

线程池运行思路

使用线程池运行任务时，调用线程池的submit 或者 execute 方法执行任务。execute方法为执行方法，没有返回值。submit方法由抽象类AbstractExecutorService实现内部调用了execute方法（这里使用了模板模式，execute方法由子类实现，AbstractExecutorService本身不实现这个方法），然后返回Future结果。

AbstractExecutorService类中submit方法的实现代码
/*** @throws RejectedExecutionException {@inheritDoc}* @throws NullPointerException       {@inheritDoc}*/public Future<?> submit(Runnable task) {if (task == null) throw new NullPointerException();RunnableFuture<Void> ftask = newTaskFor(task, null);execute(ftask);return ftask;}/*** @throws RejectedExecutionException {@inheritDoc}* @throws NullPointerException       {@inheritDoc}*/public <T> Future<T> submit(Runnable task, T result) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task, result);execute(ftask);return ftask;}/*** @throws RejectedExecutionException {@inheritDoc}* @throws NullPointerException       {@inheritDoc}*/public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask;}

运行思路：
提交一个任务到线程池中，线程池的处理流程如下：
1、判断线程池里的核心线程是否都在执行任务，如果不是（核心线程空闲或者还有核心线程没有被创建）则创建一个新的工作线程来执行任务。如果核心线程都在执行任务，则进入下个流程。
2、线程池判断工作队列是否已满，如果工作队列没有满，则将新提交的任务存储在这个工作队列里。如果工作队列满了，则进入下个流程。
3、判断线程池里的线程是否都处于工作状态，如果没有，则创建一个新的工作线程来执行任务。如果已经满了，则交给饱和策略来处理这个任务。

public void execute(Runnable command) {if (command == null)throw new NullPointerException();//AtomicInteger类中的get方法获取对象Integer值int c = ctl.get();//workerCountOf方法获取工作线程数//判断工作线程数是否小余核心线程数if (workerCountOf(c) < corePoolSize) {//直接启动新的线程。第二个参数true:addWorker中会重新检查workerCount是否小于corePoolSizif (addWorker(command, true))return;c = ctl.get();}//isRunning方法线程池是否为RUNNING状态//workQueue.offer方法添加一个任务到队列中//活动线程数 >= corePoolSize，runState为RUNNING && 队列未满if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();//runState为不为RUNNING，则从workQueue中移除任务if (! isRunning(recheck) && remove(command))// 采用线程池指定的策略拒绝任务reject(command);//线程池处于RUNNING状态 || 线程池处于非RUNNING状态但是任务移除失败//如果线程数为0else if (workerCountOf(recheck) == 0)//这行代码是为了SHUTDOWN状态下没有活动线程了，但是队列里还有任务没执行这种特殊情况。//添加一个null任务是因为SHUTDOWN状态下，线程池不再接受新任务addWorker(null, false);}//两种情况：//非RUNNING状态执行reject方法，采用线程池指定的策略拒绝任务//队列满了启动新的非核心线程-如果启动非核心线程失败则执行reject方法，采用线程池指定的策略拒绝任务else if (!addWorker(command, false))reject(command);}

Worker类源码分析

构造方法

Worker(Runnable firstTask) {//设置线程的中断状态，这里设置为-1.抑制了线程中断setState(-1); // inhibit interrupts until runWorker//将任务赋值给Worker的一个Runnable属性对象this.firstTask = firstTask;//Worker实现了Runnable接口，所以可以将自己当作一个任务创建一个thread线程this.thread = getThreadFactory().newThread(this);}

addWorker方法分析

private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// 这条语句等价：rs >= SHUTDOWN && (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty())// 满足下列三个条件之一则直接返回false，线程创建失败:// rs > SHUTDOWN:STOP || TIDYING || TERMINATED 此时不再接受新的任务，且所有任务执行结束// rs = SHUTDOWN:firtTask != null 此时不再接受任务，但是仍然会执行队列中的任务// rs = SHUTDOWN:firtTask == null:workQueue为空。// execute方法的addWorker(null,false)，任务为null && 队列为空// 最后一种情况也就是说SHUTDONW状态下，如果队列不为空还得接着往下执行，为什么？add一个null任务目的到底是什么？// 看execute方法只有workCount==0的时候firstTask才会为null结合这里的条件就是线程池SHUTDOWN了不再接受新任务// 但是此时队列不为空，那么还得创建线程把任务给执行完才行if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;//两种情况：// 1.线程池状态为RUNNING// 2.SHUTDOWN状态，但队列中还有任务需要执行for (;;) {int wc = workerCountOf(c);//做判断，工作线程如果大于Ineger二进制29位1（线程数太多）//或者是创建核心线程时大于核心线程数，创建非核心线程时大于最大线程数//则直接返回falseif (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;// 原子操作递增workCount（线程数）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 {w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {final ReentrantLock mainLock = this.mainLock;// 并发的访问线程池workers（源码中是一个HashSet对象）对象必须加锁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状态 || SHUTDONW状态下清理队列中剩余的任务if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();// 将新启动的线程添加到线程池中workers.add(w);// 更新largestPoolSizint s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}// 启动新添加的线程，这个线程首先执行firstTask，然后不停的从队列中取任务执行// 当等待keepAlieTime还没有任务执行则该线程结束。见runWoker和getTask方法的代码。if (workerAdded) {//执行Worker的run方法t.start();workerStarted = true;}}} finally {// 线程启动失败，则从wokers中移除w并递减wokerCountif (! workerStarted)// 递减wokerCount会触发tryTerminate方法addWorkerFailed(w);}return workerStarted;}

所以任务添加成功后实际执行的是run方法

run方法

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

Worker中的run方法，执行Worker的runWorker方法：

所以最后任务添加成功后实际执行的是runWorker这个方法，简单来说它做的就是：

第一次启动会执行初始化传进来的任务firstTask；
然后会从workQueue中取任务执行，如果队列为空则等待keepAliveTime这么长时间。

runWorker方法

final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;// Worker的构造函数中抑制了线程中断setState(-1)，所以这里需要unlock从而允许中断w.unlock(); // allow interrupts// 用于标识是否异常终止，finally中processWorkerExit的方法会有不同逻辑// 为true的情况：1.执行任务抛出异常；2.被中断。boolean completedAbruptly = true;try {// 如果getTask返回null那么getTask中会将workerCount递减，如果异常了这个递减操作会在processWorkerExit中处理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 interruptif ((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 {// 和beforeExecute一样，留给子类去重载afterExecute(task, thrown);}} finally {task = null;w.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {// 结束线程的一些清理工作processWorkerExit(w, completedAbruptly);}}

getTask方法

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.// 1.rs > SHUTDOWN 所以rs至少等于STOP,这时不再处理队列中的任务// 2.rs = SHUTDOWN 所以rs>=STOP肯定不成立，这时还需要处理队列中的任务除非队列为空// 这两种情况都会返回null让runWoker退出while循环也就是当前线程结束了，所以必须要decrement// wokerCountif (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {// 递减workerCount值decrementWorkerCount();return null;}int wc = workerCountOf(c);// Are workers subject to culling?// Are workers subject to culling?// 1.core thread允许被超时，那么超过corePoolSize的的线程必定有超时// 2.allowCoreThreadTimeOut == false && wc >// corePoolSize时，一般都是这种情况，core thread即使空闲也不会被回收，只要超过的线程才会boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;// 从addWorker可以看到一般wc不会大于maximumPoolSize，所以更关心后面半句的情形：// 1. timedOut == false 第一次执行循环， 从队列中取出任务不为null方法返回 或者// poll出异常了重试// 2.timeOut == true && timed ==// false:看后面的代码workerQueue.poll超时时timeOut才为true，// 并且timed要为false，这两个条件相悖不可能同时成立（既然有超时那么timed肯定为true）// 所以超时不会继续执行而是return null结束线程。（重点：线程是如何超时的？？？）if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {// workerCount递减，结束当前threadif (compareAndDecrementWorkerCount(c))return null;continue;}try {// 1.以指定的超时时间从队列中取任务// 2.core thread没有超时Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;// 超时timedOut = true;} catch (InterruptedException retry) {// 线程被中断重试timedOut = false;}}}

processWorkerExit方法

线程退出或者异常结束后会执行这个方法做一些清理工作

private void processWorkerExit(Worker w, boolean completedAbruptly) {// 正常的话再runWorker的getTask方法workerCount已经被减一了if (completedAbruptly) // If abrupt, then workerCount wasn't adjusteddecrementWorkerCount();final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// 累加线程的completedTaskscompletedTaskCount += w.completedTasks;// 从线程池中移除超时或者出现异常的线程workers.remove(w);} finally {mainLock.unlock();}// 尝试停止线程池tryTerminate();int c = ctl.get();// runState为RUNNING或SHUTDOWNif (runStateLessThan(c, STOP)) {// 线程不是异常结束if (!completedAbruptly) {// 线程池最小空闲数，允许core thread超时就是0，否则就是corePoolSizeint min = allowCoreThreadTimeOut ? 0 : corePoolSize;// 如果min == 0但是队列不为空要保证有1个线程来执行队列中的任务if (min == 0 && ! workQueue.isEmpty())min = 1;// 线程池还不为空那就不用担心了if (workerCountOf(c) >= min)return; // replacement not needed}// 1.线程异常退出// 2.线程池为空，但是队列中还有任务没执行，看addWoker方法对这种情况的处理addWorker(null, false);}}

tryTerminate方法

尝试停止线程池。processWorkerExit方法中会尝试调用tryTerminate来终止线程池。这个方法在任何可能导致线程池终止的动作后执行：比如减少wokerCount或SHUTDOWN状态下从队列中移除任务。

final void tryTerminate() {for (;;) {int c = ctl.get();// 以下状态直接返回：// 1.线程池还处于RUNNING状态// 2.SHUTDOWN状态但是任务队列非空// 3.runState >= TIDYING 线程池已经停止了或在停止了if (isRunning(c) ||runStateAtLeast(c, TIDYING) ||(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))return;// 只能是以下情形会继续下面的逻辑：结束线程池。// 1.SHUTDOWN状态，这时不再接受新任务而且任务队列也空了// 2.STOP状态，当调用了shutdownNow方法// workerCount不为0则还不能停止线程池,而且这时线程都处于空闲等待的状态// 需要中断让线程“醒”过来，醒过来的线程才能继续处理shutdown的信号。if (workerCountOf(c) != 0) { // Eligible to terminate// runWoker方法中w.unlock就是为了可以被中断,getTask方法也处理了中断。// ONLY_ONE:这里只需要中断1个线程去处理shutdown信号就可以了。interruptIdleWorkers(ONLY_ONE);return;}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));// 继续awaitTerminationtermination.signalAll();}return;}} finally {mainLock.unlock();}// else retry on failed CAS}}

shutdown方法

会将runState置为SHUTDOWN，会终止所有空闲的线程。

public void shutdown() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();// 线程池状态设为SHUTDOWN，如果已经至少是这个状态那么则直接返回advanceRunState(SHUTDOWN);// 注意这里是中断所有空闲的线程：runWorker中等待的线程被中断 → 进入processWorkerExit →// tryTerminate方法中会保证队列中剩余的任务得到执行。interruptIdleWorkers();onShutdown(); // hook for ScheduledThreadPoolExecutor} finally {mainLock.unlock();}tryTerminate();}

shutdownNow方法

将runState置为STOP。和shutdown方法的区别，这个方法会终止所有的线程。

public List<Runnable> shutdownNow() {List<Runnable> tasks;final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();// STOP状态：不再接受新任务且不再执行队列中的任务advanceRunState(STOP);// 中断所有线程interruptWorkers();// 返回队列中还没有被执行的任务tasks = drainQueue();} finally {mainLock.unlock();}tryTerminate();return tasks;}

两个方法的主要区别在于shutdown方法调用的是interruptIdleWorkers方法，而shutdownNow方法实际调用的是Worker类的interruptIfStarted方法：

interruptIdleWorkers方法

private void interruptIdleWorkers(boolean onlyOne) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers) {Thread t = w.thread;// w.tryLock能获取到锁，说明该线程没有在运行，因为runWorker中执行任务会先lock，// 因此保证了中断的肯定是空闲的线程。if (!t.isInterrupted() && w.tryLock()) {try {t.interrupt();} catch (SecurityException ignore) {} finally {w.unlock();}}if (onlyOne)break;}} finally {mainLock.unlock();}}

interruptIfStarted方法

void interruptIfStarted() {Thread t;if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {try {t.interrupt();} catch (SecurityException ignore) {}}}

这里就体现出了Woker类实现AQS的主要作用。