线程池的底层实现你了解吗？线程池模块总结【上篇】
线程池的核心理念是什么？
由我的第一篇关于线程的文章中提到的，我们服务器的线程资源是宝贵且有限的，因此我们在某些场景下需要尽可能的复用线程，不让线程频繁的创建与销毁。因此就引出了我们线程池的第一个核心理念：线程资源复用。
那么除开线程资源复用之外，我们还希望能对线程资源的使用情况进行一些控制，例如最少要有几个线程在跑，最多能创建多少个线程。那么就引出了我们线程池第二个核心理念：线程资源控制。
线程池的核心功能是如何实现的呢？
线程资源复用的实现原理
为了彻底了解线程池的工作原理，肯定是要从源码开始说起的。本文选取的核心类是ThreadPoolExecutor，这个也是《Java并发编程之美》书中用来讲解的类。首先我们从核心函数 void execute(Runnable command) 开始看起。
下面的源码看起来实在枯燥，我还是在最前面先概述一下原理，让读者能有个大概印象，然后画图阐述下具体执行流程。至于下面的代码块，可以在各位读者阅读源码的时候，再来结合这里的说明进行理解。
线程池对核心线程的复用，主要是通过内部Worker线程+生产者消费者模式去实现的，传入的线程被内部Worker线程调用run函数执行。让内部的Worker线程去消费Task，并且长期存活，用以达到线程复用的效果。
而内部的Worker线程如何长期存活的呢？主要就是通过队列的take函数阻塞住Worker线程内部的while循环，并且在各个阶段去检查是否Worker线程数量小于了设置的核心线程数。 具体代码执行流程示意图如下：
源码说明如下：
// 这个函数中有详细的文档说明，读者可以自行阅读。这里主要是逐行写下我的理解。 public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
// 获取当前线程池的状态值
int c = ctl.get();
// 如果工作线程数量小于核心线程数就创建新的核心线程
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
// 如果所有的核心线程都在执行Task，那么就将新来的任务放到队列中
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);
}
// 如果队列也放不下了，就再次创建Worker线程，直到达到Max的限制
else if (!addWorker(command, false))
// 如果达到了Max限制，还有新的任务来，就执行对应的拒绝策略。
reject(command);
}
上面这一段代码，其实就是面试官常问的，线程池执行过程。我们可以从上面的代码中看到，比较重要的一个函数就是addWorker，顾名思义，这个函数就是用来创建新的内部Worker线程的。具体代码如下：
// 分成两部分： // 上半部分：通过CAS操作增加线程数（ctl）
// 下半部分：将传入的Task与Worker通过构造函数绑定，并添加到workers中，最后启动worker这个内部线程。
private boolean addWorker(Runnable firstTask, boolean core) {
// 上半部分：通过CAS操作增加线程数（ctl）
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
// 这里配合上面的execute最后一个分支去看，会发现这里如果达到了Max限制会返回false。
if (wc >= CAPACITY ||
wc >= (core ? 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
}
}
// 下半部分：将传入的Task与Worker通过构造函数绑定，并添加到workers中，最后启动worker这个内部线程。
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 绑定
w = new Worker(firstTask);
// 这里拿到的其实就是Worker本身，因为Worker实现了Runnable
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.
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// 添加到workers中
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 启动worker线程
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
上面这段代码很长，有很多保证线程安全的细节不用去在意，我们只需要理解它这个函数所做的主要的事情，一个就是通过CAS修改ctl状态值，第二个就是通过Worker的构造函数去将传入的Task与新创建的Worker进行绑定，然后添加到workers这个set中，最后启动这个Worker线程。下面我们看具体的构造函数：
Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker
// 将传入的Task赋值给自己的成员变量
this.firstTask = firstTask;
// 将自己变成一个thread并赋值给自己的成员变量
this.thread = getThreadFactory().newThread(this);
}
完成绑定之后，最后就会执行： t.start();启动内部的这个Worker线程，那么就会走到Worker内部的run函数，代码下：
final void runWorker(Worker w) { Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 控制内部Worker线程不被回收，生产者消费者模型
while (task != null || (task = getTask()) != null) {
w.lock();
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
// 显式调用run函数消费掉task
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);
}
}
可以根据上图知道，决定内部的Worker线程会不会被回收的，就是这里的while循环和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.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// 1、timed在worker数量大于core时为true
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
// 3、那么这里，就会因为timed && timedOut为true，然后队列为空，返回一个null，
// 让worker线程内部的循环退出。最终被回收掉。
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// 2、这里就会因为timed为true后，空闲了keepAliveTime后修改timedOut = true;
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
我们可以看到，如果当前Worker线程数量小于核心线程数，就会调用 workQueue.take();被阻塞住，不让Worker线程被清理掉。从而达到持久存在的目的。而最后就是在runWorker中的后置函数，主要是清理执行完成的Worker以及再次检查核心线程数。源码如下：
private void processWorkerExit(Worker w, boolean completedAbruptly) { if (completedAbruptly) // If abrupt, then workerCount wasn"t adjusted
decrementWorkerCount();
// 清理worker线程
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
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
}
// 如果当前worker数量小于核心线程数，那么在这里也调用addWorker去创建
// 最终Worker会调用take函数阻塞住而长久存在。
addWorker(null, false);
}
}
通过以上流程图以及源码的说明，我想各位读者应该可以清晰的了解到这个线程池底层的执行流程，我们也可以知道线程池如何实现的线程复用，为什么保证一定有corePoolsize数量的核心线程数，以及传入的线程是如何被调用（消费）的等等这些问题。
线程资源控制的实现原理
如果你能耐心的看完上面那个小节，我想这个资源控制的实现原理也应该有了一个具体的理解了。回想一下我们平时如何使用线程池的呢？设置一堆参数，让线程池最少要有多少个核心线程数；最多能有多少个工作线程数；能够空闲多少秒；放进什么类型的Queue中；以及如果任务处理不过来应该执行什么样的拒绝策略。
那么我们设置的这些参数其实就是对线程池资源的量化控制，而他们其中的一些参数，例如corePoolSize、maximumPoolSize、keepAliveTime等参数在什么时候生效，都在上文中被提到了，显而易见的，这些参数都是实例变量，也就是我们调用构造器的时候去指定的——在创建线程池的时候，通过构造器去指定对应的参数，然后线程池在运行过程中去不断检查线程池的运行状态，从而达到线程资源控制的目的。 最后贴上其中一个构造函数的代码：
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.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
这里代码就不加说明了，就是一系列的赋值操作。因为代码段实在太长了，看起来很费力，所以关于【线程池模块】会分成多篇进行总结，后续会再分析线程池是如何关闭的，有什么区别；以及拒绝策略的设置；理论上线程池参数该如何设置等等。
以上是 线程池的底层实现你了解吗？线程池模块总结【上篇】 的全部内容， 来源链接： utcz.com/z/514266.html
