深刻浅出让你理解Java线程池—ThreadPoolExecutor
几句闲扯：首先，我想说java的线程池真的是很绕，之前一直都感受新建几个线程一直不退出究竟是怎么实现的，也就有了后来学习ThreadPoolExecutor源码。学习源码的过程当中，最恶心的其实就是几种状态的转换了，这也是ThreadPoolExecutor的核心。花了将近小一周才大体的弄明白ThreadPoolExecutor的机制，遂记录下来。java
线程池有多重要
线程是一个程序员必定会涉及到的一个概念，可是线程的建立和切换都是代价比较大的。因此，咱们有没有一个好的方案能作到线程的复用呢？这就涉及到一个概念——线程池。合理的使用线程池可以带来3个很明显的好处：程序员
- 下降资源消耗：经过重用已经建立的线程来下降线程建立和销毁的消耗
- 提升响应速度：任务到达时不须要等待线程建立就能够当即执行。
- 提升线程的可管理性：线程池能够统一管理、分配、调优和监控。
java多线程池的支持——ThreadPoolExecutor
java的线程池支持主要经过ThreadPoolExecutor来实现，咱们使用的ExecutorService的各类线程池策略都是基于ThreadPoolExecutor实现的，因此ThreadPoolExecutor十分重要。要弄明白各类线程池策略，必须先弄明白ThreadPoolExecutor。安全
一、实现原理
首先看一个线程池的流程图：bash
- step1.调用ThreadPoolExecutor的execute提交线程，首先检查CorePool，若是CorePool内的线程小于CorePoolSize，新建立线程执行任务。
- step2.若是当前CorePool内的线程大于等于CorePoolSize，那么将线程加入到BlockingQueue。
- step3.若是不能加入BlockingQueue，在小于MaxPoolSize的状况下建立线程执行任务。
- step4.若是线程数大于等于MaxPoolSize，那么执行拒绝策略。
二、线程池的建立
线程池的建立能够经过ThreadPoolExecutor的构造方法实现：markdown
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param threadFactory the factory to use when the executor
* creates a new thread
* @param handler the handler to use when execution is blocked
* because the thread bounds and queue capacities are reached
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue}
* or {@code threadFactory} or {@code handler} is null
*/
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 核心线程池大小
- maximumPoolSize 线程池最大容量大小
- keepAliveTime 线程池空闲时，线程存活的时间
- TimeUnit 时间单位
- ThreadFactory 线程工厂
- BlockingQueue任务队列
- RejectedExecutionHandler 线程拒绝策略
三、线程的提交
ThreadPoolExecutor的构造方法如上所示，可是只是作一些参数的初始化，ThreadPoolExecutor被初始化好以后即可以提交线程任务，线程的提交方法主要是execute和submit。这里主要说execute，submit会在后续的博文中分析。less
/**
* Executes the given task sometime in the future. The task
* may execute in a new thread or in an existing pooled thread.
*
* If the task cannot be submitted for execution, either because this
* executor has been shutdown or because its capacity has been reached,
* the task is handled by the current {@code RejectedExecutionHandler}.
*
* @param command the task to execute
* @throws RejectedExecutionException at discretion of
* {@code RejectedExecutionHandler}, if the task
* cannot be accepted for execution
* @throws NullPointerException if {@code command} is null
*/
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. * 若是当前的线程数小于核心线程池的大小，根据现有的线程做为第一个Worker运行的线程， * 新建一个Worker，addWorker自动的检查当前线程池的状态和Worker的数量， * 防止线程池在不能添加线程的状态下添加线程 * * 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. * 若是线程入队成功，而后仍是要进行double-check的，由于线程池在入队以后状态是可能会发生变化的 * * 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. * * 若是task不能入队(队列满了)，这时候尝试增长一个新线程，若是增长失败那么当前的线程池状态变化了或者线程池已经满了 * 而后拒绝task */ int c = ctl.get(); //当前的Worker的数量小于核心线程池大小时，新建一个Worker。 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command))//recheck防止线程池状态的突变，若是突变，那么将reject线程，防止workQueue中增长新线程 reject(command); else if (workerCountOf(recheck) == 0)//上下两个操做都有addWorker的操做，可是若是在workQueue.offer的时候Worker变为0， //那么将没有Worker执行新的task，因此增长一个Worker. addWorker(null, false); } //若是workQueue满了，那么这时候可能还没到线程池的maxnum，因此尝试增长一个Worker else if (!addWorker(command, false)) reject(command);//若是Worker数量到达上限，那么就拒绝此线程 } 复制代码
这里须要明确几个概念：ide
- Worker和Task的区别，Worker是当前线程池中的线程，而task虽然是runnable，可是并无真正执行，只是被Worker调用了run方法，后面会看到这部分的实现。
- maximumPoolSize和corePoolSize的区别：这个概念很重要，maximumPoolSize为线程池最大容量，也就是说线程池最多能起多少Worker。corePoolSize是核心线程池的大小，当corePoolSize满了时，同时workQueue full（ArrayBolckQueue是可能满的） 那么此时容许新建Worker去处理workQueue中的Task，可是不能超过maximumPoolSize。超过corePoolSize以外的线程会在空闲超时后终止。
核心方法：addWorker Worker的增长和Task的获取以及终止都是在此方法中实现的，也就是这一个方法里面包含了不少东西。在addWorker方法中提到了Status的概念，Status是线程池的核心概念，这里咱们先看一段关于status的注释：oop
/**
* 首先ctl是一个原子量，同时它里面包含了两个field，一个是workerCount，另外一个是runState
* workerCount表示当前有效的线程数，也就是Worker的数量
* runState表示当前线程池的状态
* The main pool control state, ctl, is an atomic integer packing
* two conceptual fields
* workerCount, indicating the effective number of threads
* runState, indicating whether running, shutting down etc
*
* 二者是怎么结合的呢？首先workerCount是占据着一个atomic integer的后29位的，而状态占据了前3位
* 因此，workerCount上限是(2^29)-1。
* In order to pack them into one int, we limit workerCount to
* (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
* billion) otherwise representable. If this is ever an issue in
* the future, the variable can be changed to be an AtomicLong,
* and the shift/mask constants below adjusted. But until the need
* arises, this code is a bit faster and simpler using an int.
*
* The workerCount is the number of workers that have been
* permitted to start and not permitted to stop. The value may be
* transiently different from the actual number of live threads,
* for example when a ThreadFactory fails to create a thread when
* asked, and when exiting threads are still performing
* bookkeeping before terminating. The user-visible pool size is
* reported as the current size of the workers set.
*
* runState是整个线程池的运行生命周期，有以下取值：
* 1. RUNNING：能够新加线程，同时能够处理queue中的线程。
* 2. SHUTDOWN：不增长新线程，可是处理queue中的线程。
* 3.STOP 不增长新线程，同时不处理queue中的线程。
* 4.TIDYING 全部的线程都终止了（queue中），同时workerCount为0，那么此时进入TIDYING
* 5.terminated()方法结束，变为TERMINATED
* The runState provides the main lifecyle control, taking on values:
*
* RUNNING: Accept new tasks and process queued tasks
* SHUTDOWN: Don't accept new tasks, but process queued tasks * STOP: Don't accept new tasks, don't process queued tasks, * and interrupt in-progress tasks * TIDYING: All tasks have terminated, workerCount is zero, * the thread transitioning to state TIDYING * will run the terminated() hook method * TERMINATED: terminated() has completed * * The numerical order among these values matters, to allow * ordered comparisons. The runState monotonically increases over * time, but need not hit each state. The transitions are: * 状态的转化主要是： * RUNNING -> SHUTDOWN（调用shutdown()） * On invocation of shutdown(), perhaps implicitly in finalize() * (RUNNING or SHUTDOWN) -> STOP(调用shutdownNow()) * On invocation of shutdownNow() * SHUTDOWN -> TIDYING（queue和pool均empty） * When both queue and pool are empty * STOP -> TIDYING（pool empty，此时queue已经为empty） * When pool is empty * TIDYING -> TERMINATED(调用terminated()) * When the terminated() hook method has completed * * Threads waiting in awaitTermination() will return when the * state reaches TERMINATED. * * Detecting the transition from SHUTDOWN to TIDYING is less * straightforward than you'd like because the queue may become
* empty after non-empty and vice versa during SHUTDOWN state, but
* we can only terminate if, after seeing that it is empty, we see
* that workerCount is 0 (which sometimes entails a recheck -- see
* below).
*/
复制代码
下面是状态的代码：学习
//利用ctl来保证当前线程池的状态和当前的线程的数量。ps：低29位为线程池容量，高3位为线程状态。
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//设定偏移量
private static final int COUNT_BITS = Integer.SIZE - 3;
//肯定最大的容量2^29-1
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
//几个状态，用Integer的高三位表示
// runState is stored in the high-order bits
//111
private static final int RUNNING = -1 << COUNT_BITS;
//000
private static final int SHUTDOWN = 0 << COUNT_BITS;
//001
private static final int STOP = 1 << COUNT_BITS;
//010
private static final int TIDYING = 2 << COUNT_BITS;
//011
private static final int TERMINATED = 3 << COUNT_BITS;
//获取线程池状态，取前三位
// Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~CAPACITY; }
//获取当前正在工做的worker,主要是取后面29位
private static int workerCountOf(int c) { return c & CAPACITY; }
//获取ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }
复制代码
接下来贴上addWorker方法看看：
/**
* Checks if a new worker can be added with respect to current
* pool state and the given bound (either core or maximum). If so,
* the worker count is adjusted accordingly, and, if possible, a
* new worker is created and started running firstTask as its
* first task. This method returns false if the pool is stopped or
* eligible to shut down. It also returns false if the thread
* factory fails to create a thread when asked, which requires a
* backout of workerCount, and a recheck for termination, in case
* the existence of this worker was holding up termination.
*
* @param firstTask the task the new thread should run first (or
* null if none). Workers are created with an initial first task
* (in method execute()) to bypass queuing when there are fewer
* than corePoolSize threads (in which case we always start one),
* or when the queue is full (in which case we must bypass queue).
* Initially idle threads are usually created via
* prestartCoreThread or to replace other dying workers.
*
* @param core if true use corePoolSize as bound, else
* maximumPoolSize. (A boolean indicator is used here rather than a
* value to ensure reads of fresh values after checking other pool
* state).
* @return true if successful
*/
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
/**
* rs!=Shutdown || fistTask！=null || workCount.isEmpty
* 若是当前的线程池的状态>SHUTDOWN 那么拒绝Worker的add 若是=SHUTDOWN
* 那么此时不能新加入不为null的Task，若是在WorkCount为empty的时候不能加入任何类型的Worker，
* 若是不为empty能够加入task为null的Worker,增长消费的Worker
*/
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
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
}
}
Worker w = new Worker(firstTask);
Thread t = w.thread;
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 c = ctl.get();
int rs = runStateOf(c);
/**
* rs!=SHUTDOWN ||firstTask!=null
*
* 一样检测当rs>SHUTDOWN时直接拒绝减少Wc，同时Terminate，若是为SHUTDOWN同时firstTask不为null的时候也要Terminate
*/
if (t == null ||
(rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null))) {
decrementWorkerCount();
tryTerminate();
return false;
}
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
} finally {
mainLock.unlock();
}
t.start();
// It is possible (but unlikely) for a thread to have been
// added to workers, but not yet started, during transition to
// STOP, which could result in a rare missed interrupt,
// because Thread.interrupt is not guaranteed to have any effect
// on a non-yet-started Thread (see Thread#interrupt).
//Stop或线程Interrupt的时候要停止全部的运行的Worker
if (runStateOf(ctl.get()) == STOP && ! t.isInterrupted())
t.interrupt();
return true;
}
复制代码
addWorker中首先进行了一次线程池状态的检测：
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
//判断当前线程池的状态是否是已经shutdown，若是shutdown了拒绝线程加入
//(rs!=SHUTDOWN || first!=null || workQueue.isEmpty())
//若是rs不为SHUTDOWN，此时状态是STOP、TIDYING或TERMINATED，因此此时要拒绝请求
//若是此时状态为SHUTDOWN，而传入一个不为null的线程，那么须要拒绝
//若是状态为SHUTDOWN，同时队列中已经没任务了，那么拒绝掉
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
复制代码
实际上是比较难懂的，主要在线程池状态判断条件这里：
- 若是是runing，那么跳过if。
- 若是rs>=SHUTDOWN,同时不等于SHUTDOWN，即为SHUTDOWN以上的状态，那么不接受新线程。
- 若是rs>=SHUTDOWN，同时等于SHUTDOWN，同时first！=null，那么拒绝新线程，若是first==null，那么多是新增长线程消耗Queue中的线程。可是同时还要检测workQueue是否isEmpty()，若是为Empty，那么队列已空，不须要增长消耗线程，若是队列没有空那么运行增长first=null的Worker。
从这里是能够看出一些策略的
- 首先，在rs>SHUTDOWN时，拒绝一切线程的增长，由于STOP是会终止全部的线程，同时移除Queue中全部的待执行的线程的，因此也不须要增长first=null的Worker了
- 其次，在SHUTDOWN状态时，是不能增长first！=null的Worker的，同时即便first=null，可是此时Queue为Empty也是不容许增长Worker的，SHUTDOWN下增长的Worker主要用于消耗Queue中的任务。
SHUTDOWN状态时，是不容许向workQueue中增长线程的，isRunning(c) && workQueue.offer(command) 每次在offer以前都要作状态检测，也就是线程池状态变为>=SHUTDOWN时不容许新线程进入线程池了。
for (;;) { int wc = workerCountOf(c); //若是当前的数量超过了CAPACITY，或者超过了corePoolSize和maximumPoolSize（试core而定） if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //CAS尝试增长线程数，若是失败，证实有竞争，那么从新到retry。 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 } 复制代码
这段代码作了一个兼容，主要是没有到corePoolSize 或maximumPoolSize上限时，那么容许添加线程，CAS增长Worker的数量后，跳出循环。 接下来实例化Worker,实例化Worker实际上是很关键的，后面会说。 由于workers是HashSet线程不安全的，那么此时须要加锁，因此mainLock.lock(); 以后从新检查线程池的状态，若是状态不正确，那么减少Worker的数量，为何tryTerminate（）目前不大清楚。若是状态正常，那么添加Worker到workers。最后：
if (runStateOf(ctl.get()) == STOP && ! t.isInterrupted()) t.interrupt(); 复制代码
注释说的很清楚，为了能及时的中断此Worker，由于线程存在未Start的状况，此时是不能响应中断的，若是此时status变为STOP，则不能中断线程。此处用做中断线程之用。 接下来咱们看Worker的方法：
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
复制代码
这里能够看出Worker是对firstTask的包装，而且Worker自己就是Runnable的，看上去真心很烦。 经过ThreadFactory为Worker本身构建一个线程。 由于Worker是Runnable类型的，因此是有run方法的,上面也看到了会调用t.start() 其实就是执行了run方法：
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
复制代码
调用了runWorker:
/**
* Main worker run loop. Repeatedly gets tasks from queue and
* executes them, while coping with a number of issues:
* 1 Worker可能仍是执行一个初始化的task——firstTask。
* 可是有时也不须要这个初始化的task（能够为null）,只要pool在运行，就会
* 经过getTask从队列中获取Task，若是返回null，那么worker退出。
* 另外一种就是external抛出异常致使worker退出。
* 1. We may start out with an initial task, in which case we
* don't need to get the first one. Otherwise, as long as pool is * running, we get tasks from getTask. If it returns null then the * worker exits due to changed pool state or configuration * parameters. Other exits result from exception throws in * external code, in which case completedAbruptly holds, which * usually leads processWorkerExit to replace this thread. * * * 2 在运行任何task以前，都须要对worker加锁来防止other pool中断worker。 * clearInterruptsForTaskRun保证除了线程池stop，那么现场都没有中断标志 * 2. Before running any task, the lock is acquired to prevent * other pool interrupts while the task is executing, and * clearInterruptsForTaskRun called to ensure that unless pool is * stopping, this thread does not have its interrupt set. * * 3. Each task run is preceded by a call to beforeExecute, which * might throw an exception, in which case we cause thread to die * (breaking loop with completedAbruptly true) without processing * the task. * * 4. Assuming beforeExecute completes normally, we run the task, * gathering any of its thrown exceptions to send to * afterExecute. We separately handle RuntimeException, Error * (both of which the specs guarantee that we trap) and arbitrary * Throwables. Because we cannot rethrow Throwables within * Runnable.run, we wrap them within Errors on the way out (to the * thread's UncaughtExceptionHandler). Any thrown exception also
* conservatively causes thread to die.
*
* 5. After task.run completes, we call afterExecute, which may
* also throw an exception, which will also cause thread to
* die. According to JLS Sec 14.20, this exception is the one that
* will be in effect even if task.run throws.
*
* The net effect of the exception mechanics is that afterExecute
* and the thread's UncaughtExceptionHandler have as accurate * information as we can provide about any problems encountered by * user code. * * @param w the worker */ final void runWorker(Worker w) { Runnable task = w.firstTask; w.firstTask = null; //标识线程是否是异常终止的 boolean completedAbruptly = true; try { //task不为null状况是初始化worker时，若是task为null，则去队列中取线程--->getTask() while (task != null || (task = getTask()) != null) { w.lock(); //获取woker的锁，防止线程被其余线程中断 clearInterruptsForTaskRun();//清楚全部中断标记 try { beforeExecute(w.thread, task);//线程开始执行以前执行此方法，能够实现Worker未执行退出，本类中未实现 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);//线程执行后执行，能够实现标识Worker异常中断的功能，本类中未实现 } } finally { task = null;//运行过的task标null w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { //处理worker退出的逻辑 processWorkerExit(w, completedAbruptly); } } 复制代码
从上面代码能够看出，execute的Task是被“包装 ”了一层，线程启动时是内部调用了Task的run方法。 接下来全部的核心集中在getTask()方法上：
/**
* Performs blocking or timed wait for a task, depending on
* current configuration settings, or returns null if this worker
* must exit because of any of:
* 1. There are more than maximumPoolSize workers (due to
* a call to setMaximumPoolSize).
* 2. The pool is stopped.
* 3. The pool is shutdown and the queue is empty.
* 4. This worker timed out waiting for a task, and timed-out
* workers are subject to termination (that is,
* {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
* both before and after the timed wait.
*
* @return task, or null if the worker must exit, in which case
* workerCount is decremented
*
*
* 队列中获取线程
*/
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
//当前状态为>stop时，不处理workQueue中的任务，同时减少worker的数量因此返回null，若是为shutdown 同时workQueue已经empty了，一样减少worker数量并返回null
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
boolean timed; // Are workers subject to culling?
for (;;) {
int wc = workerCountOf(c);
timed = allowCoreThreadTimeOut || wc > corePoolSize;
if (wc <= maximumPoolSize && ! (timedOut && timed))
break;
if (compareAndDecrementWorkerCount(c))
return null;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
复制代码
这段代码十分关键，首先看几个局部变量： boolean timedOut = false; 主要是判断后面的poll是否要超时 boolean timed; 主要是标识着当前Worker超时是否要退出。wc > corePoolSize时须要减少空闲的Worker数，那么timed为true，可是wc <= corePoolSize时，不能减少核心线程数timed为false。 timedOut初始为false，若是timed为true那么使用poll取线程。若是正常返回，那么返回取到的task。若是超时，证实worker空闲，同时worker超过了corePoolSize，须要删除。返回r=null。则 timedOut = true。此时循环到wc <= maximumPoolSize && ! (timedOut && timed)时，减少worker数，并返回null，致使worker退出。若是线程数<= corePoolSize，那么此时调用 workQueue.take()，没有线程获取到时将一直阻塞，知道获取到线程或者中断，关于中断后面Shutdown的时候会说。
至此线程执行过程就分析完了
关于终止线程池
我我的认为，若是想了解明白线程池，那么就必定要理解好各个状态之间的转换，想理解转换，线程池的终止机制是很好的一个途径。对于关闭线程池主要有两个方法shutdown()和shutdownNow(): 首先从shutdown()方法开始：
/**
* Initiates an orderly shutdown in which previously submitted
* tasks are executed, but no new tasks will be accepted.
* Invocation has no additional effect if already shut down.
*
* <p>This method does not wait for previously submitted tasks to
* complete execution. Use {@link #awaitTermination awaitTermination}
* to do that.
*
* @throws SecurityException {@inheritDoc}
*/
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//判断是否能够操做目标线程
checkShutdownAccess();
//设置线程池状态为SHUTDOWN,此处以后，线程池中不会增长新Task
advanceRunState(SHUTDOWN);
//中断全部的空闲线程
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
//转到Terminate
tryTerminate();
}
复制代码
shutdown作了几件事： 1. 检查是否能操做目标线程 2. 将线程池状态转为SHUTDOWN 3. 中断全部空闲线程
这里就引起了一个问题，什么是空闲线程？ 这须要接着看看interruptIdleWorkers是怎么回事。
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
//这里的意图很简单，遍历workers 对全部worker作中断处理。
// w.tryLock()对Worker加锁，这保证了正在运行执行Task的Worker不会被中断，那么能中断哪些线程呢？
try {
for (Worker w : workers) {
Thread t = w.thread;
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
复制代码
这里主要是为了中断worker，可是中断以前须要先获取锁，这就意味着正在运行的Worker不能中断。可是上面的代码有w.tryLock()，那么获取不到锁就不会中断，shutdown的Interrupt只是对全部的空闲Worker（正在从workQueue中取Task，此时Worker没有加锁）发送中断信号。
while (task != null || (task = getTask()) != null) { w.lock(); //获取woker的锁，防止线程被其余线程中断 clearInterruptsForTaskRun();//清楚全部中断标记 try { beforeExecute(w.thread, task);//线程开始执行以前执行此方法，能够实现Worker未执行退出，本类中未实现 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);//线程执行后执行，能够实现标识Worker异常中断的功能，本类中未实现 } } finally { task = null;//运行过的task标null w.completedTasks++; w.unlock(); } } 复制代码
在runWorker中，每个Worker getTask成功以后都要获取Worker的锁以后运行，也就是说运行中的Worker不会中断。由于核心线程通常在空闲的时候会一直阻塞在获取Task上，也只有中断才可能致使其退出。这些阻塞着的Worker就是空闲的线程（固然，非核心线程，而且阻塞的也是空闲线程）。在getTask方法中：
private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. //当前状态为>stop时，不处理workQueue中的任务，同时减少worker的数量因此返回null，若是为shutdown 同时workQueue已经empty了，一样减少worker数量并返回null if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } boolean timed; // Are workers subject to culling? for (;;) { //allowCoreThreadTimeOu是判断CoreThread是否会超时的，true为会超时，false不会超时。默认为false int wc = workerCountOf(c); timed = allowCoreThreadTimeOut || wc > corePoolSize; if (wc <= maximumPoolSize && ! (timedOut && timed)) break; if (compareAndDecrementWorkerCount(c)) return null; c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } try { Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } } 复制代码
会有两阶段的Worker：
- 刚进入getTask()，还没进行状态判断。
- block在poll或者take上的Worker。
当调用ShutDown方法时，首先设置了线程池的状态为ShutDown，此时1阶段的worker进入到状态判断时会返回null，此时Worker退出。 由于getTask的时候是不加锁的，因此在shutdown时能够调用worker.Interrupt.此时会中断退出，Loop到状态判断时，同时workQueue为empty。那么抛出中断异常，致使从新Loop，在检测线程池状态时，Worker退出。若是workQueue不为null就不会退出，此处有些疑问，由于没有看见中断标志位清除的逻辑，那么这里就会不停的循环直到workQueue为Empty退出。 这里也能看出来SHUTDOWN只是清除一些空闲Worker，而且拒绝新Task加入，对于workQueue中的线程仍是继续处理的。 对于shutdown中获取mainLock而addWorker中也作了mainLock的获取，这么作主要是由于Works是HashSet类型的，是线程不安全的，咱们也看到在addWorker后面也是对线程池状态作了判断，将Worker添加和中断逻辑分离开。 接下来作了tryTerminate()操做，这操做是进行了后面状态的转换，在shutdownNow后面说。 接下来看看shutdownNow：
/**
* Attempts to stop all actively executing tasks, halts the
* processing of waiting tasks, and returns a list of the tasks
* that were awaiting execution. These tasks are drained (removed)
* from the task queue upon return from this method.
*
* <p>This method does not wait for actively executing tasks to
* terminate. Use {@link #awaitTermination awaitTermination} to
* do that.
*
* <p>There are no guarantees beyond best-effort attempts to stop
* processing actively executing tasks. This implementation
* cancels tasks via {@link Thread#interrupt}, so any task that
* fails to respond to interrupts may never terminate.
*
* @throws SecurityException {@inheritDoc}
*/
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;
}
复制代码
shutdownNow和shutdown代码相似，可是实现却很不相同。首先是设置线程池状态为STOP，前面的代码咱们能够看到，是对SHUTDOWN有一些额外的判断逻辑，可是对于>=STOP,基本都是reject，STOP也是比SHUTDOWN更加严格的一种状态。此时不会有新Worker加入，全部刚执行完一个线程后去GetTask的Worker都会退出。 以后调用interruptWorkers：
/**
* Interrupts all threads, even if active. Ignores SecurityExceptions
* (in which case some threads may remain uninterrupted).
*/
private void interruptWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
try {
w.thread.interrupt();
} catch (SecurityException ignore) {
}
}
} finally {
mainLock.unlock();
}
}
复制代码
这里能够看出来，此方法目的是中断全部的Worker，而不是像shutdown中那样只中断空闲线程。这样体现了STOP的特色，中断全部线程，同时workQueue中的Task也不会执行了。因此接下来drainQueue：
/**
* Drains the task queue into a new list, normally using
* drainTo. But if the queue is a DelayQueue or any other kind of
* queue for which poll or drainTo may fail to remove some
* elements, it deletes them one by one.
*/
private List<Runnable> drainQueue() {
BlockingQueue<Runnable> q = workQueue;
List<Runnable> taskList = new ArrayList<Runnable>();
q.drainTo(taskList);
if (!q.isEmpty()) {
for (Runnable r : q.toArray(new Runnable[0])) {
if (q.remove(r))
taskList.add(r);
}
}
return taskList;
}
复制代码
获取全部没有执行的Task，而且返回。 这也体现了STOP的特色： 拒绝全部新Task的加入，同时中断全部线程，WorkerQueue中没有执行的线程所有抛弃。因此此时Pool是空的，WorkerQueue也是空的。 这以后就是进行到TIDYING和TERMINATED的转化了：
/**
* Transitions to TERMINATED state if either (SHUTDOWN and pool
* and queue empty) or (STOP and pool empty). If otherwise
* eligible to terminate but workerCount is nonzero, interrupts an
* idle worker to ensure that shutdown signals propagate. This
* method must be called following any action that might make
* termination possible -- reducing worker count or removing tasks
* from the queue during shutdown. The method is non-private to
* allow access from ScheduledThreadPoolExecutor.
*/
final void tryTerminate() {
for (;;) {
int c = ctl.get();
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
复制代码
上面的代码其实颇有意思有几种状态是不能转化到TIDYING的:
- RUNNING状态
- TIDYING或TERMINATED
- SHUTDOWN状态，可是workQueue不为空
也说明了两点： 1. SHUTDOWN想转化为TIDYING，须要workQueue为空，同时workerCount为0。 2. STOP转化为TIDYING，须要workerCount为0
若是知足上面的条件（通常必定时间后都会知足的），那么CAS成TIDYING，TIDYING也只是个过分状态，最终会转化为TERMINATED。
至此，ThreadPoolExecutor一些核心思想就介绍完了，想分析清楚实在是不容易，对于ThreadPoolExecutor我仍是有些不懂地方，以上只是我对源码的片面的看法，若是有不正确之处,但愿各位大佬指出
共同进步，学习分享
欢迎你们关注个人公众号【风平浪静如码】，海量Java相关文章，学习资料都会在里面更新，整理的资料也会放在里面。
以为写的还不错的就点个赞，加个关注呗！点关注，不迷路，持续更新！！！
相关文章
- 1. 深刻浅出Java线程池ThreadPoolExecutor
- 2. 深刻理解Java线程池：ThreadPoolExecutor
- 3. 深刻理解java线程池—ThreadPoolExecutor 全
- 4. 深刻浅出之java线程池
- 5. java线程池ThreadPoolExecutor理解
- 6. Java并发系列之7 深刻理解线程池ThreadPoolExecutor
- 7. 【Interview】深入理解线程池ThreadPoolExecutor
- 8. ThreadPoolExecutor 原理-- java 线程池
- 9. 深刻理解Java线程池
- 10. 深刻理解java之线程池
- 更多相关文章...
- • C# 多线程 - C#教程
- • SQLite - Java - SQLite教程
- • Java 8 Stream 教程
- • ☆技术问答集锦（13）Java Instrument原理
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章