幾句閒扯:首先,我想說java的線程池真的是很繞,之前一直都感受新建幾個線程一直不退出究竟是怎麼實現的,也就有了後來學習ThreadPoolExecutor源碼。學習源碼的過程當中,最噁心的其實就是幾種狀態的轉換了,這也是ThreadPoolExecutor的核心。花了將近小一週才大體的弄明白ThreadPoolExecutor的機制,遂記錄下來。java
線程是一個程序員必定會涉及到的一個概念,可是線程的建立和切換都是代價比較大的。因此,咱們有沒有一個好的方案能作到線程的複用呢?這就涉及到一個概念——線程池。合理的使用線程池可以帶來3個很明顯的好處:程序員
java的線程池支持主要經過ThreadPoolExecutor來實現,咱們使用的ExecutorService的各類線程池策略都是基於ThreadPoolExecutor實現的,因此ThreadPoolExecutor十分重要。要弄明白各類線程池策略,必須先弄明白ThreadPoolExecutor。安全
首先看一個線程池的流程圖:bash
線程池的建立能夠經過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; } 複製代碼
具體解釋一下上述參數:多線程
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
核心方法: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; 複製代碼
實際上是比較難懂的,主要在線程池狀態判斷條件這裏:
從這裏是能夠看出一些策略的
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:
當調用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的:
也說明了兩點: 1. SHUTDOWN想轉化爲TIDYING,須要workQueue爲空,同時workerCount爲0。 2. STOP轉化爲TIDYING,須要workerCount爲0
若是知足上面的條件(通常必定時間後都會知足的),那麼CAS成TIDYING,TIDYING也只是個過分狀態,最終會轉化爲TERMINATED。
至此,ThreadPoolExecutor一些核心思想就介紹完了,想分析清楚實在是不容易,對於ThreadPoolExecutor我仍是有些不懂地方,以上只是我對源碼的片面的看法,若是有不正確之處,但願各位大佬指出
歡迎你們關注個人公衆號【風平浪靜如碼】,海量Java相關文章,學習資料都會在裏面更新,整理的資料也會放在裏面。
以爲寫的還不錯的就點個贊,加個關注唄!點關注,不迷路,持續更新!!!