Java线程池ThreadPoolExecutor源码分析
Java线程池ThreadPoolExecutor源码分析
用户5546570
发布于 2021-02-01 12:04:28
发布于 2021-02-01 12:04:28
文章被收录于专栏：以Java架构赢天下
继承关系
Java线程池ThreadPoolExecutor源码分析
Executor接口
public interface Executor {
void execute(Runnable command);
}ExecutorService接口
public interface ExecutorService extends Executor {
void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}ExecutorService接口继承Executor接口，并增加了submit、shutdown、invokeAll等等一系列方法。
AbstractExecutorService抽象类
public abstract class AbstractExecutorService implements ExecutorService {
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
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;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks, boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {... }
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {...}
}像execute方法、线程池的关闭方法（shutdown、shutdownNow等等）就没有提供默认的实现。
构造函数与线程池状态
public ThreadPoolExecutor(int corePoolSize, //核心线程数
int maximumPoolSize, //最大线程数
long keepAliveTime, //线程存活时间
TimeUnit unit, //keepAliveTime的单位
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;
}线程池状态
//记录线程池状态和线程数量（总共32位，前三位表示线程池状态，后29位表示线程数量）
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//线程数量统计位数29 Integer.SIZE=32
private static final int COUNT_BITS = Integer.SIZE - 3;
//容量 000 11111111111111111111111111111
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
//运行中 111 00000000000000000000000000000
private static final int RUNNING = -1 << COUNT_BITS;
//关闭 000 00000000000000000000000000000
private static final int SHUTDOWN = 0 << COUNT_BITS;
//停止 001 00000000000000000000000000000
private static final int STOP = 1 << COUNT_BITS;
//整理 010 00000000000000000000000000000
private static final int TIDYING = 2 << COUNT_BITS;
//终止 011 00000000000000000000000000000
private static final int TERMINATED = 3 << COUNT_BITS;
//获取运行状态（获取前3位）
private static int runStateOf(int c) { return c & ~CAPACITY; }
//获取线程个数（获取后29位）
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }int是4个字节,32位
RUNNING：接受新任务并且处理阻塞队列里的任务
SHUTDOWN：拒绝新任务但是处理阻塞队列里的任务
STOP：拒绝新任务并且抛弃阻塞队列里的任务同时会中断正在处理的任务
TIDYING：所有任务都执行完（包含阻塞队列里面任务），当前线程池活动线程为0，将要调用terminated方法
TERMINATED：终止状态。terminated方法调用完成以后的状态
线程池状态转换：
RUNNING -> SHUTDOWN：显式调用shutdown()方法, 或者隐式调用了finalize()方法
(RUNNING or SHUTDOWN) -> STOP：显式调用shutdownNow()方法
SHUTDOWN -> TIDYING：当线程池和任务队列都为空的时候
STOP -> TIDYING：当线程池为空的时候
TIDYING -> TERMINATED：当 terminated() hook 方法执行完成时候submit方法和execute方法的区别
submit方法
- 调用submit方法，传入Runnable或者Callable对象
- 判断传入的对象是否为null，为null则抛出异常，不为null继续流程
- 将传入的对象转换为RunnableFuture对象
- 执行execute方法，传入RunnableFuture对象
- 返回RunnableFuture对象
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
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;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}execute方法
public void execute(Runnable command) {
//传进来的线程为null，则抛出空指针异常
if (command == null)
throw new NullPointerException();
//获取当前线程池的状态+线程个数变量
int c = ctl.get();
/**
* 3个步骤
*/
//1.判断当前线程池线程个数是否小于corePoolSize,小于则调用addWorker方法创建新线程运行,
//且传进来的Runnable当做第一个任务执行。
//如果调用addWorker方法返回false，则直接返回
if (workerCountOf(c) < corePoolSize) {
//添加一个core线程(核心线程)。此处参数的true，表示添加的线程是core容量下的线程
if (addWorker(command, true))
return;
//刷新数据，乐观锁就是没有锁
c = ctl.get();
}
/* isRunning方法的定义：
private static boolean isRunning(int c)
{return c < SHUTDOWN;}
2.SHUTDOWN值为0，即如果c小于0，表示在运行；offer用来判断任务是否成功入队*/
if (isRunning(c) && workQueue.offer(command)) {
//二次检查
int recheck = ctl.get();
//如果当前线程池状态不是RUNNING则从队列删除任务，并执行拒绝策略
if (! isRunning(recheck) && remove(command))
//执行拒绝策略
reject(command);
//否则如果当前线程池线程空，则添加一个线程
else if (workerCountOf(recheck) == 0)
//添加一个空线程进线程池，使用非core容量线程
//仅有一种情况，会走这步，core线程数为0，max线程数>0,队列容量>0
//创建一个非core容量的线程，线程池会将队列的command执行
addWorker(null, false);
}
//线程池停止了或者队列已满，添加maximumPoolSize容量工作线程，如果失败，执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}ThreadPoolExecutor.addWorker()
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get(); //获取运行状态和工作数量
int rs = runStateOf(c); //获取当前线程池运行的状态
// Check if queue empty only if necessary.
//条件代表着以下几个场景，直接返回false说明当前工作线程创建失败
//1.rs>SHUTDOWN 此时不再接收新任务，且所有的任务已经执行完毕
//2.rs=SHUTDOWN 此时不再接收新任务，但是会执行队列中的任务
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
//先判断当前活动的线程数是否大于最大值，如果超过了就直接返回false说明线程创建失败
//如果没有超过再根据core的值再进行以下判断
//1\. core为true，则判断当前活动的线程数是否大于corePoolSize
//2\. core为false，则判断当前活动线程数是否大于maximumPoolSize
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//比较当前值是否和c相同，如果相同，则改为c+1，并且跳出大循环，直接执行Worker进行线程创建
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
//检查下当前线程池的状态是否已经发生改变
//如果已经改变了，则进行外层retry大循环，否则只进行内层的循环
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 {
//Worker的也是Runnable的实现类
w = new Worker(firstTask);
//因为不可以直接在Worker的构造方法中进行线程创建
//所以要把它的引用赋给t方便后面进行线程创建
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.add(w);//将创建的线程添加到workers容器中
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}Worker方法
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable{
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
}因此在调用t.start()执行的是（Worker类中的方法）：
/** Delegates main run loop to outer runWorker */
public void run() {
//这里执行的是ThreadPoolExecutor中的runWorker
runWorker(this);
}ThreadPoolExecutor.runWorker()
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;//获取Worker中的任务
w.firstTask = null; //将Woeker中的任务置空
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//如果当前任务为空 那么就从getTask中获得任务
/**
* 如果task不为空，执行完task后则将task置空
* 继续进入循环，则从getTask中获取任务
*/
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 interrupt
if ((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 {
//任务结束后调用的方法
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}从上面可以简单理解，就是执行任务，只是执行任务需要进行处理，包括获得任务、任务开始前处理、任务执行、任务执行后处理。但是，关键代码还是里面所调用的一个方法getTask() 。 beforeExecute(Thread t, Runnable r) 与 afterExecute(Runnable r, Throwable t) 并未在类中有处理业务的逻辑，即可以通过继承线程池的方式来重写这两个方法，这样就能够对任务的执行进行监控。
processWorkerExit
- 从While循环体中可以知道，当线程运行时出现异常，那么都会退出循环，进入到processWorkerExit()
- 从getTask()获得结果为null，则也会进到processWorkerExit()
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);
// Are workers subject to culling?
//如果设置了allowCoreThreadTimeOut(true)
//或者当前运行的任务数大于设置的核心线程数
// timed = true
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
/** ------------------------以上的操作跟之前类似----------------------- */
/** ------------------------关键在于下面的代码------------------------- */
/** ------------------------从阻塞队列中获取任务----------------------- */
try {
Runnable r = timed ?
//对于阻塞队列，poll(long timeout, TimeUnit unit) 将会在规定的时间内去任务
//如果没取到就返回null
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
//take会一直阻塞，等待任务的添加
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}线程池能够保证一直等待任务而不被销毁，其实就是进入了阻塞状态
ThreadPoolExecutor.processWorkerExit()
/**
* @param completedAbruptly
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) //如果突然被打断，工作线程数不会被减少
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
//判断运行状态是否在STOP之前
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {//正常退出，也就是task == null
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
//新增一个工作线程，代替原来的工作线程
addWorker(null, false);
}
}线程池关闭
只要调用了这两个关闭方法中的任意一个，isShutdown方法就会返回true。当所有的任务都已关闭后，才表示线程池关闭成功， 这时调用isTerminaed方法会返回true。至于应该调用哪一种方法来关闭线程池，应该由提交到线程池的任务特性决定， 通常调用shutdown方法来关闭线程池，如果任务不一定要执行完，则可以调用shutdownNow方法。
shutdown
当调用shutdown方法时，线程池将不会再接收新的任务，然后将先前放在队列中的任务执行完成。
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//检查权限
checkShutdownAccess();
//CAS 更新线程池状态
advanceRunState(SHUTDOWN);
//中断所有空闲的线程
interruptIdleWorkers();
//关闭，此处是do nothing
onShutdown();
} finally {
mainLock.unlock();
}
//尝试结束，上面代码已分析
tryTerminate();
}shutdownNow
立即停止所有的执行任务，并将队列中的任务返回
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);
//中断所有线程
interruptWorkers();
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}总结
- 线程池优先使用corePoolSize的数量执行工作任务
- 如果超过corePoolSize，队列入队
- 超过队列，使用maximumPoolSize-corePoolSize的线程处理，这部分线程超时不干活就销毁掉。
- 每个线程执行结束的时候，会判断当前的工作线程和任务数，如果任务数多，就会创建空线程从队列拿任务。
- 线程池执行完成，不会自动销毁，需要手工shutdown，修改线程池状态，中断所有线程。
分配线程池大小的依据
从以下几个角度考虑
- 任务的性质：CPU密集型任务、IO密集型任务和混合型任务。
- 任务的优先级：高、中和低。
- 任务的执行时间：长、中和短。
- 任务的依赖性：是否依赖其他系统资源，如数据库连接。
性质不同的任务可以用不同规模的线程池分开处理。CPU密集型任务应配置尽可能小的线程，如配置cpu个数 +1个线程的线程池。 由于IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如2*cpu个数 。混合型的任务，如果可以拆分， 将其拆分成一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐量 将高于串行执行的吞吐量。如果这两个任务执行时间相差太大，则没必要进行分解。可以通过 Runtime.getRuntime().availableProcessors()方法获得当前设备的CPU个数。 优先级不同的任务可以使用优先级队列PriorityBlockingQueue来处理。它可以让优先级高的任务先执行。
执行时间不同的任务可以交给不同规模的线程池来处理，或者可以使用优先级队列，让执行时间短的任务先执行。
依赖数据库连接池的任务，因为线程提交SQL后需要等待数据库返回结果，等待的时间越长，则CPU空闲时间就越长，那么线程数应该设置得越大， 这样才能更好地利用CPU。
使用有界队列
有界队列能增加系统的稳定性和预警能力，可以根据需要设大一点儿，比如几千。有一次，我们系统里后台任务线程池的队列和线程池全满了， 不断抛出抛弃任务的异常，通过排查发现是数据库出现了问题，导致执行SQL变得非常缓慢， 因为后台任务线程池里的任务全是需要向数据库查询和插入数据的，所以导致线程池里的工作线程全部阻塞，任务积压在线程池里。 如果当时我们设置成无界队列，那么线程池的队列就会越来越多，有可能会撑满内存，导致整个系统不可用，而不只是后台任务出现问题。 当然，我们的系统所有的任务是用单独的服务器部署的，我们使用不同规模的线程池完成不同类型的任务，但是出现这样问题时也会影响到其他任务。
线程池监控
如果在系统中大量使用线程池，则有必要对线程池进行监控，方便在出现问题时，可以根据线程池的使用状况快速定位问题。 可以通过线程池提供的参数进行监控，在监控线程池的时候可以使用以下属性。
- taskCount：线程池需要执行的任务数量。
- completedTaskCount：线程池在运行过程中已完成的任务数量，小于或等于taskCount。
- largestPoolSize：线程池里曾经创建过的最大线程数量。通过这个数据可以知道线程池是否曾经满过。如该数值等于线程池的最大大小， 则表示线程池曾经满过。
- getPoolSize：线程池的线程数量。如果线程池不销毁的话，线程池里的线程不会自动销毁，所以这个大小只增不减。
- getActiveCount：获取活动的线程数。
- 通过扩展线程池进行监控。可以通过继承线程池来自定义线程池，重写线程池的beforeExecute、afterExecute和terminated方法， 也可以在任务执行前、执行后和线程池关闭前执行一些代码来进行监控。例如，监控任务的平均执行时间、最大执行时间和最小执行时间等。 这几个方法在线程池里是空方法。
本文参与 腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。
如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录
相关产品与服务
数据库
云数据库为企业提供了完善的关系型数据库、非关系型数据库、分析型数据库和数据库生态工具。您可以通过产品选择和组合搭建，轻松实现高可靠、高可用性、高性能等数据库需求。云数据库服务也可大幅减少您的运维工作量，更专注于业务发展，让企业一站式享受数据上云及分布式架构的技术红利！
腾讯云开发者
