ThreadPoolExecutor是怎么去执行一个任务的？
@TOCjava
前言
前面一遍文章 咱们看了下FutureTask的源码，知道了怎么样去获取一个任务的返回值，今天咱们看下ThreadPoolExecutor。面试
ThreadPoolExecutor 看名词 咱们就能够 看作是ThreadPool 和Executor的结合，大概意思咱们也能知道就是线程池执行器，哈哈这翻译 真棒。这篇博文 会从源码的角度去分析下 一个线程任务 加入的线程池之后 是怎么被执行的~编程
线程池
上面 说线程的时候 咱们也说过 线程是系统中极其珍贵的资源，那咱们要合理的使用他，因此有了线程池的出现，那线程池能带来哪些好处呢安全
- 下降资源的消耗：经过重复利用已经建立的线程来下降线程建立和销毁带来的消耗
- 提供响应速度：当咱们建立人物到达的时候，任务能够不须要等待线程的建立就能当即执行
- 提升线程可管理性：线程是稀缺资源，不能无限建立，因此要使用线程池对线程进行同一的管理和分配，调优和监控等等。
源码分析
继承结构
首先 咱们看下ThreadPoolExecutor 的继承关系多线程
public class ThreadPoolExecutor extends AbstractExecutorService{}
public abstract class AbstractExecutorService implements ExecutorService{}
public interface ExecutorService extends Executor {
<!--中止线程池，状态设置为SHUTDOWN，而且不在接受新的任务，已经提交的任务会继续执行-->
void shutdown();
<!--中止线程池，状态设置为STOP，不在接受先任务，尝试中断正在执行的任务，返回还未执行的任务-->
List<Runnable> shutdownNow();
<!--是不是SHUTDOWN状态-->
boolean isShutdown();
<!--是否全部任务都已经终止-->
boolean isTerminated();
<!--超时时间内，去等待任务执行任务-->
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
<!--Callable 去提交任务-->
<T> Future<T> submit(Callable<T> task);
<!--Runnable 去提交任务-->
<T> Future<T> submit(Runnable task, T result);
<!--Runnable 去提交任务-->
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;
}
public interface Executor {
void execute(Runnable command);
}
复制代码
咱们先从最下面的接口Executor 来看，这个接口就是一个实现，就是执行execute方法，这个接口就是线程执行的入口并发
ExecutorService接口继承了Executor接口，里面的的方法比较多，咱们常见的shutdownNow，shutdown 就是在这个接口里面的，还有就是咱们常见往线程池里面提交任务的时候submit方法。ExecutorService丰富了对任务执行和管理的功能函数
AbstractExecutorService是一个抽象类，实现了ExecutorService接口，这边顺带说下，为何java 源码里面存在大量 抽象类实现接口，而后类再继承抽象类，为何类不直接实现接口呢？还要套一层呢，以前我也不明白，后来我才清楚，抽象类去实现接口，就是去实现一些公共的接口方法，这样类再次去实现接口的时候，只要关心我不一样的实现就行了，由于 咱们知道接口的实现类不止一个，抽象类就是把这些要实现接口的类的公共的实现再次抽取出来，避免了大量的重复实现，尤为List,Set 接口 你看下 几乎都有响应的抽象类实现！高并发
主要的变量
<!--ctl 存储了线程池状态和线程的数量-->
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;//32-3=29
private static final int CAPACITY = (1 << COUNT_BITS) - 1;//2的29次方-1
// runState is stored in the high-order bits
<!--表示线程池正在运行，能够接受任务 处理线程池中任务-->
private static final int RUNNING = -1 << COUNT_BITS;
<!--不接受新的任务，可是任然会处理队列中的任务-->
private static final int SHUTDOWN = 0 << COUNT_BITS;
<!--不接受新的任务，不会处理队里中任务，对正在执行的任务进行中断-->
private static final int STOP = 1 << COUNT_BITS;
<!--任务被中断，正在处理整理状态-->
private static final int TIDYING = 2 << COUNT_BITS;
<!--表示终结状态-->
private static final int TERMINATED = 3 << COUNT_BITS;
// Packing and unpacking ctl
<!--获取当前线程池的运行状态-->
private static int runStateOf(int c) { return c & ~CAPACITY; }
<!--获取当前线程池中工做线程的数量->
private static int workerCountOf(int c) { return c & CAPACITY; }
<!--获取ctl的值 ->
private static int ctlOf(int rs, int wc) { return rs | wc; }
复制代码
关于Ctl是怎么处理线程状态和线程数的数量的，能够看下个人另一篇博文：blog.csdn.net/zxlp520/art…oop
构造函数
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;
}
复制代码
这个构造函数 是全部构造函数最终调用的方法，那咱们就说下 这些具体的参数源码分析
- int corePoolSize 核心的线程数量
- int maximumPoolSize 最大的线程数量
- long keepAliveTime 线程存活的最大时间设置
- TimeUnit unit 设置时间的单位 和keepAliveTime是对应的
- BlockingQueue workQueue 阻塞队里，存储要执行的任务
- ThreadFactory threadFactory 建立执行线程的工厂 默认值：Executors.defaultThreadFactory()
- RejectedExecutionHandler handler 任务的拒绝Hander方法
- 默认的是AbortPolicy就是抛出异常，
- 还有三种策略是DiscardPolicy丢弃策略，DiscardOldestPolicy丢弃队列中等待时间最长的任务策略，CallerRunsPolicy这个是让调用的线程去处理的策略
Worker
为何要先讲worker呢？由于咱们提交的任务Runnabale是以Worker这个对象去包装后运行的，这个后面我我讲addWorker方法的时候在细聊
先看下Worker的代码：
/** Worker 继承了AQS 和实现了Runnable接口 */
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
private static final long serialVersionUID = 6138294804551838833L;
/** worker 运行的主体线程 就是在哪一个线程里面运行任务的 */
final Thread thread;
/** 须要运行的任务 */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
/** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);//这边的this 就是当前的Worker 对象
}
/** 运行 当前的任务 runWorker是ThreadPoolExecutor里面的方法 */
public void run() {
runWorker(this);
}
// Lock methods
// 0 表示 没有锁住状态
// 1 表示 锁住状态
protected boolean isHeldExclusively() {
return getState() != 0;
}
<!--这个方法咱们应该很熟悉 我在将AQS的时候聊过这个方法，这边作的就是尝试修改state的状态，这样就是表示加锁的意思，表示这个worker 是锁住状态，别的线程不能执行，-->
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {//CAS 去修改State的值，1表示 已经被上锁
setExclusiveOwnerThread(Thread.currentThread());设置当前锁的占用者线程是当前线程
return true;
}
return false;
}
<!--释放锁，也就是修改State的值 为0 unused这个字段命名也挺有意思，意思是说 没用的意思-->
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);//设置当前锁的占用者线程是null
setState(0);
return true;
}
<!--给当前的Worker加锁，若是获取不到 就加入等待队里中，阻塞当前执行线程-->
public void lock() { acquire(1); }
<!--这边至关于一个非公平锁的实现 去尝试下加锁-->
public boolean tryLock() { return tryAcquire(1); }
<!--释放锁-->
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
<!--尝试去中断运行的线程任务，就是咱们调用shutdownNow 的时候-->
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
复制代码
首先看下 这个Worker的继承结构，首先是实现了Runnable,又了这样的关系，Worker就能够被Thread去执行了，另一个还有一个继承了一个抽象类AbstractQueuedSynchronizer，简称AQS,这个类 哈哈 真的是好久不见了，我以前花了5篇文章解释了这个AQS,可想而知其重要性，JUC 中不少实现都是 基于这个去作的，仍是不清楚的小伙伴能够去到个人博客里面去找下。
这边又一行代码 咱们须要留意下，挺有意思的，this.thread = getThreadFactory().newThread(this);这边 的this 就是咱们构建的Worker,thread 就是用ThreadFactory去建立的一个线程而且执行的任务就是Worker,也就是调用thread.start()就能够执行Worker了
execute
execute是实现Executor接口的方法，就是执行的任务的入口方法，咱们看下一个任务的提交进来是怎么作的
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn't, by returning false. * * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. */
int c = ctl.get();//获取当前的ctl 值
/* * workerCountOf方法我上面也讲过，就是获取当前的工做线程数 * 若是当前的工做线程数小于设置的核心线程数量，就调用addWorker去新增一个工做线程，ture是表示要添加核心工做线程 * addWorker 若是添加成功就直接返回，若是添加失败就继续后去下ctl,这边重写获取是为了 防止在addWorker过程当中 ctl发生了改变 */
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
/* * 走到这步 说明当前的工做线程数大于核心线程数或者是addWorker发生了失败 * 首先去判断了下 当前的线程状态是不是Running 而后把当前任务加入到阻塞队列workQueue中 * 若是都成功了 那就再次获取下ctl,由于咱们在offer Runnable的时候可能ctl也会发生变化 *这边的多重验证 考虑到高并发的状况，代码逻辑很是的严谨 * 继续走下去的逻辑是 再次判断下线程池状态 若是是非Running，那就移除当前的任务,最后执行reject方法 根据不一样的拒绝策略，作不一样的行为 * 最后走到 判断当前线程数量若是是0，仍是回去调用addWorker方法，传入一个空的Runnalbe,false 是表示建立一个非核心的工做线程 */
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);
}
/* * 走到这个判断 说明当前线程池状态是非Running或者入队任务失败，队列多是满了 * 这边是去建立非核心线程去处理任务，若是建立失败 就执行拒绝策略 */
else if (!addWorker(command, false))
reject(command);
}
复制代码
这边的英文注释 我没舍得删除，读者能够去本身翻译下 描述的可能比我准确，我相信 你们能看的懂，而后再对比下 我下面的中文注释，我相信能清楚 一个任务新增进来 是怎么个处理流程！
看完本身再回想下，何时去建立核心线程？何时去建立非核心线程？何时任务会加入的阻塞队列中？最后执行拒绝策略 有那几种状况？知道这些答案 那么execute方法你应该了然于心了！
addWorker
下面咱们看下一个重点的方法，这个方法 调用的频次很高，咱们进入去看下
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
//这个是一个自旋 套了一个自旋 其目的就是CAS 新增线程池的数量
for (;;) {
int c = ctl.get();//获取ctl的值
int rs = runStateOf(c);//获取当前的线程状态
// 这边这个条件看上去很绕头，可是仔细看看就能知道
// 第一个条件rs >= SHUTDOWN 说明线程池状态不正常
// 后面有一个非的判断 其实就是括号里面的条件有一个不成立 整个条件就是false
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
// 下面是获取线程里面的工做线程 若是大于最大值或者设置的阈值，就返回直接返回false 方法结束
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//这个的意思是 若是CAS修改workerCount成功 整个最外层的自旋就结束
if (compareAndIncrementWorkerCount(c))
break retry;
// 这边为何要用2个自旋 主要是这边又判断了下 当前这个自旋CAS修改WorkerCount失败后，ctl会发生变化
//若是和外层的不相等，就要返回外层的自旋 去重写作
这边就是为何用的是 continue retry
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;//worker是否开始执行了
boolean workerAdded = false;//worker 是否添加成功
Worker w = null;
try {
w = new Worker(firstTask);//将Runnable 传入到worker的构造函数中，上面也讲过，其实就是用firstTask去构造了先的Thread
final Thread t = w.thread;//当前的t就是执行Runnable的线程，在worker中建立
if (t != null) {
final ReentrantLock mainLock = this.mainLock;//重入锁
mainLock.lock();//保证添加workder时候的线程安全
try {
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);//添加worker到一个工做worker集合中HashSet存储的
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();//释放锁
}
if (workerAdded) {//若是添加成功
t.start();//这个是真正执行Worker的地方 就是这儿
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);//若是最终Worker没有运行，那就清理掉他 修改对应的WorkerCount
}
return workerStarted;
}
复制代码
方法最开始的地方 用了2个自旋去解决并发状况下的CAS修改workerCount失败的状况，这边每一个细节，每种状况都考虑的很到位，状态判断的特别的严谨，真正看明白，感受多线程状况下的编程是多么的麻烦，辛亏帮咱们作了封装！
咱们看下 t.Start() 这边方法，咱们知道t就是Worker里面建立线程主体，是以本身为任务传入到Thread中的，咱们知道start是开始运行线程，最终是会调用到run方法的，那么就是说会调到Worker里面的run 方法，咱们在回看下Worker里面的run方法
public void run() {
runWorker(this);//ThreadPoolExecutor里面的方法
}
复制代码
runWorker
上面我也说了 线程start后会调用run方法，那么也就是调用 runWorker方法，咱们在看下这个里面写的时候什么
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;//获取Worker里面的任务
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//一直while循环
while (task != null || (task = getTask()) != null) {
w.lock();//锁住Worker
//判断若是当前的线程池状态是stop 而且检测当前线程的中断状态若是是false 就帮助当前线程执行中断调用interrupt()
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);//执行任务的前置Action
Throwable thrown = null;
try {
task.run();//执行最终的Runnable任务
} 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);///执行任务的后置Action
}
} finally {
task = null;
w.completedTasks++;//Worker完成的任务+1
w.unlock();//释放锁
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);//Worker执行结束后退出
}
}
复制代码
RunWorker方法是整个线程池运行任务的核心方法，线程会使用While循环 不断的从阻塞队里里面去获取任务，而后去执行任务，若是阻塞队列里面没有任务，这个时候 getTask() 方法就会阻塞线程，直至新任务的到来，因此咱们在作单元测试的时候，用到线程池，若是你不调用Shutdown 方法 ，你的debug 小红点就一直在运行，就是这个缘由！
getTask
这个方法就是从阻塞队列中取获取任务
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//判断线程池的状态若是是SHUTDOWN而且队列为空 或者直接状态就是null 就不会从阻塞队列中 取出任务 直接返回null
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
//timed 就是用来控制 获取阻塞队列中的任务 是否有等待时间，咱们设置的keepAliveTime值就会在这边用到，若是一个工做线程在等待任务超过了设置的值就会退出等待，回收线程
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))//工做线程数减1
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,调用shutdownnow的时候线程状态是Stop，那么这2种状态是怎么处理阻塞队列里面的任务的呢，看了上文咱们应该能找到答案，当状态是stop的时候，咱们获取队列中的任务是直接返回的null的也就是说队列中的任务不会在执行了，可是当状态是shutdown的时候 只有 队列为空的时候 才会返回null，也就是队列不空 仍是能够获取队列中的任务的，这种问题 在面试题中常常出现，若是要正在知道答案，仍是要经过从源码中去真正理解，光是被答案我相信你很快仍是会忘记的！
submit
掌握了execute方法 在看submit方法 其实就很简单了，submit通常是用于添加 带返回值的任务，咱们看下 代码
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);//将Runnable 包装成FutureTask任务 去让线程执行
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);//将Runnable 包装成FutureTask任务 去让线程执行
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;
}
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);
}
复制代码
看到这边的代码，应该有点儿熟悉的味道，应该上篇文章聊FutureTask的时候 不少已经将过了，包括Runnable和Callable怎么转换的，Future是怎么获取返回值的? 不清楚的小伙伴 能够去看下我以前的文章！blog.csdn.net/zxlp520/art…
上面三个构造函数，就是对应着FutureTask的构造函数，说白了就是咱们使用execute的时候都是用FutureTask去传入的，由于FutureTask也是实现了Runable接口的
执行流程图
最后 用一张流程图，来描述下一个任务从添加到运行结束，经历了哪些方法！
总结
ThreadPoolExecutor 虽然里面执行方法不少，可是你若是掌握了常见的逻辑运算符，AQS，线程，FutureTask 等相关知识的基础前提下 去看源码，也不会那么的累。最后我画的流程图，就是一个任务在新增到线程池中执行的整个流程！
最后分享下最近看到的一段话： 什么是危机？
真正的危机，来源于在正确的时间作不正确的事。没有在正确的时间，为下一步作出积累，这才是危机的根源。
若是你正在这条成长路上的朋友，晚醒不如早醒，这就是我想说的。千万别等到中年才发现本身没有创建好本身的护城河，这个时候才知道努力。在本身努力的阶段，不只不努力反了选择了纵容本身，这才是危机的根源。
但愿你们会有所收获，不负时光，不负卿！
相关文章
- 1. Java多线程-ThreadPoolExecutor是怎么去执行一个任务的？源码分析
- 2. ThreadPoolExecutor线程池任务执行失败的时候会怎样
- 3. ThreadpoolExecutor任务调度执行过程
- 4. 谈谈 Callable 任务是怎么运行的？它的执行结果又是怎么获取的?
- 5. 一个java文件是怎么一步一步执行的
- 6. 使用ThreadPoolExecutor并行执行独立的单线程任务
- 7. JavaScript 到底是怎样去执行的？
- 8. C语言的一个程序是怎么执行的
- 9. 定时任务去执行php
- 10. 任务并发执行是一个宏观概念，微观上是串行的。
- 更多相关文章...
- • Eclipse 任务管理 - Eclipse 教程
- • Hibernate是什么 - Hibernate教程
- • 互联网组织的未来：剖析GitHub员工的任性之源
- • 常用的分布式事务解决方案
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章