线程池概述
多线程可以最大限度地发挥多核CPU的计算能力，提高生产系统的吞吐量和性能，但也会带来一些问题，比如：线程数量过大可能耗尽CPU资源；线程的创建和销毁开销比较大；线程本身占用内存空间，大量线程会抢占内存资源，可能会导致OOM，即便没有，大量的线程回收也会给GC带来很大压力。因此出现了线程池的概念，对线程进行复用。
Executors
java.util.concurrent包中，提供了ThreadPoolExecutor表示一个线程池，Executors表示线程工厂，通过Executors可以取得一个拥有特定功能的线程池，工厂方法的具体说明如下：
newFixedThreadPool()方法：该方法返回一个固定线程数量的线程池，该线程池中的线程数量始终不变。当有一个新的任务提交时，线程池中若有空闲线程，则立即执行。若没有，则新的任务会被暂存在一个任务队列中，待有线程空闲时，便处理在任务队列中的任务。代码实现如下：
[图片上传失败...(image-25e1b1-1597311845997)]
newSingleThreadExecutor()方法：该方法返回一个只有一个线程的线程池。若多余一个任务被提交到该线程池，任务会被保存在一个任务队列中，待线程空闲，按先进先出的顺序执行队列中的任务，实现代码如下：
在这里插入图片描述
newCachedThreadPool()方法：该方法返回一个可根据实际情况调整线程数量的线程池。线程池的线程数量不确定，但若有空闲线程可以复用，则会优先使用可复用的线程。若所有线程均在工作，又有新的任务提交，则会创建新的线程处理任务。所有线程在当前任务执行完毕后，将返回线程池进行复用，实现代码如下：
[图片上传失败...(image-ee4182-1597311845998)]
newSingleThreadScheduledExecutor()方法：该方法返回一个ScheduledExecutorService对象，线程池大小为1。ScheduledExecutorService接口在ExecutorService接口之上扩展了在给定事件执行某任务的功能，如在某个固定延时之后执行，或者周期性执行某个任务，实现代码如下：
[图片上传失败...(image-718a2a-1597311845998)]
newScheduledThreadPool()方法：该方法也返回一个ScheduledExecutorService对象，但该线程池可以指定线程数量，实现代码如下：
[图片上传失败...(image-e68208-1597311845998)]
构造参数
通过上面几个方法可以看出，其实创建的基本都是ThreadPoolExecutor，其中也提供了几个可配置的参数：
corePoolSize制定了线程池中的线程数量；
maximumPoolSize指定了线程池中的最大线程数量；
keepAliveTime设置当线程池线程数量超过corePoolSize时，多余的空闲线程的存活时间；
unit是keepAliveTime的单位；
workQueue是任务队列，被提交单尚未被执行的任务；threadFactory是线程工厂，用于创建线程；
handler设置拒绝策略，当任务太多来不及处理时，如何拒绝任务。构造方法实现如下：
在这里插入图片描述
任务队列
workQueue指被提交但未执行的任务队列，它是一个BlockingQueue接口的对象，仅用于存放Runnable对象，主要有如下几种队列：
直接提交队列：由SynchronousQueue对象提供，没有容量，每一个插入操作都要等待一个相应的删除操作，提交的任务不会被真实保存，将新任务提交给线程执行，如果没有空闲线程，则尝试创建新的线程，如果线程数量已经达到最大值，则执行拒绝策略，因此通常需要设置很大的maximumPoolSize值。
有界的任务队列：有界的任务队列使用ArrayBlockingQueue实现，构造参数必带一个容量参数，表示最大容量。当使用有界的任务队列时，有新的任务要执行，如果线程池的实际线程数小于corePoolSize，则会优先创建新的线程，若大于corePoolSize，则会将新任务加入等待队列，若等待队列已满，则在线程总数不大于maximumPoolSize的前提下，创建新的线程执行任务，若大于maximumPoolSize则执行拒绝策略。
无界任务队列：无界任务队列可以通过LinkedBlockingQueue类实现。除非资源耗尽，否则不存在任务入队失败的情况。新任务到来时，如果线程数小于corePoolSize，会创建新的线程执行任务，如果达到corePoolSize，就不会继续增长，新任务直接进入队列，任务过多则扩充队列。
优先任务队列：通过PriorityBlockingQueue实现，带有执行优先级的无界队列，可以根据自身任务优先级顺序先后执行。
淘汰策略
最后一个参数handler设置拒绝策略，JDK内置提供了四种拒绝策略如下：
AbortPolicy策略：该策略会直接抛出异常，阻止系统正常工作。
CallerRunsPolicy策略：只要线程池未关闭，该策略直接在调用者线程中，运行当前被丢弃的任务。
DiscardOledestPolicy策略：该策略会直接丢弃最老的一个请求，也就是即将被执行的一个任务，并尝试再次提交当前任务。
DiscardPolicy策略：该策略会丢弃无法处理的任务，不作任何处理。
源码分析
上文已经介绍了配置参数，这里描述下状态参数，代码如实现如下：
//状态字段，高三位作为线程状态的描述，低29位统计线程数量
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;
// 高三位标记运行状态
// 111 接收新任务，处理队列任务
private static final int RUNNING = -1 << COUNT_BITS;
// 000 不接收新任务，会处理队列剩余任务
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 001 中断正在执行的任务，抛弃阻塞队列任务
private static final int STOP = 1 << COUNT_BITS;
// 010 所有任务已经结束，任务线程为0
private static final int TIDYING = 2 << COUNT_BITS;
// 011 线程池关闭
private static final int TERMINATED = 3 << COUNT_BITS;
线程池的入口主要在execute方法，主要看一下execute方法的实现，代码实现如下：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
// 当前线程数小于corePoolSize，调用addWorker创建一个新的线程去执行任务
if (workerCountOf(c) < corePoolSize) {
// 添加成功后直接返回，失败后继续做后续的判断
if (addWorker(command, true))
return;
c = ctl.get();
}
// 如果线程池处于让running状态，尝试将其放入阻塞队列中
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 做二次校验，如果此时线程池状态变成了不可用，就把这个任务删除，执行拒绝策略
if (! isRunning(recheck) && remove(command))
reject(command);
// 如果线程池可用，但没有线程在运行，则创建一个新的线程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 尝试做大于coolPoolSize的扩容，如果不允许则执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}
execute中主要完成了接收新任务后的操作判断，如果线程池当前线程数量少于corePoolSize，则addWorker创建新worker线程，如创建成功返回，如没创建成功，则执行后续步骤；如果线程池还在running状态，将task加入workQueue阻塞队列中，如果加入成功，进二次校验，如果加入失败（可能是队列已满），则执行后续步骤；开始线程扩容，如果addWork(command, false)失败了，执行拒绝策略。
从上面的代码可以看到，创建新线程的工作主要在addWorker方法中，跟进到addWorker方法，实现代码如下：
private boolean addWorker(Runnable firstTask, boolean core) {
// retry:标记的下一行就是for循环，在for循环里面调用continue（或者break）
// 再紧接着retry标记时，就表示从这个地方开始执行continue（或者break）操作
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 根据线程池状态、firstTask和workQueue判断是否满足创建线程的条件
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
// 内存循环，worker数量 + 1
for (;;) {
// worker数量
int wc = workerCountOf(c);
// 如果worker数量大于线程池最大上限，返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 调用CAS操作，使的worker数量+1，成功则跳出retry循环
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();
// 如果状态不等于之前获取的state，跳出内层循环，继续去外层循环判断
if (runStateOf(c) != rs)
continue retry;
// else CAS失败时因为workerCount改变了，继续内层循环尝试CAS对worker数量+1
}
}
// worker数量+1成功后，添加到workers集合，启动worker线程
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 创建worker对象
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
// 上锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 上锁后做二次检查
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // 如果线程已经启动，抛出异常
throw new IllegalThreadStateException();
workers.add(w);
// 设置最大线程池代销，然后将workerAdded设置为true
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
// 解锁
mainLock.unlock();
}
// 成功加入worker集合后启动线程
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
// 如果启动线程失败
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
主要工作是判断线程状态，创建Worker对象，添加到workers集合中，线程数量+1，启动线程worker线程。可以看到，其中的核心是Worker对象，Worker是一个私有的内部类，继承了AQS，实现了Runnable接口，实现代码如下：
在这里插入图片描述
重点关注其run方法，run方法中只调用了runWorker方法执行任务，具体执行任务逻辑在runWorker方法中，实现代码如下：
在这里插入图片描述
代码中能比较清晰看到主流程，循环中获取任务然后直接执行其run方法，还留了beforeExecute预处理方法，和afterExecute后处理方法，都是空方法，留给子类自定义的扩展。如果因为异常导致线程终止，退出循环，执行processWorkerExit处理worker退出流程。此处主要展开介绍下getTask方法，实现代码如下：
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 如果线程池目前不能接受新任务并且任务队列是空或者线程池不再接受新任务，就减少当前的工作线程
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// //从队列中取任务的时候是否会等待一段时间过期
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//如果线程数量小于最大要求数量并且不设置核心线程过期就什么都不做
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// 如果可以等待过期了就直接调用Pull，否则调用take。
//调用take时候如果队列中没有任务的话就会一直阻塞知道有任务进来
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
内部也维护了一个for循环，做从阻塞队列中获取任务操作。
线程池常用的方法中除了execute方法外还有submit方法，会有一个返回值，实现如下：
在这里插入图片描述
可以看到其实内部调用的也是execute方法，只是维护了一个Future对象用于获取线程运行结果。
参考：
《实战Java高并发程序设计》
网友评论