某年某月某日,和我的臥龍同事聊一個需求,說是有個數據查詢的功能,因為涉及到多個第三方接口調用,想用線程池并行來做。
很正常的一個方案,但是上線后發現,每次服務發布的時候,這個數據查詢的功能就會掛掉,后來發現是線程池沒有做好關閉,這里總結一下。
關鍵字:線程池;shutdown;shutdownNow;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
線程已中斷,退出執行
了解完線程中斷,再來看看線程池的關閉方法。
關閉線程池有兩個方法 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() 做到這一點。
線程池狀態流轉如下。調用 shutdown() 方法會把線程池的狀態置為 SHUTDOWN,后續再往線程池提交任務就會被拒絕(execute() 方法中做了判斷)。
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(); }}
這個沒啥,就是個留空的方法。
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,不再接受新的任務添加進來。
/** * 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() 取消任務,因此任何無法響應中斷的任務都可能永遠不會終止。
和 shutdown() 方法不同的是,shutdownNow() 方法會把線程池的狀態設置為 STOP。
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(); }}
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;}
shutdownNow() 方法干三件事:
我們來看個例子,代碼合剛才的一樣,只是關閉線程用的是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,不再接受新的任務添加
實際工作中,我們一般是使用 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); } }
想要優雅的關閉線程池,首先要理解線程中斷的含義。
其次,關閉線程池有兩種方式:shutdown() 和 shutdownNow(),二者最大的區別是 shutdown() 只是把空閑的 woker 置為中斷,不影響正在運行的woker,并且會繼續把待執行的任務給處理完。shutdonwNow() 則是把所有的 woker 都置為中斷,待執行的任務全部抽出并返回,日常工作中更多是使用 shutdown()。
最后,單純的使用 shutdown() 也不靠譜,還得使用 awaitTermination() 和 JVM 的鉤子,才算優雅的關閉線程池。
本文鏈接:http://www.www897cc.com/showinfo-26-50770-0.html優雅的關閉Java線程池,這樣做才是yyds
聲明:本網頁內容旨在傳播知識,若有侵權等問題請及時與本網聯系,我們將在第一時間刪除處理。郵件:2376512515@qq.com
上一篇: 一種用于在多個進程之間共享數據的機制