Worker
先前,笔者讲解到ThreadPoolExecutor.addWorker(Runnable firstTask, boolean core),在这个方法中工作线程可能创建成功,也可能创建失败,具体视线程池的边界条件,以及当前内存情况而定。
那么,如果线程池当前的状态,是允许创建Worker对象的,那么创建Worker的内部流程又是怎样呢?线程池为何要使用Worker包装Thread来创建一个线程,为何不直接使用原生的Thread来创建线程?如果创建Worker的firstTask不为空,那么Worker理所当然应该优先执行firstTask任务,如果firstTask为空,那Worker又要如何获取任务来执行呢?我们还有一堆亟待解决的问题。
首先我们来解决前两个问题,Worker的创建流程,以及为什么不使用原生Thread代替Worker?首先,Doug Lea用Worker包装Thread,意味着Worker比Thread拥有更多的功能。例如:Worker会统计它所对应的线程执行了多少任务、通过Worker可以知道线程是否已启动、线程是否正在执行任务?而这些信息都是原生Thread所没有的,所以需要一个Worker类来扩展Thread。
在创建Worker时,会先设置其state的值为-1,代表Worker所对应的线程尚未启动,即还没有调用Worker.thread.start(),之后会进行firstTask的赋值,向线程工厂申请创建线程,创建完毕后,等待外部调用Worker.thread.start()启动一个线程执行Worker.run()方法。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /* * 当初始化一个Worker时,会向线程工厂申请创建一个Thread对象 * 用来执行任务,为null代表线程工厂创建失败。 */ final Thread thread; //首要执行任务,该字段可能为null。 Runnable firstTask; //thread已完成任务数。 volatile long completedTasks; Worker(Runnable firstTask) { /* * state初始为-1,代表还未调用Worker.thread.start(), * Worker对应的线程尚未被创建,还不能中断。线程启动后, * 如果线程正在执行任务,state为1,如果线程启动后没在 * 执行任务的状态则state为0。 */ setState(-1);//<1> this.firstTask = firstTask; /* * 创建Thread对象的时候,会把Worker对象本身传入,而Worker * 本身实现了Runnable接口,当调用thead.start()启动一个线程 * 执行thread.run()时,会进而调用Worker.run()方法。 */ this.thread = getThreadFactory().newThread(this); } //Worker对象将任务的执行委托给ThreadPoolExecutor.runWorker(Worker w). public void run() { runWorker(this); } //判断Worker是否处于被某个线程持有状态。 protected boolean isHeldExclusively() { return getState() != 0;//<2> } /* * 线程尝试持有worker对象,如果worker没有被某个线程 * 持有,则state为0,则用CAS的方式将worker的state * 改为1,并设置exclusiveOwnerThread为当前线程。 */ protected boolean tryAcquire(int unused) { if (compareAndSetState(0, 1)) {//<3> setExclusiveOwnerThread(Thread.currentThread());//<4> return true; } return false; } /* * 线程释放worker对象,将state改为0,exclusiveOwnerThread * 改为null。 */ protected boolean tryRelease(int unused) { setExclusiveOwnerThread(null); setState(0); return true; } /* * 调用父类acquire(int arg)时,会进而调用到Worker本身实现的 * tryAcquire(int unused)。 */ public void lock() { acquire(1);//<5> } public boolean tryLock() { return tryAcquire(1); } /* * 调用父类的release(int arg)时,会进而调用到Worker本身实现的 * tryRelease(int unused)。 */ public void unlock() { release(1);//<6> } public boolean isLocked() { return isHeldExclusively(); } /* * 尝试中断worker的对应线程,如果线程已经启动。创建 * worker时,state为-1,直到调用worker.thread.start() * 后,worker的state为0,如果worker的state>=0,则尝试 * 中断线程。 */ void interruptIfStarted() { Thread t; if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { try { t.interrupt(); } catch (SecurityException ignore) { } } } }
从上面的代码我们可以注意到,<1>、<2>、<3>、<4>、<5>、<6>处的方法并不是Worker本身有的方法,而是Worker继承自父类AbstractQueuedSynchronizer的方法。那么Worker为什么需要继承AbstractQueuedSynchronizer(AQS)?AQS又是何方神圣呢?
这里先简单介绍下AQS,它定义了若干接口交由程序员实现,诸如:lock()、unlock()、tryAcquire(int arg)、tryRelease(int arg)……等,以保证多个线程不会同时访问同一资源。ThreadPoolExecutor中的字段mainLock为可重入锁ReentrantLock,某种程度上来说也是实现了AQS,ThreadPoolExecutor通过mainLock的lock()、unlock()以保证线程池内一些非线程安全的对象不会出现并发读写,如:workers、completedTaskCount……等。
public class ReentrantLock implements Lock, java.io.Serializable { private final Sync sync; abstract static class Sync extends AbstractQueuedSynchronizer {//...} static final class NonfairSync extends Sync {//...} public ReentrantLock() { sync = new NonfairSync(); } //... }
那么Worker继承了AQS,Worker也实现了lock()、unlock()方法,说明Worker本身也存在多线程访问的可能,那是什么时候会出现多线程访问Worker呢?这里我们先按下这个问题,在介绍完Worker.run()之后你就会明白为何Worker要继承AQS以保证线程访问的顺序性。
我们知道当调用Worker.thread.start()方法时,会进而调用Worker.run()方法,而Worker.run()方法会进而调用ThreadPoolExecutor.runWorker(Worker w)。下面,我们来看看runWorker(Worker w)的执行流程:
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; /* * 将worker.firstTask设置为null,因为worker可能会存在较长的一段时间, * 而task可能很快执行完毕,避免worker长时间引用已完成task,以便GC回收 * 已完成task。 */ w.firstTask = null; /* * 新创建的worker对象state为-1,调用worker.unlock()会进而调用 * worker.tryRelease(int unused),将state设置为0,代表worker * 对应工作线程已启动,线程处于可中断状态。 */ w.unlock(); // allow interrupts /* * 标志worker是否异常完成,如果在<1>处task为空,且无法通过 * getTask()从任务队列中获取新的任务,则会跳出循环,并在<3>处 * 赋值为false,代表worker并没有异常完成。 * 不管worker是正常完成还是异常完成,最后都会将completedAbruptly的结果 * 传给processWorkerExit(...),如果是worker是正常退出,则将workerCount-1, * 并将worker从workers集合中移除。如果是异常退出,则不减少workerCount,仅仅 * 是将异常worker从workers集合中移除,并尝试新增一个worker。 */ boolean completedAbruptly = true; try { /* * 如果task不为空,或者调用getTask()能任务队列中获取到新的任务, * 则进入while块的代码。如果任务队列中没有待执行的任务,调用getTask() * 会让当前线程陷入阻塞,直到超时或者有新任务进入任务队列。 */ while (task != null || (task = getTask()) != null) {//<1> /* * 如果能进入循环,代表worker准备开始执行任务,但在执行任务 * 前会先上锁,等到任务执行结束又会在<2>处释放锁,然而线程池 * 又不会让多个线程同时执行同一个任务,那么为什么在执行任务前 * 要让worker先上锁,执行完毕再释放锁呢? * 我们假设有一个线程池有5个线程,其中A、B线程正在执行任务, * C、D、E处于空闲状态。所以我们能确定A、B两个worker已经 * 获得了锁,而C、D、E还阻塞在getTask()方法中。现在线程池 * 执行shutdown()方法,该方法会进而调用interruptIdleWorkers() * 中断处于空闲状态的工作线程。而在interruptIdleWorkers()方法中 * 判断一个worker是否处于空闲,会调用worker.tryLock(),如果能 * 成功获取到锁,则代表该worker处于空闲状态,则中断该worker对应的 * 线程。因此,一个worker是有可能被多个线程访问的,比如worker本身 * 对应的线程,又或者关闭线程池的线程。 */ w.lock(); /* * 如果线程池的运行状态>=STOP,则中断当前线程。如果运行状态<STOP, * 则确保线程没有被中断。 */ if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { //空方法,在执行任务前执行 beforeExecute(wt, task); try { task.run();//开始执行任务 afterExecute(task, null);//空方法,在执行任务完毕后执行。 } catch (Throwable ex) { afterExecute(task, ex); throw ex; } } finally { /* * 设置当前任务为空,这样就可以在下一次循环中获取新的任务。 * 对worker执行的任务数+1,并释放锁。 */ task = null; w.completedTasks++; w.unlock();//<2> } } completedAbruptly = false;//<3> } finally { processWorkerExit(w, completedAbruptly); } }
在上面的runWorker(Worker w)中如果执行完worker的首要任务,或者首要任务为null,便会调用getTask()尝试从任务队列中获取任务,但调用getTask()可能会使当前线程陷入等待或者阻塞直到有任务入队。getTask()也有可能返回null导致当前worker对应的线程退出,有以下几个原因可能导致工作线程退出:
- 如果调用setMaximumPoolSize(int maximumPoolSize)改小最大线程数,导致工作线程数大于maximumPoolSize。
- 线程池运行状态为STOP。
- 线程池运行状态为SHUTDOWN,且队列为空。
- 线程等待任务超时后,如果线程池当前存在可回收的空闲线程(即allowCoreThreadTimeOut为true或者工作线程数大于核心线程数),如果队列为空,则可直接退出,如果队列不为空,工作线程数必须大于1,即线程池中最少两个工作线程,如果只有一个线程还退出的话,就会存在队列不为空,但线程池中没有一个工作线程的尴尬情况。
private Runnable getTask() { //超时标志,默认为false,获取任务如果超时则会在<5>赋值为true。 boolean timedOut = false; for (; ; ) { int c = ctl.get(); /* * 如果线程池处于RUNNING状态,则runStateAtLeast(c, SHUTDOWN) * 为false,不会再判断之后的逻辑为true或者false。 * 如果线程池处于SHUTDOWN状态,且workQueue.isEmpty()为true,即 * 任务队列为空,则直接返回。如果任务队列不为空,则无法进入if分支, * 依然要返回任务,按照SHUTDOWN的要求不再接受新任务,但仍要处理队列 * 中的任务。 * 如果线程池处于STOP状态,即便任务队列不为空,也不再处理,则直接进入 * if分支后返回。 * 所以总结一下,只有两种情况不会进入此分支: * 1.线程池处于RUNNING状态。 * 2.线程池处于SHUTDOWN状态且任务队列不为空。 */ if (runStateAtLeast(c, SHUTDOWN) && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {//<1> decrementWorkerCount(); return null; } int wc = workerCountOf(c); /* * timed决定是如果任务队列没有任务的话,是以无限期的方式 * 等待任务入队,或者一旦等待时间超过keepAliveTime,则 * 返回null。 * 如果allowCoreThreadTimeOut为true,则核心线程等待 * 任务时间超过keepAliveTime后会被回收。 * 如果当前工作线程数量workerCount大于核心线程数,也会在 * 线程等待任务超过keepAliveTime后回收线程。 */ boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; /* * 我们先看进入此分支后会做的事,再分析如何进入这个分支。进入 * 此分支后,会用CAS减少workerCount的数量,成功则返回null。 * 否则continue重新开始新一轮的for循环。 * 现在,我们来分析下进入此分支的逻辑: * 首先是wc > maximumPoolSize,一般workerCount不会大于 * maximumPoolSize,除非线程池运行期间通过 * setMaximumPoolSize(int maximumPoolSize)将线程池 * 的最大线程数改小。 * (timed && timedOut)在第一轮for循环永远为false,因为 * timedOut要为true的条件,首先是timed为true,即线程池内存在空闲后 * 可回收的线程,不管是线程池允许回收核心线程,或者线程数大于核心线程数。 * 只有在<4>获取任务超时后workQueue返回null,才有可能到达<5>处将 * timedOut赋值为true,并且开始新一轮的循环。 * 之后的两个判断wc>1和workQueue.isEmpty(),判断队列为空还好理解, * 为什么要判断wc>1?首先我们要知道maximumPoolSize必须大于等于1,当我们 * 往线程池传入的maximumPoolSize<=0会抛出异常。其次,如果我们将<2>处改为: * (wc >= 1 || workQueue.isEmpty()),有可能出现线程池内只有一个线程, * 但任务队列不为null。依旧会进入此分支内部执行<3>的代码以CAS的方式对 * workerCount-1,从而出现一个尴尬的情况,任务队列中有任务,但工作线程数 * 为0。所以<2>处必须保证wc>1。 * 思考一种情况:假设一个线程池核心线程数为3,最大线程数为5, * allowCoreThreadTimeOut为false,线程池当前工作线程数量也为5,5个线程 * 同时完成任务,并执行getTask()获取任务,可想而知timed为true,因为工作 * 线程数(5)大于核心线程数(3),5个工作线程都是调用<4>处workQueue.poll(...) * 等待任务,超时则返回null。如果超时时间到达,3个核心线程如何重新进入等待状态, * 剩余2个线程如何被回收? * 当5个线程超时返回后,会将timedOut赋值为true,然后重新开始新一轮的for循环,一直 * 执行到此分支,此时(timed && timedOut)都为true,队列也都为null,所以5个线程会 * 进入此分支。用CAS成功对workerCount-1的线程将被回收,失败的线程则continue又开始 * 新一轮的for循环,直到wc<=corePoolSize,timed为false,最后剩余的工作线程数调用 * workQueue.take()无限期地等待任务的到来,除非线程被中断。 */ if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {//<2> if (compareAndDecrementWorkerCount(c))//<3> return null; continue; } try { /* * workQueue.poll(...)和workQueue.take()都有可能使当前线程陷入 * 等待,直到返回任务,只不过前者相比后者多了一个超时时间,到达超时时间 * 如果有任务入队,则r不为null,直接返回任务。 * 线程等待期间,如果线程被中断,则会抛出InterruptedException异常。 * 一般关闭线程池时,会尝试中断空闲线程,而处于等待任务的空闲线程会跳到<6>处, * 重新开始新一轮的for循环,并且在<1>处判断线程池处于SHUTDOWN或者STOP, * 从而退出。 */ Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) ://<4> workQueue.take(); if (r != null) return r; /* * 只有等待任务超时控制流才会执行到这里,将超时标志赋值为true, * 重新开始新一轮的for循环。 */ timedOut = true;//<5> } catch (InterruptedException retry) { timedOut = false;//<6> } } }
在runWorker(Worker w)中有两种方式可以进到下面的处理线程退出processWorkerExit(Worker w, boolean completedAbruptly)方法:
- 工作线程执行任务遇到异常,从而跳出while循环。
- 工作线程获取任务超时等待,正常退出循环。
上面两种方式传递给processWorkerExit(…)的completedAbruptly是不同的,第一个方式传入的completedAbruptly为true,第二个方式为false,虽然worker是同一个。那么当completedAbruptly为true或者false,processWorkerExit(…)的流程又是怎么走的呢?
private void processWorkerExit(Worker w, boolean completedAbruptly) { /* * 从runWorker(Worker w)传递而来的变量,标志worker是否意外完成, * 当worker执行任务时抛出异常,该变量为true。如果是意外完成,则表明 * workerCount尚未-1。如果worker获取任务超时从而要让线程被回收,在 * getTask()方法中会对workerCount-1。 */ if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { /* * 获取可重入锁后,将worker已完成的任务数加到线程池已完成任务数, * 并将worker从workers集合中移除。 */ completedTaskCount += w.completedTasks; workers.remove(w); } finally { mainLock.unlock(); } //尝试终止线程。 tryTerminate(); int c = ctl.get(); //如果线程池运行状态处于RUNNING或SHUTDOWN,则进入此分支。 if (runStateLessThan(c, STOP)) { //如果线程是正常退出,则进入此分支 if (!completedAbruptly) { /* * min为线程池创建核心线程后,允许最小的核心线程数,如果 * allowCoreThreadTimeOut为true则代表核心线程可以被回收, * 则min为0,否则min为核心线程数量。 * 如果线程池允许回收线程,且队列不为空,则判断在移除当前worker * 后,线程池工作线程的数量是否还大于等于1,避免出现队列里有任务 * 但没有线程执行的情况,如果工作线程数大于等于1,则退出线程线程。 * 否则调用addWorker(Runnable firstTask, boolean core) * 尝试增加新的线程。 */ int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && !workQueue.isEmpty()) min = 1; if (workerCountOf(c) >= min) return; // replacement not needed } addWorker(null, false); } }
最后,我们还需要介绍关闭线程池之后做的操作。关闭线程池会修改线程池运行状态,在advanceRunState(int targetState)会使用CAS自旋的方式,将线程池状态修改为SHUTDOWN。之后调用interruptIdleWorkers()中断空闲线程,这里我们看到中断的时候调用worker的tryLock()和unlock()。Worker之所以继承AQS就是为了方便区分哪些worker正在执行任务,哪些worker处于空闲中,以便在关闭线程池时中断所有空闲的worker。
/** * 调用此方法后不再接受新任务,但会执行现有队列任务。 * 如果调用此方法前该方法已被调用,则不会有任何效果。 * 该方法不会等待所有任务完成,需要调用: * awaitTermination(long timeout, TimeUnit unit) */ public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(SHUTDOWN);//设置线程池状态为SHUTDOWN interruptIdleWorkers();//中断空闲线程 onShutdown(); //钩子方法,按用户需要实现。 } finally { mainLock.unlock(); } /* * 尝试终止线程池,可能线程池有任务在执行,当前线程终止失败。 * 但随着工作线程逐个退出,最后一个工作线程将成功终止线程池。 */ tryTerminate(); } private void advanceRunState(int targetState) { for (; ; ) { int c = ctl.get(); /* * 如果线程池运行状态>=targetState,则 * runStateAtLeast(c, targetState)为true直接退出。 * 否则调用ctlOf(int rs, int wc),根据运行状态和工作 * 线程数生成的值以CAS自旋的方式set进ctl。 */ if (runStateAtLeast(c, targetState) || ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c)))) break; } } //调用shutdown()会进而调用此方法,中断空闲线程。 private void interruptIdleWorkers() { interruptIdleWorkers(false); } private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) { Thread t = w.thread; /* * 如果线程未被中断,则尝试获取worker的锁,如果能 * 成功获取,代表worker线程处于空闲中。 */ if (!t.isInterrupted() && w.tryLock()) { try { t.interrupt(); } catch (SecurityException ignore) { } finally { w.unlock(); } } //如果onlyOne为true代表最多只中断一个线程 if (onlyOne) break; } } finally { mainLock.unlock(); } }
在中断空闲线程后shutdown()还会调用tryTerminate(),如果查看tryTerminate()的引用,可以发现不单单shutdown()有调用,像addWorkerFailed(Worker w)、processWorkerExit(…)……都有调用。这个方法旨在终止线程池,如果当前有线程关闭了线程池,线程池如果没有存活线程,或者线程都处于空闲状态,自然而然执行尝试终止线程池方法;如果关闭线程池时,线程池仍然有线程处于执行任务状态,无法终止线程池,就要靠这些工作线程在退出时终止线程池。
final void tryTerminate() { for (; ; ) { int c = ctl.get(); /* * 如果线程池运行状态处于RUNNING、TIDYING则退出, * 或者运行状态小于STOP(即处于RUNNING、SHUTDOWN) * 且队列不为空,则退出。 * 总结一下,只有当运行状态处于STOP时或者状态处于SHUTDOWN * 但队列为空,才不会进此分支。 */ if (isRunning(c) || runStateAtLeast(c, TIDYING) ||//<1> (runStateLessThan(c, STOP) && !workQueue.isEmpty())) return; //如果工作线程数不为0,则尝试最多中断一个空闲线程后退出。 if (workerCountOf(c) != 0) { interruptIdleWorkers(ONLY_ONE); return; } /* * 在processWorkerExit(...)方法中如果worker是 * 异常退出会对workerCount-1,如果是正常退出,则 * workerCount在getTask()中-1。之后在processWorkerExit(...) * 中移除worker。 * 不论worker是正常退出还是异常退出,终归workerCount会慢慢回到0。 * 而processWorkerExit(...)中在对workerCount-1后,还会调用 * tryTerminate()。因此一个被关闭的线程池,它的最后一个线程,会 * 执行到此处,处理终止线程池的工作。 */ final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { /* * 用CAS设置线程池运行状态为TIDYING,可能存在多个线程同时 * 调用shutdown()后并依次执行到这一步(因为要获得可重入锁), * 但只有一个线程可以CAS成功,其他线程CAS失败后返回<1>处后 * 判断运行状态>=TIDYING则退出。 */ if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { try { //空函数,具体由用户实现,主要用于终止线程池后的操作。 terminated(); } finally { //最后设置线程池的状态为TERMINATED ctl.set(ctlOf(TERMINATED, 0)); termination.signalAll(); } return; } } finally { mainLock.unlock(); } } }