Java多執行緒-ThreadPool執行緒池-2（四）
執行緒池是個神器，用得好會非常地方便。本來覺得執行緒池的構造器有些複雜，即使講清楚了對今後的用處可能也不太大，因為有一些Java定義好的執行緒池可以直接使用。但是（凡事總有個但是），還是覺得講一講可能跟有助於理解後面的常用執行緒池，所以該打臉還是打吧
因為直接結合程式碼看會更清楚一些，所以我把帶注釋的程式碼貼出來：
public class ThreadPoolExecutor { public ThreadPoolExecutor( /** * corePoolSize：初始化時指定的核心執行緒數，包括空閑執行緒，必須大於等於0，當有新任務提交時，會執行以下判斷（workCount為當前活躍的執行緒數量）: * 當workCount＜ corePoolSize：即使執行緒池中有空閑執行緒，也會創建新執行緒 * 當corePoolSize ≤ workCount ＜ maximumPoolSize：只有workQueue滿時才創建新執行緒 * 當corePoolSize < workCount ＜ maximumPoolSize：且超過corePoolSize部分的執行緒空閑時間達到keepAliveTime時，就回收這些執行緒，當設置allowCoreThreadTimeOut(true)時， * 執行緒池中corePoolSize範圍內的執行緒空閑時間達到keepAliveTime也將被回收 * 當設置corePoolSize == maximumPoolSize：執行緒池的大小固定，此時如有新任務提交，且workQueue未滿時，會將請求放入workQueue，等待有空閑的執行緒從workQueue中取任務並處理 * 當workCount ≥ maximumPoolSize：若workQueue滿，則採取handler對應的策略 */ int corePoolSize, // maximumPoolSize：初始化時指定的最大執行緒數量 int maximumPoolSize, // keepAliveTime：執行緒池維護執行緒所允許的空閑時間。當執行緒池中的執行緒數量大於corePoolSize時，如果這時沒有新的任務提交，核心執行緒外的執行緒不會立即銷毀，而是等待，直到等待的時間超過了keepAliveTime long keepAliveTime, // 空閑時間單位 TimeUnit unit, /** * workQueue：阻塞隊列的類型是保存等待執行的任務的阻塞隊列，主要有四種提交方式： * SynchronousQueue：同步隊列，這個「隊列」內部只包含了一個元素，隊列的size始終為0，每執行一個put，就需要一個take來解除阻塞，反之也一樣。飽和狀態下，執行緒池能處理的最大執行緒數量為maximumPoolSize * 使用SynchronousQueue隊列，提交的任務不會保存，而是會馬上提交執行 * 需要對程式的並發量有個準確的評估，才能設置合適的maximumPoolSize數量，否則很容易就會執行拒絕策略 * ArrayBlockingQueue：有界任務隊列，飽和狀態下，執行緒池能處理的最大執行緒數量為maximumPoolSize + ArrayBlockingQueue.SIZE * LinkedBlockingQueue：無界任務隊列，執行緒池的任務隊列可以無限制的添加新的任務，此時執行緒池能夠創建的最大執行緒數是corePoolSize， * 而maximumPoolSize就無效了，執行緒池飽和狀態下能處理的最大執行緒數量只取決於系統的性能 * PriorityBlockingQueue：優先任務隊列，同LinkedBlockingQueue一樣，它也是一個無界的任務隊列，只不過需要自己實現元素的Comparable排序介面 */ BlockingQueue<Runnable> workQueue, // threadFactory：創建新執行緒，使新創建的執行緒有相同的優先順序且為非守護執行緒，同時設置執行緒的名稱，默認使用Executors.DefaultThreadFactory類創建 ThreadFactory threadFactory, /** * handler：表示執行緒池的飽和策略，意思就是如果阻塞隊列滿了並且沒有空閑的執行緒，此時如果繼續提交任務，就需要採取一種策略處理該任務，執行緒池提供了4種策略 * AbortPolicy：直接拋出異常，這是默認策略 * CallerRunsPolicy：如果執行緒池的執行緒數量達到上限，則把任務隊列中的任務放在調用者的執行緒當運行 * DiscardOldestPolicy：丟棄阻塞隊列中靠最前的任務，並執行當前任務 * DiscardPolicy：直接丟棄任務 */ RejectedExecutionHandler handler) { // balabala… … } }
這樣就清晰多了。
其中，最主要是要清楚幾種workQueue，也就是BlockingQueue<Runnable>的作用。
SynchronousQueue同步隊列，這個隊列沒有所謂的緩衝，這樣做是為了排除阻塞時隊列丟消息的可能。如果沒有其他微服務並行執行的話，可以放心地用這個隊列，不然還是小心一點為妙。它的示例程式碼：
/** * 同步隊列 */ public class SynchronousQueueTest { public static void main(String[] args) { ExecutorService service = new ThreadPoolExecutor( 1, // 當要處理的執行緒數超過maximumPoolSize時，拋出異常 2, 1000, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy() ); for (int i = 0; i < 10; i++) { service.execute(() -> System.out.println("當前執行緒 " + Thread.currentThread().getName()) ); } service.shutdown(); } }
ArrayBlockingQueue，它的使用範圍非常廣，一般可以用於輕量級的同步鎖，也就是在同一個服務中（也就是非微服務架構），如果要具有分散式鎖的功能又不想部署zookeeper這麼麻煩的話，ArrayBlockingQueue就是一個非常不錯的選擇。
/** * 有界阻塞隊列 */ public class ArrayBlockingQueueTest { public static void main(String[] args) { ExecutorService service = new ThreadPoolExecutor( // 要處理的執行緒數超過maximumPoolSize + workQueue.SIZE時，拋出異常 1, 2, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(10), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy() ); // 因為 maximumPoolSize（2） + ArrayBlockingQueue.SIZE（10） < 13，所以會拋出異常 for (int i = 0; i < 13; i++) { service.execute(() -> System.out.println("當前執行緒 " + Thread.currentThread().getName()) ); } service.shutdown(); } }
再來看看ArrayBlockingQueue的另一個例子，可以加深印象：
public class ArrayBlockingQueueTester { public static BlockingQueue<String> queue = new ArrayBlockingQueue<String>(5); // 一個往裡放 class Producer implements Runnable { @Override public void run() { try { queue.put("川菜"); System.out.println(Thread.currentThread().getName() + " 廚師做好 川菜"); } catch (InterruptedException e) { e.printStackTrace(); } } } // 一個往外拿 class Consumer implements Runnable { @Override public void run() { try { String food = queue.take(); System.out.println( Thread.currentThread().getName() + " 客人消費 " + food ); } catch (InterruptedException e) { e.printStackTrace(); } } } public static void main(String[] args) { // 客人等著菜 for (int i = 0; i < 5; i++) { new Thread(new ArrayBlockingQueueTester().new Consumer()).start(); } // 廚師做好菜 for (int i = 0; i < 5; i++) { new Thread(new ArrayBlockingQueueTester().new Producer()).start(); } } }
ArrayBlockingQueue說白了就是一個往裡放，一個往外拿：
1、往裡放的，只能最多放指定個數就不能再放了（阻塞，等待，這裡是5個）；
2、往外拿的，如果沒有可以拿的了，就等著（阻塞，等待）。
咱們點菜的時候不就是這樣嗎？
LinkedBlockingQueue這個就牛逼了，相當於無底洞，有多少處理多少，此時執行緒池能夠創建的最大執行緒數是corePoolSize，而maximumPoolSize就成了擺設，這等於說是完全取決於系統的性能。
/** * 無界阻塞隊列 */ public class LinkedBlockingQueueTest { public static void main(String[] args) { // 要處理的執行緒數過大時，是否拋出異常，取決於機器的性能 ExecutorService service = new ThreadPoolExecutor( 1, 2, 1000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy() ); for (int i = 0; i < 10000; i++) { service.execute(() -> System.out.println("當前執行緒 " + Thread.currentThread().getName()) ); } service.shutdown(); } }
最後一個隊列是PriorityBlockingQueue，它是一種有優先順序的無界阻塞隊列，默認的元素執行順序是升序，可以通過自定義介面Comparable<T>實現compareTo()方法來指定隊列中的元素執行順序。
/** * 測試類 */ public class Test1 implements Runnable, Comparable<Test1> { private int priority; public Test1(int priority) { this.priority = priority; } public int getPriority() { return priority; } public void setPriority(int priority) { this.priority = priority; } @Override public int compareTo(Test1 o) { // 返回1時為升序 // 返回-1為降序 return this.priority > o.priority ? -1 : 1; } @Override public void run() { System.out.println( "當前執行緒 " + Thread.currentThread().getName() + ", priority = " + this.priority ); } } /** * 有優先順序的無界阻塞隊列 */ public class PriorityBlockingQueueTest { public static void main(String[] args) { ExecutorService service = new ThreadPoolExecutor( 1, 2, 1000, TimeUnit.MILLISECONDS, new PriorityBlockingQueue<>(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy() ); for (int i = 0; i < 10; i++) { service.execute(new Test1(i)); } service.shutdown(); } }
如果想在執行緒池的執行執行緒中加入一點自己希望的動作，可以通過自定義ThreadFactory實現。
/** * 測試類 */ public class Test2 implements Runnable { private String name; public Test2(String name) { this.name = name; } public String getName() { return name; } public void setName(String name) { this.name = name; } @Override public void run() { System.out.println(this.getName() + "，當前執行緒 " + Thread.currentThread().getName()); } } /** * 自定義ThreadFactory */ public class SelfDefineThreadPoolExecutor { public static void main(String[] args) { ExecutorService service = new ThreadPoolExecutor( 1, 2, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(8), // 自定義ThreadFactory new ThreadFactory() { @Override public Thread newThread(Runnable r) { System.out.println("執行緒 " + r.hashCode() + " 創建"); return new Thread(r, "thread-pool-" + r.hashCode()); } }, // 加入自定義動作 new ThreadPoolExecutor.CallerRunsPolicy() ) { public void beforeExecute(Thread thread, Runnable runnable) { System.out.println(((Test2) runnable).getName() + " 準備執行"); } public void afterExecute(Thread thread, Runnable runnable) { System.out.println(((Test2) runnable).getName() + " 執行完畢"); } public void terminated() { System.out.println("執行緒池關閉"); } }; for (int i = 0; i < 10; i++) { service.execute(new Test2("Test2" + i)); } service.shutdown(); } }
其實主要是把常用那幾個workQueue搞搞清楚，因為這幾個在今後的工作中可能會用到，尤其是ArrayBlockingQueue，它和後面會說的另兩個神器，可以說是是「執行緒三寶」。
