Java并发之ThreadPoolExecutor源码解析（二）
ThreadPoolExecutor
ThreadPoolExecutor是ExecutorService的一种实现，能够用若干已经池化的线程执行被提交的任务。使用线程池能够帮助咱们限定和整合程序资源，尽量避免建立新的线程来执行任务从而下降任务调用的开销，在执行大量异步任务的时候反而能得到更好的性能。此外，ThreadPoolExecutor还会维护一些统计信息，好比已完成的任务数量。java
juc包的做者Doug Lea推荐程序员尽可能使用更为便利的Executors类的工厂方法来配置线程池：程序员
- Executors.newCachedThreadPool()：建立一个线程池，能够根据线程池的须要来建立任务。这个线程池比较适用执行周期较短且量大的异步任务，在调用execute(...)方法时若是线程池中存在闲置的线程，将复用闲置线程，不然建立一个新线程执行任务。若是线程的闲置时间超过60s，线程将被终止并从线程池内移除。所以，该线程池即使空闲时间再长，也不会有资源的消耗。
- Executors.newFixedThreadPool(int nThreads)：建立一个线程池，nThreads为池内线程的数量，池内最多同时有nThreads个线程并行处理任务，若是有新的任务提交到线程池，会先暂存在线程池中的无边界任务队列进行等待，直到有线程可用。若是有线程在执行期间由于错误提早终止，线程池将启动一个新的线程代替原先的线程继续处理任务队列中处于等待的任务。除非显式调用shutdown()，不然线程池中的线程将一直存在。
- Executors.newSingleThreadExecutor()：建立一个Executor，该Executor使用一个工做线程处理任务。若是线程在执行期间由于错误而终止，将启动一个新的线程代替原先的线程继续处理无边界任务队列中处于等待的任务。队列中的任务是按顺序执行，任什么时候刻都不会有多个任务处于活跃状态。与newFixedThreadPool(1)不一样，newFixedThreadPool(int nThreads)生成的线程池，能够强转为ThreadPoolExecutor类型，再调用setCorePoolSize(int corePoolSize)方法设置核心线程数，而newSingleThreadExecutor()的实现类为FinalizableDelegatedExecutorService，没法直接设置核心线程数。
上面三种是较为常见的配置线程池的工厂方法，若是有须要根据业务场景特殊配置线程池的，请看下面的参数：安全
核心线程数和最大线程数
ThreadPoolExecutor将根据corePoolSize（核心线程数）和maximumPoolSize（最大线程数）设置的边界自动调整线程池内工做线程的数量（经过getPoolSize()），corePoolSize能够经过getCorePoolSize()、setCorePoolSize(int corePoolSize)获取和设置核心线程数，maximumPoolSize能够经过getMaximumPoolSize()、setMaximumPoolSize(int maximumPoolSize)获取和设置最大线程数。当有新的任务提交时，若是工做线程少于核心线程数，将会建立一个新线程来执行该任务，即使其余工做线程处于闲置状态。若是工做线程多于corePoolSize但少于maximumPoolSize，则当任务队列满的时候才会建立新线程。若是corePoolSize和maximumPoolSize数值同样，则建立一个固定大小的线程池；若是将maximumPoolSize设置为Integer.MAX_VALUE，则线程池能够容纳任意数量的并发任务。并发
按需构造
核心线程只有当有新任务到达时才会建立，但咱们能够重写prestartCoreThread() 或者prestartAllCoreThreads()来预先启动核心线程。若是在构造一个线程池时，传入的任务队列已经存在任务，则须要线程池初始化完毕后，预先启动线程。异步
建立新线程
使用ThreadFactory（线程工厂）建立新线程。若是没有特别指定，则使用Executors.defaultThreadFactory()做为默认的线程工厂，该线程工厂所建立的线程都位于相同的线程组（ThreadGroup）中，线程的优先级都是NORM_PRIORITY，线程守护状态都为false。经过提供不一样线程工厂的实现，你能够修改线程名、线程组、线程优先级和守护状态等等。函数
活跃时间
若是线程池中的线程数超过核心线程数，多出的线程若是空闲时间超出keepAliveTime（活跃时间）将会终止，回收再也不活跃的线程。当有须要时，新的线程会从新建立。能够经过setKeepAliveTime(long time, TimeUnit unit)动态设置活跃时间。若是time设置为Long.MAX_VALUE，unit设置为TimeUnit.NANOSECONDS，那么多余的空闲线程将不会在关闭线程池以前回收。若是调用allowCoreThreadTimeOut(boolean value)传入的value为true，那么keepAliveTime将适用于核心线程，若是allowCoreThreadTimeOut为true且keepAliveTime不为0，核心线程的空闲时间超出活跃时间，核心线程也会被回收。oop
队列
阻塞队列（BlockingQueue）容许在获取元素时陷入等待，直到有元素加入到队列中。调用阻塞队列方法时，有些方法不必定立刻返回，可能会在将来某个时刻达成某些条件时返回。阻塞队列的方法伴随四种形式：布局
- 抛出异常。
- 返回特殊值，null或者false，具体视操做而定。
- 调用线程无限期陷入阻塞直到某些条件达成。
- 限定阻塞时长。
| 抛异常 | 特殊值 | 阻塞 | 超时 | |
| 插入 | add(e) | offer(e) | put(e) | offer(e, time, unit) |
| 移除 | remove() | poll() | take() | poll(time, unit) |
| 检查（获取但不移除队列头部元素） | element() | peek() |
阻塞队列不接受null元素，若是调用add、put、offer尝试添加一个null元素，将会抛出NullPointerException异常，当调用poll操做失败时也会返回null。阻塞队列可能有容量限制，不管什么时候都不能向队列添加超过剩余容量的元素，不然只能调用put方法陷入阻塞，直到有剩余的空间能够容纳元素。若是对队列的容纳空间没有限制，则剩余容量返回Integer.MAX_VALUE。阻塞队列的实现通常用于生产者-消费者队列的场景，此外阻塞队列还实现了Collection接口，所以，队列还可使用remove(x)来移除元素。性能
阻塞队列是线程安全的，全部排队方法的实现都是用内部锁或者其余并发控制手段来实现原子性的。然而，除非是特殊规定，不然大部分集合操做，如：addAll、containsAll、retainAll 、removeAll不必定要保证原子性。所以，可能出如今调用addAll(c)时，只添加c中一部分的元素就抛出异常。阻塞队列本质上并不支持关闭的操做，如：close或shutdown，当有须要让队列再也不接受新元素。若是有这种须要或者特性更倾向于以来队列的实现。一种常见的策略是生产者往队列插入具备特殊标识的对象，当消费者使用对象时，会对特殊标识进行解释。ui
注意，阻塞队列容许多个生产者和消费者同时使用，以下：
class Producer implements Runnable {
private final BlockingQueue queue;
Producer(BlockingQueue q) { queue = q; }
public void run() {
try {
while (true) { queue.put(produce()); }
} catch (InterruptedException ex) { ... handle ...}
}
Object produce() { ... }
}
class Consumer implements Runnable {
private final BlockingQueue queue;
Consumer(BlockingQueue q) { queue = q; }
public void run() {
try {
while (true) { consume(queue.take()); }
} catch (InterruptedException ex) { ... handle ...}
}
void consume(Object x) { ... }
}
class Setup {
void main() {
BlockingQueue q = new SomeQueueImplementation();
Producer p = new Producer(q);
Consumer c1 = new Consumer(q);
Consumer c2 = new Consumer(q);
new Thread(p).start();
new Thread(c1).start();
new Thread(c2).start();
}
}
public interface BlockingQueue<E> extends Queue<E> {
/**
* 在不超过队列容量的状况下插入一个元素将返回true，，若是队列没有多余的空间抛出
* IllegalStateException异常，当使用容量受限的队列时最好使用offer。
*
* @param e 待添加进队列的元素
* @return 返回true表明元素加入队列成功
*/
boolean add(E e);
/**
* 相比add(E)若是队列满时插入元素不报错，只是返回false。
*
* @param e 待添加进队列的元素
* @return 返回true表明元素加入队列成功，队列满时没法插入返回false
*/
boolean offer(E e);
/**
* 将一个元素插入到队列，若是有必要会等待队列有多余空间能够插入。若是调用
* put(E)的线程被中断，将抛出中断异常InterruptedException
*
* @param e 待添加进队列的元素
*/
void put(E e) throws InterruptedException;
/**
* 相比offer(E)多了插入元素时陷入等待，若是等待期间队列依旧
* 没有多余的空间容纳元素，则返回false，若是等待期间能插入则返回true。
* 若是等待期间线程被中断，则抛出中断异常InterruptedException。
*
* @param e 待添加进队列的元素
* @param timeout 等待时长
* @param unit 等待时长单位
*/
boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException;
/**
* 取出并删除队列的头部元素，若是队列为空，则会陷入等待，直到队列有新的元素加入，
* 若是等待期间线程被中断，将抛出中断异常
*
* @return 对头元素
*/
E take() throws InterruptedException;
/**
* 相比take()多了一个等待时长，若是队列自己有元素，或者队列原先有空但等待期间有元素
* 加入则返回头部元素，不然队列为空且等待期间没有元素加入，则返回null。若是等待期间调用线程
* 被中断，则抛出InterruptedException异常。
*/
E poll(long timeout, TimeUnit unit)
throws InterruptedException;
/**
* 返回队列理想情况下可无阻塞容纳元素的容量。注意：咱们不能经过此方法判断元素是否插入成功，
* 由于可能存在别的线程插入或删除队列中的元素。
*/
int remainingCapacity();
/**
* 从队列中移除指定的元素，若是队列中存在一个或多个相同的元素，即：o.equal(e)，则删除并返回true。
*/
boolean remove(Object o);
/**
* 若是队列存在一个或多个相同的元素，即：o.equal(e)，则返回true。
*/
boolean contains(Object o);
/**
* 删除此队列中全部可用元素，并将它们移动到给定的集合c中。当咱们把元素从原队列取出时添加到集合c时，
* 可能出现异常致使元素既不在原队列，也不在集合中。队列一样实现了Collection接口，若是将原队列当作
* 参数传入将抛出IllegalArgumentException异常
* @param c 将队列元素传输到给定的集合c。
* @return 加入到集合c中元素的数量。
*/
int drainTo(Collection<? super E> c);
/**
* 最多将maxElements个元素从队列传输到给定集合，其余和drainTo(Collection<? super E> c)同样。
*/
int drainTo(Collection<? super E> c, int maxElements);
}
任何阻塞队列均可以用来获取和保存任务，如何使用队列视当前线程池的大小而定：
- 若是工做线程的数量小于corePoolSize，那么Executor更倾向于添加新线程执行而非让任务排队。
- 若是工做线程的数量大于等于corePoolSize，那么Executor更倾向于把任务添加到队列等待执行，而非建立新线程。
- 若是队列已满，且线程池内的线程数小于maximumPoolSize，Executor会建立一个新线程来执行任务，不然任务会被拒绝。
一般有三种排队策略：
- 同步队列（SynchronousQueue）：若是有线程从同步队列获取任务，则移交给线程，不然持有任务，若是有新的任务尝试入队将返回失败，能够根据入队结果判断是否要构造一个新线程。同步队列能够避免当处理多个请求时内部依赖出现锁定，直接交接任务要求对maximumPoolSize这一参数不作限制，即maximumPoolSize为Integer.MAX_VALUE，避免线程池拒绝提交任务。但若是线程池处理任务的速度不够快，可能出现线程无限增加。
- 无界队列（LinkedBlockingQueue）：若是线程池核心工做线程都在执行任务时，新提交的任务将在队列中等待。若是任务提交速度过快而执行任务的速度又慢，将致使队列中的任务无限增加。
- 有界队列（ArrayBlockingQueue）：当与有限的maximumPoolSize配合使用时，有界队列能够防止资源耗尽，但如何设定有界队列的大小是一个很难的问题。若是队列过大而maximumPoolSize太小，能够减小CPU的使用、操做系统资源和上下文切换的开销，这有可能致使低吞吐量。若是队列太小而maximumPoolSize过大，这会使得CPU十分繁忙，甚至出现巨大的调度开销，一样也会下降吞吐量。
拒绝任务
若是线程池关闭后有新任务提交、或者在任务队列已满的状况下，线程池到达最大线程数且全部线程都在执行任务，将调用RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor)拒绝任务。默认提供四种预约义拒绝策略：
- ThreadPoolExecutor.AbortPolicy：默认状况下使用的策略，抛出RejectedExecutionException异常。
- ThreadPoolExecutor.CallerRunsPolicy：使用调用线程执行任务，这种机制能够下降任务提交的速度。
- ThreadPoolExecutor.DiscardPolicy：将没法入队也不能执行的任务直接丢弃。
- ThreadPoolExecutor.DiscardOldestPolicy：若是线程池未关闭，则获取并丢弃队列的头部元素，再尝试用线程池执行任务，这一策略可能会失败，若是失败则重复以前的步骤。
除了上述四种步骤，咱们也能够自定义拒绝策略。
钩子函数
ThreadPoolExecutor提供了可重写函数beforeExecute(java.lang.Thread, java.lang.Runnable)、afterExecute(java.lang.Runnable, java.lang.Throwable)，分别容许咱们在执行任务前和执行任务后作一些操做，这些方法能够控制执行环境，例如：初始化ThreadLocals、收集统计信息、添加日志等等。此外，也能够重写terminated()方法，当线程池彻底终止后会调用此方法。
若是钩子函数或者回调函数抛出异常，工做线程可能会终止。
队列维护
ThreadPoolExecutor提供了getQueue()容许外部获取队列进行监控和调试，但无论出于什么目的尽可能少使用此方法。此外ThreadPoolExecutor还提供了remove(java.lang.Runnable) 和purge()用于删除任务，purge()能够取消大量处于排队等待的任务。
销毁
当一个线程池再也不有引用指向，且线程池内没有存活线程将会自动关闭。若是你指望一个未手动调用shutdown()方法的线程池会被回收，你要设置合理的线程存活时间（keep-alive times）、设置核心线程数为0，或者设置allowCoreThreadTimeOut为true，当核心线程空闲时间超过存活时间将被回收，当线程池没有引用指向，且无存活线程，就会被自动关闭并回收。
源码解析
ThreadPoolExecutor的ctl变量类型为AtomicInteger，这个数值有32位，包含两个部分：
- runState（运行状态）：线程池是否处于运行中、是否已关闭等等。
- workerCount（工做线程数）：线程池当前有多少个存活（忙碌或空闲）的线程。
为了将运行状态和工做线程数放在一个int字段，咱们划分前3位存储运行状态，后29位存储存活线程数量（2^29）-1（约5亿）。将来有可能调整ctl为AtomicLong类型，这可能须要调整移位和掩码，但若是使用AtomicInteger，ThreadPoolExecutor的代码会更简单也更高效一些。
workerCount是线程池中还存活的线程数，该值有时候可能会短暂不一样于池内实际的存活线程数。当须要增长工做线程时，会先用CAS的方式对workerCount+1，而后才向ThreadFactory申请建立一个线程。
runState为线程池提供了生命周期控制，有如下几种状态：
- RUNNING：容许接受新任务和处理队列中任务。
- SHUTDOWN：不接受新任务，但处理队列中任务。
- STOP：不接受新任务，不处理队列中任务，同时尝试中断正在执行的任务。
- TIDYING：全部任务都终止，workerCount为0，线程池状态过分到TIDYING，将要执行terminated()钩子函数。
- TERMINATED：terminated()函数执行完毕。
下面，咱们来看看线程池状态的转换：
- RUNNING->SHUTDOWN：调用shutdown()。
- (RUNNING or SHUTDOWN)->STOP：调用shutdownNow()。
- SHUTDOWN->TIDYING：当线程池再也不有存活线程且队列为空。
- STOP->TIDYING：当线程池再也不有存活线程。
- TIDYING->TERMINATED：调用terminated()。
检测线程池的状态从SHUTDOWN过分到TIDYING并不是易事，由于在SHUTDOWN状态下，队列可能从非空变为空，即仍然有存活的线程处理队列中的任务。只有workerCount为0且队列为空，才能结束线程池。
public class ThreadPoolExecutor extends AbstractExecutorService {
/**
* 类型为AtomicInteger的ctl能够保证线程安全，该数值分两个部分：
* 前3位为runState表明线程池当前的状态：RUNNING~TERMINATED，
* 后29位为workerCount表明线程池内存活线程数量。
*/
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
/**
* Integer.SIZE=32，COUNT_BITS=32-3=29，用COUNT_BITS来划分
* runState和workerCount。
*/
private static final int COUNT_BITS = Integer.SIZE - 3;
/**
* 1 << COUNT_BITS = 0010 0000 0000 0000 0000 0000 0000 0000
* COUNT_MASK = (1 << COUNT_BITS) - 1 = 0001 1111 1111 1111 1111 1111 1111 1111
* ~COUNT_MASK = 1110 0000 0000 0000 0000 0000 0000 0000
* COUNT_MASK能够帮助咱们计算线程池当前的runState和workerCount。
* 假设ctl的值为：0000 0000 0000 0000 0000 0000 0000 0011
* 调用runStateOf(ctl.get())，将作ctl.get() & ~COUNT_MASK运算：
* 0000 0000 0000 0000 0000 0000 0000 0011
* & 1110 0000 0000 0000 0000 0000 0000 0000
* = 0000 0000 0000 0000 0000 0000 0000 0000
* 由此咱们能够获得，线程池当前状态为0，即SHUTDOWN。
* 调用workerCountOf(ctl.get())，将作ctl.get() & COUNT_MASK运算：
* 0000 0000 0000 0000 0000 0000 0000 0011
* & 0001 1111 1111 1111 1111 1111 1111 1111
* = 0000 0000 0000 0000 0000 0000 0000 0011
* 由此咱们能够获得，线程池还有3个存活的线程。
*/
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
/**
* -1的二进制表示为：1111 1111 1111 1111 1111 1111 1111 1111，
* 左移29位为：1110 0000 0000 0000 0000 0000 0000 0000
*/
private static final int RUNNING = -1 << COUNT_BITS;
/**
* 0的二进制表示为：0000 0000 0000 0000 0000 0000 0000 0000，
* 左移29位和原先没有变化。
*/
private static final int SHUTDOWN = 0 << COUNT_BITS;
/**
* 0的二进制表示为：0000 0000 0000 0000 0000 0000 0000 0001，
* 左移29位为：0010 0000 0000 0000 0000 0000 0000 0000
*/
private static final int STOP = 1 << COUNT_BITS;
/**
* 2的二进制表示为：0000 0000 0000 0000 0000 0000 0000 0010，
* 左移29位为：0100 0000 0000 0000 0000 0000 0000 0000
*/
private static final int TIDYING = 2 << COUNT_BITS;
/**
* 3的二进制表示为：0000 0000 0000 0000 0000 0000 0000 0011，
* 左移29位为：0110 0000 0000 0000 0000 0000 0000 0000
*/
private static final int TERMINATED = 3 << COUNT_BITS;
private static int runStateOf(int c) {
return c & ~COUNT_MASK;
}
private static int workerCountOf(int c) {
return c & COUNT_MASK;
}
/**
* 根据runState和workerCount生成ctl，好比初始化线程池时，
* ctl = new AtomicInteger(ctlOf(RUNNING, 0))，表明线程池
* 的状态为RUNNING，存活线程数量为0。
* @param rs 线程池状态
* @param wc 存活线程数量
* @return
*/
private static int ctlOf(int rs, int wc) {
return rs | wc;
}
/**
* 经过位运算在进行一些状态的判断时，咱们不须要解析ctl的运行状态，
* 假设当前ctl：1110 0000 0000 0000 0000 0000 0000 0011，
* 咱们要判断线程池状态是否小于STOP，因为ctl开头为1110，以补码的
* 方式来计算，ctl的值必然为负，STOP开头为0010，以补码方式计算为正数，
* 因此ctl必然小于STOP。
* ctl的布局还能保证workerCount永远不会为负数。
*/
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
/**
* 判断线程池至少处于某个状态，假设线程池如今队列为空且无任何存活线程，
* 因此能保证线程池处于TIDYING状态，若是s咱们传入STOP，TIDYING的开头
* 为0100，STOP的开头为0010，TIDYING>STOP，因此咱们能知道，线程池至少
* 处于STOP以上包含STOP的状态。
*/
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
//线程池是否处于RUNNING状态
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
//CAS增长worker数量
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
//CAS减小worker数量
private boolean compareAndDecrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect - 1);
}
}
咱们知道AbstractExecutorService.submit(...)方法最终会调用execute(Runnable command)方法，而AbstractExecutorService类中并无实现execute(Runnable command)方法，它将execute(Runnable command)的实现交由子类。那么咱们来看看ThreadPoolExecutor又是如何实现execute(Runnable command)方法呢？当一个任务提交到线程池，它的执行流程又是如何呢？来看下面的代码注释：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//若是工做线程的数量小于核心线程的数量，则尝试增长工做线程
if (workerCountOf(c) < corePoolSize) {//<1>
/*
* 若是成功增长工做线程，工做线程会执行咱们提交的任务，咱们就能够安心退出，
* 但线程池能够并发提交任务，可能存在在<1>处时工做线程小于核心线程数，
* 执行<2>处的addWorker(Runnable firstTask, boolean core)时，
* 其余线程先当前线程提交任务并增长工做线程，线程池内工做线程数超过核心线程数，
* 当前线程增长工做线程失败，不能直接退出。
* 注：addWorker(Runnable firstTask, boolean core)，core为true
* 表明增长核心线程，会将任务做为firstTask传入；而false表明增长非核心线程，
* 若是传入firstTask为null，则表明让工做线程去队列中拉取任务。
*/
if (addWorker(command, true))//<2>
return;
//若是<2>处增长工做线程失败，则从新获取ctl的值。
c = ctl.get();
}
//判断线程池是否处于运行中，且任务能够入队成功，若是二者成立，则进入<3>分支
if (isRunning(c) && workQueue.offer(command)) {//<3>
int recheck = ctl.get();
/*
* 从新获取ctl，由于可能在进入<3>分支的时候，线程池被关闭，
* 因此要从新判断线程池状态，若是线程池不是处于运行状态，且
* 任务成功被移除，则进入<4>分支，拒绝任务。
*/
if (!isRunning(recheck) && remove(command))//<4>
reject(command);
/*
* 为何这里要判断工做线程数量是否为0？由于若是设置allowCoreThreadTimeOut
* 为true的话，核心线程是能够为0的，可能代码执行到<3>处workQueue.offer(command)以前，
* 即任务还未入队，工做线程数量已经为0了，因此这里要从新根据ctl判断工做线程是否为0，
* 若是为0得再增长非核心线程去队列拉取并执行任务。
*/
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
/*
* 若是没有进入<3>分支，而到达<5>分支，通常分两种状况：
* 1.线程池被关闭，<3>处isRunning(c)为false，此时调用<5>处的
* addWorker(...)必然返回false，而后执行拒绝策略。
* 2.线程池处于运行状态，<3>处isRunning(c)为true，但队列已满
* workQueue.offer(command)返回false，入队失败。
* 这时候应该尝试建立非核心工做线程执行任务，若是工做线程数量没到达最大线程数，
* 则建立线程并执行任务，若是工做线程到达最大线程数，则addWorker(...)返回
* false，执行拒绝策略。
*/
else if (!addWorker(command, false))//<5>
reject(command);
}
从ThreadPoolExecutor.execute(Runnable command)的实现，咱们能够知道addWorker(Runnable firstTask, boolean core)方法是相当重要的，它决定了是否将任务添加进线程池执行。下面，咱们再来看看addWorker(Runnable firstTask, boolean core)方法：
/**
* 此方法会根据线程池当前的运行状态、线程池所设定的边界（核心线程数和最大线程数）。
* 若是线程池容许建立线程执行任务，则建立线程执行firstTask并相应调整工做线程的数量。
* 若是线程池状态处于已中止（STOP）、关闭（SHUTDOWN）则会返回false。若是向线程工
* 厂请求建立线程失败，也会返回false。线程建立失败分两种状况，一种是线程工厂返回null，
* 或者执行Thread.start()时出现异常（一般为OOM异常）。
*
* @param firstTask：新线程首要执行任务，若是没有则传入null。当工做线程数少于核心 线程数，线程池老是建立一个新线程来执行firstTask。
* @param core：根据core为true或者false，决定是以核心线程数或者最大线程数做为界限， 判断当前线程池的工做线程池是否小于界限，若是小于则容许建立线程。
* @return 若是成功添加工做线程则返回true。
*/
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (int c = ctl.get(); ; ) {//<1>
// Check if queue empty only if necessary.
/*
* 在这个地方addWorker(...)会返回false，即添加工做线程失败，
* 咱们来看看是什么状况下会进入这个分支：
* runStateAtLeast(c, SHUTDOWN)表明线程池运行状态至少处于
* SHUTDOWN，若是线程池还处于RUNNING运行状态，此方法不会当即
* 返回失败。因此咱们知道，要进入此分支，首要条件就是运行状态大于
* 等于SHUTDOWN。
* 以后若是runStateAtLeast(c, STOP)、firstTask != null、
* workQueue.isEmpty())这三个条件其一为true，则添加线程失败。
* 首先是runStateAtLeast(c, STOP)，若是线程池当前处于STOP
* 状态，这时候既不接受新任务，也不处理队列里的任务，因此无论
* firstTask是否为null，都返回false。
* 若是runStateAtLeast(c, STOP)为false，那运行状态只能是
* SHUTDOWN，SHUTDOWN状态下会处理队列里的任务，但再也不接受新
* 任务，因此firstTask不为null，也直接返回false。
* 若是运行状态既处于SHUTDOWN、firstTask也会空，且任务队列也
* 为空，则毫无必要增长工做线程，也直接返回false。
* 因此总结一下有两种状况不会进入此分支：
* 1.线程池处于RUNNING状态的时候。
* 2.线程池处于SHUTDOWN，但firstTask为空队列不为空时。
*/
if (runStateAtLeast(c, SHUTDOWN)
&& (runStateAtLeast(c, STOP)
|| firstTask != null
|| workQueue.isEmpty()))
return false;
for (; ; ) {//<2>
//根据core判断工做线程的上限，若是大于上限则返回false。
if (workerCountOf(c)
>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
return false;
/*
* 若是用CAS的方式成功增长工做线程的数量，则用break retry的方式
* 结束了retry对应的外层循环（即<1>处for循环），而不是break所在
* 的本层循环（即<2>处循环），代码会从<3>处开始执行。
*/
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
/*
* 若是上面用CAS的方式增长工做线程失败，则会从新判断线程池当前
* 状态是否至少处于SHUTDOWN，若是线程池已关闭，代码会跳到retry
* 处从新执行<1>处的for循环。若是线程池仍然处于RUNNING状态，则
* 重复执行<2>处的循环。
*/
if (runStateAtLeast(c, SHUTDOWN))
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
//<3>
boolean workerStarted = false;//若是工做线程启动成功，则赋值为true
boolean workerAdded = false;//若是工做线程添加成功则赋值为true
Worker w = null;
try {
//建立一个Worker对象，Worker对象会向线程工厂申请建立一个线程
w = new Worker(firstTask);
final Thread t = w.thread;
//若是线程工厂建立的Thread对象不为null，则进入此分支
if (t != null) {
/*
* 这里用可重入锁锁住try模块代码，由于要将以前建立好的
* w对象放进workers集合。
* 注：重入锁ReentrantLock的概念笔者会在之后的文章里
* 单独介绍，这里先简单理解，可重入锁就是禁止其余线程同时
* 访问mainLock.lock()到mainLock.unlock()之间的代码,
* 和synchronized有些相似。
*/
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int c = ctl.get();
/*
* 从新获取ctl，判断线程池当前是否处于运行状态，或小于STOP状态，
* 即线程池处于RUNNING或SHUTDOWN，若是处于RUNNING则直接进入分支，
* 若是处于SHUTDOWN且首要执行任务为空，表明可能要启动一个工做线程
* 来执行队列中的任务。
*/
if (isRunning(c) ||
(runStateLessThan(c, STOP) && firstTask == null)) {
/*
* 判断线程是否已经启动，若是使用的是Executors.DefaultThreadFactory
* 默认的线程工厂，正常来讲建立出来的Thread对象都是线程未启动的，即：还没有
* 调用Thread.start()。但ThreadPoolExecutor容许咱们传入定制化的线程
* 工厂，因此会存在线程工厂建立出Thread对象，但Thread对象已调用过start()
* 方法的可能。
*/
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//将建立好的worker添加进集合workers。
workers.add(w);
//更新历史上最大的工做线程数，即workers.size()。
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
//将worker添加进workers后，更新workerAdded的值为true。
workerAdded = true;
}
} finally {
mainLock.unlock();
}
//若是worker成功加入集合，则启动线程，并更新workerStarted为true。
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
/*
* 若是worker没有启动，表明worker没有加入到workers集合，
* 可能线程池状态>=STOP，则须要执行添加工做线程失败操做。
*/
if (!workerStarted)
addWorkerFailed(w);
}
//返回工做线程是否启动成功
return workerStarted;
}
- 1. Java并发之ThreadPoolExecutor源码解析（三）
- 2. Java并发之ThreadPoolExecutor源码解析（一）
- 3. java并发之ThreadPoolExecutor分析
- 4. Java并发之AQS源码分析（二）
- 5. java并发之线程池（ThreadPoolExecutor源码分析）
- 6. Java并发之线程池ThreadPoolExecutor源码分析学习
- 7. ThreadPoolExecutor源码解析
- 8. ThreadPoolExecutor 源码解析
- 9. Java线程池之ThreadPoolExecutor源码分析
- 10. Java源码解析 - ThreadPoolExecutor 线程池
- 更多相关文章...
- • Rust 并发编程 - RUST 教程
- • XML DOM 解析器 - XML DOM 教程
- • Java Agent入门实战（二）-Instrumentation源码概述
- • 互联网组织的未来：剖析GitHub员工的任性之源
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. eclipse设置粘贴字符串自动转义
- 2. android客户端学习-启动模拟器异常Emulator: failed to initialize HAX: Invalid argument
- 3. android.view.InflateException: class com.jpardogo.listbuddies.lib.views.ListBuddiesLayout问题
- 4. MYSQL8.0数据库恢复 MYSQL8.0ibd数据恢复 MYSQL8.0恢复数据库
- 5. 你本是一个肉体，是什么驱使你前行【1】
- 6. 2018.04.30
- 7. 2018.04.30
- 8. 你本是一个肉体，是什么驱使你前行【3】
- 9. 你本是一个肉体，是什么驱使你前行【2】
- 10. 【资讯】LocalBitcoins达到每周交易比特币的7年低点
- 1. Java并发之ThreadPoolExecutor源码解析（三）
- 2. Java并发之ThreadPoolExecutor源码解析（一）
- 3. java并发之ThreadPoolExecutor分析
- 4. Java并发之AQS源码分析（二）
- 5. java并发之线程池（ThreadPoolExecutor源码分析）
- 6. Java并发之线程池ThreadPoolExecutor源码分析学习
- 7. ThreadPoolExecutor源码解析
- 8. ThreadPoolExecutor 源码解析
- 9. Java线程池之ThreadPoolExecutor源码分析
- 10. Java源码解析 - ThreadPoolExecutor 线程池