1 背景
某年某月某日，和我的臥龍同事聊一個需求，說是有個數據查詢的功能，因為涉及到多個第三方接口調用，想用線程池并行來做。
很正常的一個方案，但是上線后發現，每次服務發布的時候，這個數據查詢的功能就會掛掉，后來發現是線程池沒有做好關閉，這里總結一下。
關鍵字：線程池；shutdown；shutdownNow；interrupt
2 線程中斷 interrupt
先補一補基礎的知識：線程中斷。
線程中斷的含義，并不是強制把運行中的線程給“咔嚓”中斷，而是把線程的中斷標志位置為true，這樣等線程之后阻塞（wAIt、join、sleep）的時候，就會拋出 InterruptedException，程序通過捕獲 InterruptedException 來做一定的善后處理，然后讓線程退出。
來看個例子，下面這段代碼是起一個線程，打印一百行文本，打印過程中，會把線程的中斷標志位置為true
public static void test02() throws InterruptedException {
Thread t = new Thread(() -> {
for (int i = 0; i < 100; i++) {
System.out.println("process i=" + i + "，interrupted：" + Thread.currentThread().isInterrupted());
}
});
t.start();
Thread.sleep(1);
t.interrupt();
}
看看控制臺的輸出，發現在打印到 57 的時候，中斷標志位已經成功置為true了，但是線程任然在打印，說明只是設置了中斷標志位，而不是直接粗暴的把線程中斷。
...
process i=55，interrupted：false
process i=56，interrupted：false
process i=57，interrupted：true
process i=58，interrupted：true
process i=59，interrupted：true
...
再看看這個示例，同樣是打印一百行文本，打印過程中會判斷中斷標志位，如果中斷就自行退出。
public static void test02() throws InterruptedException {
Thread t = new Thread(() -> {
for (int i = 0; i < 100; i++) {
if (Thread.interrupted()) {
System.out.println("線程已中斷，退出執行");
break;
}
System.out.println("process i=" + i + "，interrupted：" + Thread.currentThread().isInterrupted());
}
});
t.start();
Thread.sleep(1);
t.interrupt();
}
控制臺輸出如下，：
process i=49，interrupted：false
process i=50，interrupted：false
process i=51，interrupted：false
線程已中斷，退出執行
3 線程池的關閉 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();
advanceRunState(SHUTDOWN); // 1. 把線程池的狀態設置為 SHUTDOWN
interruptIdleWorkers(); // 2. 把空閑的工作線程置為中斷
onShutdown(); // 3. 一個空實現，暫不用關注
} finally {
mainLock.unlock();
}
tryTerminate();
}
看源碼先看注釋，我用我英語四級的超高水準水平翻譯下：
啟動有序關閉會執行以前提交的任務，但不接受任何新任務。
如果已經關閉，則調用不會產生額外的影響。
此方法不等待活動執行的任務終止。如果需要，可使用 awaitTermination() 做到這一點。
3.1 第一步：advanceRunState(SHUTDOWN) 把線程池置為 SHUTDOWN
線程池狀態流轉如下。調用 shutdown() 方法會把線程池的狀態置為 SHUTDOWN，后續再往線程池提交任務就會被拒絕（execute() 方法中做了判斷）。
3.2 第二步：interruptIdleWorkers() 把空閑的工作線程置為中斷
interruptIdleWorkers() 方法遍歷所有的工作線程，如果 tryLock() 成功，就把線程置為中斷。
這里，如果 tryLock() 成功，說明對應的 woker 是一個空閑的，沒有在執行任務的線程，如果沒成功，說明對應的 worker 正在執行任務。也就是說，這里的中斷，對正在執行中的任務并沒有影響。
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
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();
}
}
3.3 第三步：onShutdown() 一個空實現，暫不用關注
這個沒啥，就是個留空的方法。
3.4 總結
shutdown() 方法干兩件事：
- 把線程池狀態置為 SHUTDOWN 狀態
- 中斷空閑線程
我們來看個例子，加深下印象。
public static void test01() throws InterruptedException {
// corePoolSize 是 2，maximumPoolSize 是 2
ThreadPoolExecutor es = new ThreadPoolExecutor(2, 2,
60L, TimeUnit.SECONDS,
new LinkedBlockingQueue<>());
es.prestartAllCoreThreads(); // 啟動所有 worker
es.execute(new Task()); // Task是一個訪問某網站的 HTTP 請求，跑的慢，后面會貼出來完整代碼，這里把他當做一個跑的慢的異步任務就行
es.shutdown();
es.execute(new Task()); // 在線程池 shutdown() 后 繼續添加任務，這里預期是拋出異常
}
這個例子我們主要觀察兩個現象。
一個是線程池會有兩個woker（ prestartAllCoreThreads() 方法的調用使得已啟動就有兩個 worker），其中一個正在執行，一個處于空閑。 所以當調用shutdown() 方法，走進 interruptIdleWorkers() 的時候，只有那個空閑的線程會調用 t.interrupt()。
第二個是調用 shutdown() 方法后，再調用 execute() 時，會拋出異常，因為線程池的狀態已經置為 SHUTDOWN，不再接受新的任務添加進來。
4 線程池的關閉 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); // 1：把線程池設置為STOP
interruptWorkers(); // 2.中斷工作線程
tasks = drainQueue(); // 3.把線程池中的任務都 drain 出來
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
注釋的意思是：
嘗試停止所有正在執行的任務，暫停正在等待的任務的處理，并返回等待執行的任務列表。從該方法返回時，這些任務將從任務隊列中清空(移除)。
此方法不等待活動執行的任務終止。如果需要，可使用 awaitTermination() 做到這一點。
除了盡最大努力嘗試停止處理主動執行的任務之外，沒有其他保證。
此實現通過 Thread.Interrupt() 取消任務，因此任何無法響應中斷的任務都可能永遠不會終止。
4.1 第一步：advanceRunState() 把線程池設置為STOP
和 shutdown() 方法不同的是，shutdownNow() 方法會把線程池的狀態設置為 STOP。
4.2 第二步：interruptWorkers() 中斷工作線程
interruptWorkers() 如下，可以看到，和 shutdown() 方法不同的是，所有的工作線程都調用了 interrupt() 方法
/**
* 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)
w.interruptIfStarted();
} finally {
mainLock.unlock();
}
}
4.3 第三步：drainQueue() 把線程池中的任務都 drain 出來
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;
ArrayList<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;
}
4.4 總結
shutdownNow() 方法干三件事：
- 把線程池狀態置為 STOP 狀態
- 中斷工作線程
- 把線程池中的任務都 drain 出來并返回
我們來看個例子，代碼合剛才的一樣，只是關閉線程用的是shutdownNow()
public static void test01() throws InterruptedException {
// corePoolSize 是 1，maximumPoolSize 是 1，無限容量
ThreadPoolExecutor es = new ThreadPoolExecutor(1, 1,
60L, TimeUnit.SECONDS,
new LinkedBlockingQueue<>());
es.prestartAllCoreThreads(); // 啟動所有 worker
es.execute(new Task()); // Task是一個訪問某網站的 HTTP 請求，跑的慢，后面會貼出來完整代碼，這里把他當做一個跑的慢的異步任務就行
es.execute(new Task());
List<Runnable> result = es.shutdownNow();
System.out.println(result);
es.execute(new Task()); // 在線程池 shutdownNow() 后 繼續添加任務，這里預期是拋出異常
}
這個例子我們主要觀察三個現象。 一個是線程池有兩個woker，所以當調用shutdownNow() 方法，走進 interruptWorkers() 的時候，所有的 woker 都會調用 t.interrupt()。
第二個是 shutdownNow() 方法會返回還沒來得及執行的task，并打印出來。
第三個是調用 shutdownNow() 方法后，再調用 execute() 時，會拋出異常，因為線程池的狀態已經置為 STOP，不再接受新的任務添加
5 實戰，與 JVM 鉤子配合
實際工作中，我們一般是使用 shutdown() 方法，因為它比較“溫和”，會等待我們把線程池中的任務都執行完，這里也已 shutdown() 方法為例。
我們回到最開頭聊到的那個 case，機器重新發布，但是線程池中還有沒執行完任務，機器一關，這些任務全部被kill，怎么辦呢？有什么機制能夠阻塞一下，等待這個任務執行完再關閉嗎？
有的，用 JVM 的鉤子！（深入了解 JVM 鉤子可以再看看這篇博文：掃盲 JVM 安全退出機制：shutdownHook，signalHandler[1]）
實例代碼如下，一個線程池，提交了三個任務去執行，執行完得半分鐘。然后增加一個JVM的鉤子，這個鉤子可以簡單理解為監聽器，注冊后，JVM在關閉的時候就會調用這個方法，調用完才會正式關閉JVM。
public static void test01() throws InterruptedException {
ThreadPoolExecutor es = new ThreadPoolExecutor(1, 1,
60L, TimeUnit.SECONDS,
new LinkedBlockingQueue<>());
es.execute(new Task());
es.execute(new Task());
es.execute(new Task());
Thread shutdownHook = new Thread(() -> {
es.shutdown();
try {
es.awaitTermination(3, TimeUnit.MINUTES);
} catch (InterruptedException e) {
e.printStackTrace();
System.out.println("等待超時，直接關閉");
}
});
Runtime.getRuntime().addShutdownHook(shutdownHook);
}
在機器上執行，會發現，我使用 ctrl + c （注意不是ctrl + z ）關閉進程，會發現進程并沒有直接關閉，線程池任然執行，一直等到線程池的任務執行完，進程才會正式退出。
怎么樣，是不是很神奇。
本文中涉及的 Task 的源碼如下。這個任務是對 stackoverflow 網站發起 10 次請求，用來模擬跑的比較慢的任務，當然這不是重點，可以忽略，有興趣動手試一下本文代碼的同學可以參考下。
public static class Task implements Runnable {
@Override
public void run() {
System.out.println("task start");
for (int i = 0; i < 10; i++) {
httpGet();
System.out.println("task execute " + i);
}
System.out.println("task finish");
}
private void httpGet() {
String url = "https://stackoverflow.com/";
String result = "";
BufferedReader in = null;
try {
String urlName = url;
URL realUrl = new URL(urlName);
// 打開和URL之間的連接
URLConnection conn = realUrl.openConnection();
// 設置通用的請求屬性
conn.setRequestProperty("accept", "*/*");
conn.setRequestProperty("connection", "Keep-Alive");
conn.setRequestProperty("user-agent",
"Mozilla/4.0 (compatible; MSIE 6.0; windows NT 5.1; SV1)");
// 建立實際的連接
conn.connect();
// 獲取所有響應頭字段
Map<String, List<String>> map = conn.getHeaderFields();
// 遍歷所有的響應頭字段
// for (String key : map.keySet()) {
// System.out.println(key + "--->" + map.get(key));
// }
// 定義BufferedReader輸入流來讀取URL的響應
in = new BufferedReader(
new InputStreamReader(conn.getInputStream()));
String line;
while ((line = in.readLine()) != null) {
result += "/n" + line;
}
} catch (Exception e) {
e.printStackTrace();
}
// 使用finally塊來關閉輸入流
finally {
try {
if (in != null) {
in.close();
}
} catch (Exception ex) {
ex.printStackTrace();
}
}
// System.out.print(result);
}
}
6 總結
想要優雅的關閉線程池，首先要理解線程中斷的含義。
其次，關閉線程池有兩種方式：shutdown() 和 shutdownNow()，二者最大的區別是 shutdown() 只是把空閑的 woker 置為中斷，不影響正在運行的woker，并且會繼續把待執行的任務給處理完。shutdonwNow() 則是把所有的 woker 都置為中斷，待執行的任務全部抽出并返回，日常工作中更多是使用 shutdown()。
最后，單純的使用 shutdown() 也不靠譜，還得使用 awaitTermination() 和 JVM 的鉤子，才算優雅的關閉線程池。
