在上一篇文章中，我們介紹了一種最簡單的分工模式——Thread-Per-Message 模式，對應到現實世界，其實就是委托代辦。這種分工模式如果用 Java Thread 實現，頻繁地創建、銷毀線程非常影響性能，同時無限制地創建線程還可能導致 OOM，所以在 Java 領域使用場景就受限了。
要想有效避免線程的頻繁創建、銷毀以及 OOM 問題，就不得不提今天我們要細聊的，也是 Java 領域使用最多的 Worker Thread 模式。

Worker Thread 模式及其實現

Worker Thread 模式可以類比現實世界里車間的工作模式：車間里的工人，有活兒了，大家一起干，沒活兒了就聊聊天等著。你可以參考下面的示意圖來理解，Worker Thread 模式中 Worker Thread 對應到現實世界里，其實指的就是車間里的工人。不過這里需要注意的是，車間里的工人數量往往是確定的。

那在編程領域該如何模擬車間的這種工作模式呢？或者說如何去實現 Worker Thread 模式呢？通過上面的圖，你很容易就能想到用阻塞隊列做任務池，然后創建固定數量的線程消費阻塞隊列中的任務。其實你仔細想會發現，這個方案就是 Java 語言提供的線程池。

線程池有很多優點，例如能夠避免重復創建、銷毀線程，同時能夠限制創建線程的上限等等。學習完上一篇文章后你已經知道，用 Java 的 Thread 實現 Thread-Per-Message 模式難以應對高并發場景，原因就在于頻繁創建、銷毀 Java 線程的成本有點高，而且無限制地創建線程還可能導致應用 OOM。線程池，則恰好能解決這些問題。
那我們還是以 echo 程序為例，看看如何用線程池來實現。

下面的示例代碼是用線程池實現的 echo 服務端，相比于 Thread-Per-Message 模式的實現，改動非常少，僅僅是創建了一個最多線程數為 500 的線程池 es，然后通過 es.execute() 方法將請求處理的任務提交給線程池處理。

ExecutorService es = Executors.newFixedThreadPool(500); final ServerSocketChannel ssc = ServerSocketChannel.open().bind(new InetSocketAddress(8080)); //處理請求 try {while (true) {// 接收請求SocketChannel sc = ssc.accept();// 將請求處理任務提交給線程池es.execute(()->{try {// 讀SocketByteBuffer rb = ByteBuffer.allocateDirect(1024);sc.read(rb);//模擬處理請求Thread.sleep(2000);// 寫SocketByteBuffer wb = (ByteBuffer)rb.flip();sc.write(wb);// 關閉Socketsc.close();}catch(Exception e){throw new UncheckedIOException(e);}});} } finally {ssc.close();es.shutdown(); }

正確地創建線程池

Java 的線程池既能夠避免無限制地創建線程導致 OOM，也能避免無限制地接收任務導致 OOM。只不過后者經常容易被我們忽略，例如在上面的實現中，就被我們忽略了。所以強烈建議你用創建有界的隊列來接收任務。
當請求量大于有界隊列的容量時，就需要合理地拒絕請求。如何合理地拒絕呢？這需要你結合具體的業務場景來制定，即便線程池默認的拒絕策略能夠滿足你的需求，也同樣建議你在創建線程池時，清晰地指明拒絕策略。
同時，為了便于調試和診斷問題，我也強烈建議你在實際工作中給線程賦予一個業務相關的名字。
綜合以上這三點建議，echo 程序中創建線程可以使用下面的示例代碼。

ExecutorService es = new ThreadPoolExecutor(50, 500,60L, TimeUnit.SECONDS,//注意要創建有界隊列new LinkedBlockingQueue<Runnable>(2000),//建議根據業務需求實現ThreadFactoryr->{return new Thread(r, "echo-"+ r.hashCode());},//建議根據業務需求實現RejectedExecutionHandlernew ThreadPoolExecutor.CallerRunsPolicy());

避免線程死鎖

使用線程池過程中，還要注意一種線程死鎖的場景。如果提交到相同線程池的任務不是相互獨立的，而是有依賴關系的，那么就有可能導致線程死鎖。實際工作中，我就親歷過這種線程死鎖的場景。具體現象是應用每運行一段時間偶爾就會處于無響應的狀態，監控數據看上去一切都正常，但是實際上已經不能正常工作了。
這個出問題的應用，相關的邏輯精簡之后，如下圖所示，該應用將一個大型的計算任務分成兩個階段，第一個階段的任務會等待第二階段的子任務完成。在這個應用里，每一個階段都使用了線程池，而且兩個階段使用的還是同一個線程池。

我們可以用下面的示例代碼來模擬該應用，如果你執行下面的這段代碼，會發現它永遠執行不到最后一行。執行過程中沒有任何異常，但是應用已經停止響應了。

//L1、L2階段共用的線程池 ExecutorService es = Executors.newFixedThreadPool(2); //L1階段的閉鎖 CountDownLatch l1=new CountDownLatch(2); for (int i=0; i<2; i++){System.out.println("L1");//執行L1階段任務es.execute(()->{//L2階段的閉鎖 CountDownLatch l2=new CountDownLatch(2);//執行L2階段子任務for (int j=0; j<2; j++){es.execute(()->{System.out.println("L2");l2.countDown();});}//等待L2階段任務執行完l2.await();l1.countDown();}); } //等著L1階段任務執行完 l1.await(); System.out.println("end");

當應用出現類似問題時，首選的診斷方法是查看線程棧。下圖是上面示例代碼停止響應后的線程棧，你會發現線程池中的兩個線程全部都阻塞在 l2.await(); 這行代碼上了，也就是說，線程池里所有的線程都在等待 L2 階段的任務執行完，那 L2 階段的子任務什么時候能夠執行完呢？永遠都沒那一天了，為什么呢？因為線程池里的線程都阻塞了，沒有空閑的線程執行 L2 階段的任務了。

原因找到了，那如何解決就簡單了，最簡單粗暴的辦法就是將線程池的最大線程數調大，如果能夠確定任務的數量不是非常多的話，這個辦法也是可行的，否則這個辦法就行不通了。其實這種問題通用的解決方案是為不同的任務創建不同的線程池。對于上面的這個應用，L1 階段的任務和 L2 階段的任務如果各自都有自己的線程池，就不會出現這種問題了。
最后再次強調一下：提交到相同線程池中的任務一定是相互獨立的，否則就一定要慎重。

總結

我們曾經說過，解決并發編程里的分工問題，最好的辦法是和現實世界做對比。對比現實世界構建編程領域的模型，能夠讓模型更容易理解。上一篇我們介紹的 Thread-Per-Message 模式，類似于現實世界里的委托他人辦理，而今天介紹的 Worker Thread 模式則類似于車間里工人的工作模式。如果你在設計階段，發現對業務模型建模之后，模型非常類似于車間的工作模式，那基本上就能確定可以在實現階段采用 Worker Thread 模式來實現。

Worker Thread 模式和 Thread-Per-Message 模式的區別有哪些呢？從現實世界的角度看，你委托代辦人做事，往往是和代辦人直接溝通的；對應到編程領域，其實現也是主線程直接創建了一個子線程，主子線程之間是可以直接通信的。而車間工人的工作方式則是完全圍繞任務展開的，一個具體的任務被哪個工人執行，預先是無法知道的；對應到編程領域，則是主線程提交任務到線程池，但主線程并不關心任務被哪個線程執行。

Worker Thread 模式能避免線程頻繁創建、銷毀的問題，而且能夠限制線程的最大數量。Java 語言里可以直接使用線程池來實現 Worker Thread 模式，線程池是一個非常基礎和優秀的工具類，甚至有些大廠的編碼規范都不允許用 new Thread() 來創建線程的，必須使用線程池。

不過使用線程池還是需要格外謹慎的，除了今天重點講到的如何正確創建線程池、如何避免線程死鎖問題，還需要注意前面我們曾經提到的 ThreadLocal 內存泄露問題。同時對于提交到線程池的任務，還要做好異常處理，避免異常的任務從眼前溜走，從業務的角度看，有時沒有發現異常的任務后果往往都很嚴重。

總結

以上是生活随笔為你收集整理的Java并发编程实战~Worker Thread模式的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。