java多线程系列：ThreadPoolExecutor源码分析
前言
这篇主要讲述ThreadPoolExecutor的源码分析，贯穿类的建立、任务的添加到线程池的关闭整个流程，让你知其然因此然。但愿你能够经过本篇博文知道ThreadPoolExecutor是怎么添加任务、执行任务的，以及延伸的知识点。那么先来看看ThreadPoolExecutor的继承关系吧。java
继承关系
Executor接口
public interface Executor {
void execute(Runnable command);
}
Executor接口只有一个方法execute,传入线程任务参数git
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等等一系列方法。github
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 {...}
}
AbstractExecutorService抽象类实现ExecutorService接口，而且提供了一些方法的默认实现，例如submit方法、invokeAny方法、invokeAll方法。数据库
像execute方法、线程池的关闭方法（shutdown、shutdownNow等等）就没有提供默认的实现。编程
ThreadPoolExecutor
先介绍下ThreadPoolExecutor线程池的状态吧安全
线程池状态
int 是4个字节，也就是32位（注：一个字节等于8位）多线程
//记录线程池状态和线程数量（总共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; }
- 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 方法执行完成时候
构造函数
有四个构造函数，其余三个都是调用下面代码中的这个构造函数函数
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
}
参数介绍源码分析
| 参数 | 类型 | 含义 |
|---|---|---|
| corePoolSize | int | 核心线程数 |
| maximumPoolSize | int | 最大线程数 |
| keepAliveTime | long | 存活时间 |
| unit | TimeUnit | 时间单位 |
| workQueue | BlockingQueue
|
存放线程的队列 |
| threadFactory | ThreadFactory | 建立线程的工厂 |
| handler | RejectedExecutionHandler | 多余的的线程处理器（拒绝策略） |
提交任务
submit
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;
}
流程步骤以下
- 调用submit方法，传入Runnable或者Callable对象
- 判断传入的对象是否为null，为null则抛出异常，不为null继续流程
- 将传入的对象转换为RunnableFuture对象
- 执行execute方法，传入RunnableFuture对象
- 返回RunnableFuture对象
流程图以下
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) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//2.若是线程池处于RUNNING状态，则添加任务到阻塞队列
if (isRunning(c) && workQueue.offer(command)) {
//二次检查
int recheck = ctl.get();
//若是当前线程池状态不是RUNNING则从队列删除任务，并执行拒绝策略
if (! isRunning(recheck) && remove(command))
reject(command);
//否者若是当前线程池线程空，则添加一个线程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//3.新增线程，新增失败则执行拒绝策略
else if (!addWorker(command, false))
reject(command);
}
其实从上面代码注释中能够看出就三个判断，
- 核心线程数是否已满
- 队列是否已满
- 线程池是否已满
而后根据这三个条件进行不一样的操做，下图是Java并发编程的艺术书中的线程池的主要处理流程，或许会比较容易理解些
下面是整个流程的详细步骤
- 调用execute方法，传入Runable对象
- 判断传入的对象是否为null，为null则抛出异常，不为null继续流程
- 获取当前线程池的状态和线程个数变量
- 判断当前线程数是否小于核心线程数，是走流程5，不然走流程6
- 添加线程数，添加成功则结束，失败则从新获取当前线程池的状态和线程个数变量,
- 判断线程池是否处于RUNNING状态，是则添加任务到阻塞队列，不然走流程10，添加任务成功则继续流程7
- 从新获取当前线程池的状态和线程个数变量
- 从新检查线程池状态，不是运行状态则移除以前添加的任务，有一个false走流程9，都为true则走流程11
- 检查线程池线程数量是否为0，不然结束流程，是调用addWorker(null, false)，而后结束
- 调用!addWorker(command, false)，为true走流程11，false则结束
- 调用拒绝策略reject(command)，结束
可能看上面会有点绕，不清楚的能够看下面的流程图
addWorker
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 检查当前线程池状态是不是SHUTDOWN、STOP、TIDYING或者TERMINATED
// 且！（当前状态为SHUTDOWN、且传入的任务为null，且队列不为null）
// 条件都成立则返回false
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
//循环
for (;;) {
int wc = workerCountOf(c);
//若是当前的线程数量超过最大容量或者大于（根据传入的core决定是核心线程数仍是最大线程数）核心线程数 || 最大线程数，则返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//CAS增长c，成功则跳出retry
if (compareAndIncrementWorkerCount(c))
break retry;
//CAS失败执行下面方法，查看当前线程数是否变化，变化则继续retry循环，没变化则继续内部循环
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
}
}
//CAS成功
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//新建一个线程
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
//加锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//从新检查线程池状态
//避免ThreadFactory退出故障或者在锁获取前线程池被关闭
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // 先检查线程是不是可启动的
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
//判断worker是否添加成功，成功则启动线程，而后将workerStarted设置为true
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
//判断线程有没有启动成功，没有则调用addWorkerFailed方法
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
这里能够将addWorker分为两部分，第一部分增长线程池个数，第二部分是将任务添加到workder里面并执行。
第一部分主要是两个循环，外层循环主要是判断线程池状态，下面描述来自Java中线程池ThreadPoolExecutor原理探究
rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())展开！运算后等价于
s >= SHUTDOWN && (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty())也就是说下面几种状况下会返回false：
- 当前线程池状态为STOP，TIDYING，TERMINATED
- 当前线程池状态为SHUTDOWN而且已经有了第一个任务
- 当前线程池状态为SHUTDOWN而且任务队列为空
内层循环做用是使用cas增长线程个数，若是线程个数超限则返回false，否者进行cas，cas成功则退出双循环，否者cas失败了，要看当前线程池的状态是否变化了，若是变了，则从新进入外层循环从新获取线程池状态，否者进入内层循环继续进行cas尝试。
到了第二部分说明CAS成功了，也就是说线程个数加一了，可是如今任务还没开始执行，这里使用全局的独占锁来控制workers里面添加任务，其实也可使用并发安全的set，可是性能没有独占锁好（这个从注释中知道的）。这里须要注意的是要在获取锁后从新检查线程池的状态，这是由于其余线程可可能在本方法获取锁前改变了线程池的状态，好比调用了shutdown方法。添加成功则启动任务执行。
因此这里也将流程图分为两部分来描述
第一部分流程图
第二部分流程图
Worker对象
Worker是定义在ThreadPoolExecutor中的finnal类，其中继承了AbstractQueuedSynchronizer类和实现Runnable接口，其中的run方法以下
public void run() {
runWorker(this);
}
线程启动时调用了runWorker方法，关于类的其余方面这里就不在叙述。
runWorker
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock();
boolean completedAbruptly = true;
try {
//循环获取任务
while (task != null || (task = getTask()) != null) {
w.lock();
// 当线程池是处于STOP状态或者TIDYING、TERMINATED状态时，设置当前线程处于中断状态
// 若是不是，当前线程就处于RUNNING或者SHUTDOWN状态，确保当前线程不处于中断状态
// 从新检查当前线程池的状态是否大于等于STOP状态
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;
//统计当前worker完成了多少个任务
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
//整个线程结束时调用，线程退出操做。统计整个线程池完成的任务个数之类的工做
processWorkerExit(w, completedAbruptly);
}
}
getTask
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;
//（当前线程数是否大于最大线程数或者）
//且（线程数大于1或者任务队列为空）
//这里有个问题(timed && timedOut)timedOut = false，好像(timed && timedOut)一直都是false吧
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//获取任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
关闭线程池
shutdown
当调用shutdown方法时，线程池将不会再接收新的任务，而后将先前放在队列中的任务执行完成。
下面是shutdown方法的源码
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} 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;
}
shutdown和shutdownNow区别
shutdown和shutdownNow这两个方法的做用都是关闭线程池，流程大体相同，只有几个步骤不一样，以下
- 加锁
- 检查关闭权限
- CAS改变线程池状态
- 设置中断标志(线程池不在接收任务，队列任务会完成)/中断当前执行的线程
- 调用onShutdown方法（给子类提供的方法）/获取队列中的任务
- 解锁
- 尝试将线程池状态变成终止状态TERMINATED
- 结束/返回队列中的任务
总结
线程池能够给咱们多线程编码上提供极大便利，就好像数据库链接池同样，减小了线程的开销，提供了线程的复用。并且ThreadPoolExecutor也提供了一些未实现的方法，供咱们来使用，像beforeExecute、afterExecute等方法，咱们能够经过这些方法来对线程进行进一步的管理和统计。
在使用线程池上好须要注意，提交的线程任务能够分为CPU 密集型任务和IO 密集型任务，而后根据任务的不一样进行分配不一样的线程数量。
- CPU密集型任务：
- 应当分配较少的线程，好比
CPU个数至关的大小
- 应当分配较少的线程，好比
- IO 密集型任务：
- 因为线程并非一直在运行，因此能够尽量的多配置线程，好比 CPU 个数 * 2
- 混合型任务：
- 能够将其拆分为
CPU密集型任务以及IO密集型任务，这样来分别配置。
- 能够将其拆分为
好了，这篇博文到这里就结束了，文中可能会有些纰漏，欢迎留言指正。
若是本文对你有所帮助，给个star呗，谢谢。本文GitHub地址：点这里点这里
参考资料
- 并发编程网-Java中线程池ThreadPoolExecutor原理探究
- Java并发编程的艺术
相关文章
- 1. java多线程系列：ThreadPoolExecutor
- 2. Java线程池：ThreadPoolExecutor源码分析
- 3. Java线程池ThreadPoolExecutor源码分析
- 4. Java线程池之ThreadPoolExecutor源码分析
- 5. Java ThreadPoolExecutor 线程池源码分析
- 6. java ThreadPoolExecutor线程池源码分析
- 7. Java 多线程线程池-ThreadPoolExecutor分析
- 8. Java线程池ThreadPoolExecutor类源码分析 java线程池ThreadPoolExecutor类使用详解
- 9. Java源码解析 - ThreadPoolExecutor 线程池
- 10. java线程池——ThreadPoolExecutor源码解析
- 更多相关文章...
- • C# 多线程 - C#教程
- • Hibernate多对多映射关系 - Hibernate教程
- • Java Agent入门实战（二）-Instrumentation源码概述
- • Git五分钟教程
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. java多线程系列：ThreadPoolExecutor
- 2. Java线程池：ThreadPoolExecutor源码分析
- 3. Java线程池ThreadPoolExecutor源码分析
- 4. Java线程池之ThreadPoolExecutor源码分析
- 5. Java ThreadPoolExecutor 线程池源码分析
- 6. java ThreadPoolExecutor线程池源码分析
- 7. Java 多线程线程池-ThreadPoolExecutor分析
- 8. Java线程池ThreadPoolExecutor类源码分析 java线程池ThreadPoolExecutor类使用详解
- 9. Java源码解析 - ThreadPoolExecutor 线程池
- 10. java线程池——ThreadPoolExecutor源码解析