Java多线程 -- JUC包源码分析12 -- ThreadPoolExecutor源码分析

时间:2022-09-21 10:09:43

在JUC包中,线程池部分本身有很多组件,可以说是前面所分析的各种技术的一个综合应用。从本文开始,将综合前面的知识,逐个分析线程池的各个组件。
-Executor/Executors
-ThreadPoolExecutor使用介绍
-ThreadPoolExecutor实现原理
ThreadPoolExecutor的中断与优雅关闭 shutdown + awaitTermination
shutdown的一个误区


Executor/Executors

Executor是线程池框架最基本的几个接口:

public interface Executor {
void execute(Runnable command);
}

而Executors是线程池框架的一个工具类,利用它可以方便的创建不同策略的线程池:

//单线程线程池:corePoolSize = maxPoolSize = 1, 队列用的LinkedBlockingQueue
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
//固定数目的线程池:corePoolSize = maxPoolSize = n, 队列用的LinkedBlockingQueue
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
//1。CachedThreadPool,corePoolSize = 0, 队列为SynchronousQueue,maxPoolSize = Integer.MAX_VALUE(这也就意味着,每来一个任务,就创建一个线程。
//2。关于SynchronousQueue,后面会单独用一篇来分析。它是个特殊的队列,没本身没有容量,放进去一个,就得等有线程拿出来,才能解除阻塞
//3。从构造参数可以看出,空闲线程,60s没人用,回收
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
//单线程的,具有周期调度功能的线程池
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
}

//多线程的,具有周期调度功能的线程池
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
new DelayedWorkQueue());
}

从上面可以看出,Executors的各个工具函数,都用的ThreadPoolExecutor/ScheduledThreadPoolExecutor这2个类,下面做详细分析。

ThreadPoolExecutor

ThreadPoolExecutor构造函数详解

下面是ThreadPoolExecutor的参数最全的构造函数,搞清楚了每个参数的含义,也就明白了线程池的各种不同策略,也就明白了上述Executors工具类中的各个工具函数。

    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: 线程池始终维护的线程个数
maxPoolSize: corePooSize满了,队列也满的情况下,扩充线程至这个值
keepAliveTime/TimeUnit: maxPoolSize中的空闲线程,过多长时间销毁,总线程数收缩回corePoolSize
blockingQueue: 线程池所用的队列类型
threadFactory: 线程创建工厂,可以自定义,也有一个缺省的
RejectedExecutionHandler: corePoolSize满了,队列满了,maxPoolSize满了,最后的拒绝策略。

ThreadPool任务处理流程

从上述构造函数解释,可以看出每次submit的任务,有如下的处理流程:
step1: 判断当前线程数 >= corePoolSize。如果小于,新建线程执行;如果大于,进入step2
step2: 判断队列是否已满。未满,放入;已满,进入step3
step3: 判断当前线程数 >= maxPoolSize。如果小于,新建线程执行;如果大于,进入step4
step4: 根据拒绝策略,拒绝任务

总结一下:先判断corePoolSize, 再判断blockingQueue,再判断maxPoolSize,最后使用拒绝策略

ThreadPool的4中拒绝策略

ThreadPoolExecutor的4个内部类,分别定义了4种策略。缺省是AbortPolicy

//策略1:让调用者直接在自己的线程里面执行,线程池不做处理
public static class CallerRunsPolicy implements RejectedExecutionHandler {

public CallerRunsPolicy() { }

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}

//策略2:线程池直接抛异常
public static class AbortPolicy implements RejectedExecutionHandler {

public AbortPolicy() { }

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}


//策略3:线程池直接把任务丢掉,当作什么也没发生
public static class DiscardPolicy implements RejectedExecutionHandler {

public DiscardPolicy() { }

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}

//策略4:把队列里面最老的任务删除掉,把该任务放入队列
public static class DiscardOldestPolicy implements RejectedExecutionHandler {

public DiscardOldestPolicy() { }

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}

ThreadPoolExecutor实现原理

一般都知道,ThreadPool的基本实现原理就是一个队列 + 一组worker线程,调用中不断往队列中放,worker线程不断去取。但在具体实现中,有不同的实现策略:
策略1: 阻塞队列 vs. 非阻塞队列
在ThreadPoolExecutor中,使用的是阻塞队列,即如下的BlockingQueue接口:

private final BlockingQueue<Runnable> workQueue;

这也就意味着,worker内部不需要自己设置wait/notify机制,它只管从队列中取,取的到执行,取不到,自动会阻塞。

也有使用非阻塞队列的,比如Tomcat 6里面的线程池实现(以后会源码详细分析),当没有请求处理时,worker内部自己实现阻塞,然后又新的请求进来,再通知woker。

策略2:新来的请求,是直接放入队列,还是先new一个新的thread?
ThreadPool的处理方式是优先new thread处理,thread count >= corePoolSize的时候,再考虑放入队列。

策略3: *队列 vs. 有界队列?
如果*队列,意味着maxPoolSize的逻辑永远不会执行。这在上面的Executors中,FixedThreadPool已有所体现。

除此之外,还有诸多实现上的细节,下面代码详细分析

源码分析

//核心结构:一个BlockingQueue + 一个线程的Set + 一把锁(控制对workers, 各种threadCount的互斥访问)
public class ThreadPoolExecutor extends AbstractExecutorService {
。。。
private final BlockingQueue<Runnable> workQueue;

private final ReentrantLock mainLock = new ReentrantLock();

private final HashSet<Worker> workers = new HashSet<Worker>();
。。。
}
    public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) { //小于corePoolSize的判断
if (runState == RUNNING && workQueue.offer(command)) { //入队列
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command); //进入队列之后,2次检测
}
else if (!addIfUnderMaximumPoolSize(command)) //小于maxPoolSize的判断
reject(command); // 大于maxPoolSize,拒绝请求
}
}

//poolSize < corePoolSize的时候,直接new Thread,加入hashSet
private boolean addIfUnderCorePoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < corePoolSize && runState == RUNNING)
t = addThread(firstTask);
} finally {
mainLock.unlock();
}
return t != null;
}

//队列满了,poolSize < maxPoolSize,再次new thread,加入hashSet
private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < maximumPoolSize && runState == RUNNING)
t = addThread(firstTask);
} finally {
mainLock.unlock();
}
return t != null;
}

Worker的实现

    private final class Worker implements Runnable {
。。。
private Runnable firstTask; //至所以有firstTask这个变量,是因为创建worker的时候,可以直接赋给它一个task执行;有可以不赋给task,让它自己到blockingQueue里面去循环取
Worker(Runnable firstTask) {
this.firstTask = firstTask;
}

//1个死循环,不断从blockingQueue中,取task执行。取不到,就会阻塞在getTask()里面
public void run() {
try {
hasRun = true;
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) {
runTask(task);
task = null;
}
} finally {
workerDone(this); //worker线程退出
}
}
。。。
}

//getTask里面有个关键点:当poolSize <= corePoolSize时,是无限期阻塞下去,线程也就会一直存在,不会退出,死掉;当poolSize > corePoolSize或者允许coreThread也死去时,线程就只阻塞keepAliveTime的时间,时间到了,队列还是空的,没有请求,线程就退出,死掉了,同时poolSize--.

Runnable getTask() {
for (;;) {
try {
int state = runState;
if (state > SHUTDOWN)
return null;
Runnable r;
if (state == SHUTDOWN)
r = workQueue.poll(); //poll是非阻塞调用,没有直接返回null
else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS); //等待1个超时时间,默认就是构造函数里面传进去的那个60s
else
r = workQueue.take(); //take是阻塞调用,没有,一直阻塞
if (r != null)
return r;
if (workerCanExit()) {
if (runState >= SHUTDOWN)
interruptIdleWorkers();
return null;
}
} catch (InterruptedException ie) {
// On interruption, re-check runState
}
}
}

中断与优雅关闭

线程池状态切换图

Java多线程 -- JUC包源码分析12 -- ThreadPoolExecutor源码分析

    volatile int runState;
static final int RUNNING = 0;
static final int SHUTDOWN = 1;
static final int STOP = 2;
static final int TERMINATED = 3;

初始处于RUNNING状态,当调用shutdown()之后,切换到SHUTDOWN状态;调用shutdownNow(),切换到STOP状态。

那shutdown与shutdownNow有什么区别吗?
shutdown(): 不会清空队列里面的任务,会等所有任务执行完毕。并且它只会中断那些 > corePoolSize的idle线程

shutdownNow(): 清空队列里面所有任务,同时向所有线程发送中断信号

当队列为空 && pool也为空时,线程池进入Terminated状态。

shutdown/shutdownNow源码解析

 public void shutdown() {
SecurityManager security = System.getSecurityManager();
if (security != null)
security.checkPermission(shutdownPerm); //权限检查,check当前调用者,是否有权限关闭线程池。没有权限,抛出异常。

final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (security != null) {
for (Worker w : workers)
security.checkAccess(w.thread); //权限检查
}

int state = runState;
if (state < SHUTDOWN)
runState = SHUTDOWN; //从running切换到shutdown。不能从stop或者terminated切换到shutdown

try {
for (Worker w : workers) {
w.interruptIfIdle(); //遍历所有线程,向其发送信号
}
} catch (SecurityException se) {
runState = state;
throw se;
}

tryTerminate(); //试图终止线程池
} finally {
mainLock.unlock();
}
}

public List<Runnable> shutdownNow() {
SecurityManager security = System.getSecurityManager();
if (security != null)
security.checkPermission(shutdownPerm);

final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (security != null) {
for (Worker w : workers)
security.checkAccess(w.thread);
}

int state = runState;
if (state < STOP)
runState = STOP; //切换到stop状态

try {
for (Worker w : workers) {
w.interruptNow(); //变量所有线程,发中断信号,不管是否正在执行任务
}
} catch (SecurityException se) { // Try to back out
runState = state;
// tryTerminate() here would be a no-op
throw se;
}

List<Runnable> tasks = drainQueue(); //清空队列请求
tryTerminate(); // 试图终止线程池
return tasks;
} finally {
mainLock.unlock();
}
}

从上面,可以看出,shutdown和shutdownNow的区别有3点:
(1)一个是切换到shutdown状态,一个是切换到stop状态
(2)遍历所有线程,一个调用的interruptIfIdle, 一个调用的interruptNow。
(3)shutdownNow会清空队列中的任务

那interruptIfIdle和interruptNow有什么区别呢?

    private final class Worker implements Runnable {

。。。
private final ReentrantLock runLock = new ReentrantLock();

void interruptIfIdle() {
final ReentrantLock runLock = this.runLock;
if (runLock.tryLock()) {
try {
if (hasRun && thread != Thread.currentThread())
thread.interrupt();
} finally {
runLock.unlock();
}
}
}

void interruptNow() {
if (hasRun)
thread.interrupt();
}

public void run() {
try {
hasRun = true;
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) { //getTask内部,也有响应中断的逻辑
runTask(task);
task = null;
}
} finally {
workerDone(this);
}
}

//每次从队列中拿出一个任务,执行之前,会加锁
private void runTask(Runnable task) {
final ReentrantLock runLock = this.runLock;
runLock.lock();
try {
if ((runState >= STOP ||
(Thread.interrupted() && runState >= STOP)) &&
hasRun)
thread.interrupt();

boolean ran = false;
beforeExecute(thread, task);
try {
task.run();
ran = true;
afterExecute(task, null);
++completedTasks;
} catch (RuntimeException ex) {
if (!ran)
afterExecute(task, ex);
throw ex;
}
} finally {
runLock.unlock();
}
}

可以看出,interruptIfIdle和interuptNow的关键区别是:前者会加锁访问,这也就意味着,如果被中断的线程,正在执行runTask,则锁是拿不到的。此时shutdown会阻塞,直到woker执行完runTask。

shutdown的一个误区

根据上面分析,是不是shutdown一定会阻塞到队列中所有请求都执行完,再返回呢?或者说,shutdown返回的时候,是不是队列里面的请求就一定执行完了呢?

不一定!shutdown返回之后,线程池不一定立即关闭!为什么呢?

请看下面的getTask函数

    Runnable getTask() {
for (;;) {
try {
int state = runState;
if (state > SHUTDOWN)
return null;
Runnable r;
if (state == SHUTDOWN) //如果线程池是shutdown状态,就不阻塞了,不管是否能拿到,都是直接返回
r = workQueue.poll(); //关键点:如果是shutdown状态,会一直循环,直到拿空队列里面所有任务
else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
else
r = workQueue.take(); //case1: 别的线程先置了中断标志位,然后当前线程调用take //case 2: 先调用take阻塞在这,然后别的线程置了中断标志位 //2种case,都会抛出异常,进入下面的InterruptedException
if (r != null)
return r;
if (workerCanExit()) {
if (runState >= SHUTDOWN) // Wake up others
interruptIdleWorkers();
return null;
}
// Else retry
} catch (InterruptedException ie) { //阻塞的时候,收到中断,不处理,再次循环检查
// On interruption, re-check runState
}
}
}

public void run() {
try {
hasRun = true;
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) { //getTask内部,也有响应中断的逻辑
runTask(task);
task = null;
}
} finally {
workerDone(this);
}
}

//每次从队列中拿出一个任务,执行之前,会加锁
private void runTask(Runnable task) {
final ReentrantLock runLock = this.runLock;
runLock.lock();
try {
if ((runState >= STOP ||
(Thread.interrupted() && runState >= STOP)) &&
hasRun)
thread.interrupt();

boolean ran = false;
beforeExecute(thread, task);
try {
task.run();
ran = true;
afterExecute(task, null);
++completedTasks;
} catch (RuntimeException ex) {
if (!ran)
afterExecute(task, ex);
throw ex;
}
} finally {
runLock.unlock();
}
}

总结一下:当执行shutdown的时候,woker线程可能处于以下几种情况:
情况1: 正在执行runTask,此时拿着runLock锁,调用者会阻塞在shutdown上面。
情况2: 正要进入getTask。runTask执行完了,锁释放了,正要去getTask。此时shutdown不会阻塞,状态切换到shutdown状态,就返回了。 而getTask里面,会调用blockingQueue.poll
情况3: 在getTask里面,阻塞在blockQueue.take上面。此时调用shutdown, getTask里面收到中断,再次开始for(;;)循环

情况2,情况3,shutdown不会阻塞,就返回了。

所以不管是shutdown, 还是shutdownNow(),结尾都调用了tryTeminate,下面看看这个函数:

    private void tryTerminate() {
if (poolSize == 0) { //线程池里线程没了
int state = runState;
if (state < STOP && !workQueue.isEmpty()) {
state = RUNNING; //关键点:线程池里线程没了,状态是shutdown状态,队列还不为空,此时把状态切会到Running状态。并且重新创建线程,消化队列中的任务
addThread(null);
}
if (state == STOP || state == SHUTDOWN) {
runState = TERMINATED;
termination.signalAll(); //通知awaitTermination函数,不要再等了,线程池关闭
terminated();
}
}
}

所以,正确的使用shutdown的方式,应该是如下代码:

     executor.shutdown();//只是不能再提交新任务,等待执行的任务不受影响 

//调完shutdown,要循环调用awaitTermination,等待线程池真的终止
try {
boolean loop = true;
do { //等待所有任务完成
loop = !executor.awaitTermination(2, TimeUnit.SECONDS); //阻塞,直到线程池里所有任务结束
} while(loop);
} catch (InterruptedException e) {
e.printStackTrace();
}

总结:无论是shutdown,还是shutdownDown,都无法保证线程池立即关闭。他们的本质都只是切换了线程池的状态,发送了中断信号,然后等队列里面的任务为空了,所有线程自己销毁自己。

要让主线程等待线程池彻底终止,需要调用awaitTermination函数。

关于SechduledThreadPoolExecutor,会在接下来的篇章中,详细的单独阐述。