2. 程式人生 > >ThreadPoolExecutor中策略的選擇與工作隊列的選擇（java線程池）

ThreadPoolExecutor中策略的選擇與工作隊列的選擇（java線程池）

阿新 • • 發佈：2017-08-17
完全 系統 rod 大小 font 操作系統 重復 null 定義

工作原理

1、線程池剛創建時，裏面沒有一個線程。任務隊列是作為參數傳進來的。不過，就算隊列裏面有任務，線程池也不會馬上執行它們。

2、當調用 execute() 方法添加一個任務時，線程池會做如下判斷：

a. 如果正在運行的線程數量小於 corePoolSize，那麽馬上創建線程運行這個任務；

　　 b. 如果正在運行的線程數量大於或等於 corePoolSize，那麽將這個任務放入隊列。

　　 c. 如果這時候隊列滿了，而且正在運行的線程數量小於 maximumPoolSize，那麽還是要創建線程運行這個任務；

　　 d. 如果隊列滿了，而且正在運行的線程數量大於或等於 maximumPoolSize，那麽線程池會拋出異常，告訴調用者“我不能再接受任務了”。

3、當一個線程完成任務時，它會從隊列中取下一個任務來執行。

4、當一個線程無事可做，超過一定的時間（keepAliveTime）時，線程池會判斷，如果當前運行的線程數大於 corePoolSize，那麽這個線程就被停掉。所以線程池的所有任務完成後，它最終會收縮到 corePoolSize 的大小。

　 　這樣的過程說明，並不是先加入任務就一定會先執行。假設隊列大小為 10，corePoolSize 為 3，maximumPoolSize 為 6，那麽當加入 20 個任務時，執行的順序就是這樣的：首先執行任務 1、2、3，然後任務 4~13 被放入隊列。這時候隊列滿了，任務 14、15、16 會被馬上執行，而任務 17~20 則會拋出異常。最終順序是：1、2、3、14、15、16、4、5、6、7、8、9、10、11、12、13。下面是一個線程池使用的例子：

排隊

所有 BlockingQueue 都可用於傳輸和保持提交的任務。可以使用此隊列與池大小進行交互：

  • 如果運行的線程少於 corePoolSize，則 Executor 始終首選添加新的線程，而不進行排隊。
  • 如果運行的線程等於或多於 corePoolSize，則 Executor 始終首選將請求加入隊列，而不添加新的線程。
  • 如果無法將請求加入隊列，則創建新的線程，除非創建此線程超出 maximumPoolSize，在這種情況下，任務將被拒絕。

排隊有三種通用策略：

  1. 直接提交。工作隊列的默認選項是 SynchronousQueue，它將任務直接提交給線程而不保持它們。在此，如果不存在可用於立即運行任務的線程，則試圖把任務加入隊列將失敗，因此會構造一個新的線程。此策略可以避免在處理可能具有內部依賴性的請求集時出現鎖。直接提交通常要求無界 maximumPoolSizes 以避免拒絕新提交的任務。當命令以超過隊列所能處理的平均數連續到達時，此策略允許無界線程具有增長的可能性。
  2. 無界隊列。使用無界隊列（例如，不具有預定義容量的 LinkedBlockingQueue）將導致在所有 corePoolSize 線程都忙時新任務在隊列中等待。這樣，創建的線程就不會超過 corePoolSize。（因此，maximumPoolSize 的值也就無效了。）當每個任務完全獨立於其他任務，即任務執行互不影響時，適合於使用無界隊列；例如，在 Web 頁服務器中。這種排隊可用於處理瞬態突發請求，當命令以超過隊列所能處理的平均數連續到達時，此策略允許無界線程具有增長的可能性。
  3. 有界隊列。當使用有限的 maximumPoolSizes 時，有界隊列（如 ArrayBlockingQueue）有助於防止資源耗盡，但是可能較難調整和控制。隊列大小和最大池大小可能需要相互折衷：使用大型隊列和小型池可以最大限度地降低 CPU 使用率、操作系統資源和上下文切換開銷，但是可能導致人工降低吞吐量。如果任務頻繁阻塞（例如，如果它們是 I/O 邊界），則系統可能為超過您許可的更多線程安排時間。使用小型隊列通常要求較大的池大小，CPU 使用率較高，但是可能遇到不可接受的調度開銷，這樣也會降低吞吐量。

被拒絕的任務當 Executor 已經關閉，並且 Executor 將有限邊界用於最大線程和工作隊列容量，且已經飽和時，在方法 execute(java.lang.Runnable) 中提交的新任務將被拒絕。在以上兩種情況下，execute 方法都將調用其RejectedExecutionHandler 的RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) 方法。下面提供了四種預定義的處理程序策略：

  1. 在默認的 ThreadPoolExecutor.AbortPolicy 中，處理程序遭到拒絕將拋出運行時RejectedExecutionException。
  2. 在 ThreadPoolExecutor.CallerRunsPolicy 中，線程調用運行該任務的execute 本身。此策略提供簡單的反饋控制機制，能夠減緩新任務的提交速度。
  3. 在 ThreadPoolExecutor.DiscardPolicy 中，不能執行的任務將被刪除。
  4. 在 ThreadPoolExecutor.DiscardOldestPolicy 中，如果執行程序尚未關閉，則位於工作隊列頭部的任務將被刪除，然後重試執行程序（如果再次失敗，則重復此過程）。

定義和使用其他種類的 RejectedExecutionHandler 類也是可能的，但這樣做需要非常小心，尤其是當策略僅用於特定容量或排隊策略時。

公用的類ThreadPoolTask

[java] view plain copy
  1. public class ThreadPoolTask implements Runnable {
  2. // 保存任務所需要的數據
  3. private Object threadPoolTaskData;
  4. private static int consumerTaskSleepTime = 2000;
  6. ThreadPoolTask(Object tasks) {
  7. this.threadPoolTaskData = tasks;
  8. }
  10. public void run() {
  11. // 處理一個任務，這裏的處理方式太簡單了，僅僅是一個打印語句
  12. System.out.println("start .." + threadPoolTaskData);
  13. try {
  14. //便於觀察，等待一段時間
  15. Thread.sleep(consumerTaskSleepTime);
  16. } catch (Exception e) {
  17. e.printStackTrace();
  18. }
  19. System.out.println("finish " + threadPoolTaskData);
  20. threadPoolTaskData = null;
  21. }
  23. }



ThreadPoolExecutor.AbortPolicy

[java] view plain copy
  1. import java.util.concurrent.ArrayBlockingQueue;
  2. import java.util.concurrent.LinkedBlockingQueue;
  3. import java.util.concurrent.ThreadPoolExecutor;
  4. import java.util.concurrent.TimeUnit;
  6. public class ThreadPool {
  8. //讓可執行程序休息一下
  9. private static int executePrograms = 0;
  10. private static int produceTaskMaxNumber = 10;
  12. public static void main(String[] args) {
  13. // 構造一個線程池
  14. ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3,
  15. TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(3),
  16. new ThreadPoolExecutor.CallerRunsPolicy());
  18. for (int i = 1; i <= produceTaskMaxNumber; i++) {
  19. try {
  20. String task = "task@ " + i;
  21. System.out.println("put " + task);
  22. threadPool.execute(new ThreadPoolTask(task));
  24. // 便於觀察，等待一段時間
  25. Thread.sleep(executePrograms);
  26. } catch (Exception e) {
  27. e.printStackTrace();
  28. }
  29. }
  30. }
  31. }

運行結果

[java] view plain copy
  1. put task@ 1
  2. put task@ 2
  3. start ..task@ 1
  4. put task@ 3
  5. put task@ 4
  6. start ..task@ 2
  7. put task@ 5
  8. put task@ 6
  9. put task@ 7
  10. start ..task@ 6
  11. put task@ 8
  12. start ..task@ 7
  13. java.util.concurrent.RejectedExecutionException
  14. at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(Unknown Source)
  15. at java.util.concurrent.ThreadPoolExecutor.reject(Unknown Source)
  16. at java.util.concurrent.ThreadPoolExecutor.execute(Unknown Source)
  17. at ThreadPool.main(ThreadPool.java:22)
  18. put task@ 9
  19. java.util.concurrent.RejectedExecutionException
  20. at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(Unknown Source)
  21. at java.util.concurrent.ThreadPoolExecutor.reject(Unknown Source)
  22. at java.util.concurrent.ThreadPoolExecutor.execute(Unknown Source)
  23. at ThreadPool.main(ThreadPool.java:22)
  24. put task@ 10
  25. java.util.concurrent.RejectedExecutionException
  26. at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(Unknown Source)
  27. at java.util.concurrent.ThreadPoolExecutor.reject(Unknown Source)
  28. at java.util.concurrent.ThreadPoolExecutor.execute(Unknown Source)
  29. at ThreadPool.main(ThreadPool.java:22)
  30. finish task@ 2
  31. finish task@ 6
  32. finish task@ 7
  33. start ..task@ 4
  34. start ..task@ 3
  35. finish task@ 1
  36. start ..task@ 5
  37. finish task@ 4
  38. finish task@ 3
  39. finish task@ 5



ThreadPoolExecutor.CallerRunsPolicy

[java] view plain copy
  1. import java.util.concurrent.ArrayBlockingQueue;
  2. import java.util.concurrent.LinkedBlockingQueue;
  3. import java.util.concurrent.ThreadPoolExecutor;
  4. import java.util.concurrent.TimeUnit;
  6. public class ThreadPool {
  8. //讓可執行程序休息一下
  9. private static int executePrograms = 0;
  10. private static int produceTaskMaxNumber = 10;
  12. public static void main(String[] args) {
  13. // 構造一個線程池
  14. ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3,
  15. TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(3),
  16. new ThreadPoolExecutor.CallerRunsPolicy());
  18. for (int i = 1; i <= produceTaskMaxNumber; i++) {
  19. try {
  20. String task = "task@ " + i;
  21. System.out.println("put " + task);
  22. threadPool.execute(new ThreadPoolTask(task));
  24. // 便於觀察，等待一段時間
  25. Thread.sleep(executePrograms);
  26. } catch (Exception e) {
  27. e.printStackTrace();
  28. }
  29. }
  30. }
  31. }

運行結果

[java] view plain copy
  1. put task@ 1
  2. put task@ 2
  3. start ..task@ 1
  4. put task@ 3
  5. start ..task@ 2
  6. put task@ 4
  7. put task@ 5
  8. put task@ 6
  9. put task@ 7
  10. start ..task@ 6
  11. put task@ 8
  12. start ..task@ 8
  13. start ..task@ 7
  14. finish task@ 2
  15. finish task@ 1
  16. start ..task@ 3
  17. start ..task@ 4
  18. finish task@ 6
  19. start ..task@ 5
  20. finish task@ 7
  21. finish task@ 8
  22. put task@ 9
  23. put task@ 10
  24. start ..task@ 9
  25. finish task@ 4
  26. finish task@ 3
  27. start ..task@ 10
  28. finish task@ 9
  29. finish task@ 5
  30. finish task@ 10



ThreadPoolExecutor.DiscardPolicy

[java] view plain copy
  1. import java.util.concurrent.ArrayBlockingQueue;
  2. import java.util.concurrent.LinkedBlockingQueue;
  3. import java.util.concurrent.ThreadPoolExecutor;
  4. import java.util.concurrent.TimeUnit;
  6. public class ThreadPool {
  8. //讓可執行程序休息一下
  9. private static int executePrograms = 0;
  10. private static int produceTaskMaxNumber = 10;
  12. public static void main(String[] args) {
  13. // 構造一個線程池
  14. ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3,
  15. TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(3),
  16. new ThreadPoolExecutor.CallerRunsPolicy());
  18. for (int i = 1; i <= produceTaskMaxNumber; i++) {
  19. try {
  20. String task = "task@ " + i;
  21. System.out.println("put " + task);
  22. threadPool.execute(new ThreadPoolTask(task));
  24. // 便於觀察，等待一段時間
  25. Thread.sleep(executePrograms);
  26. } catch (Exception e) {
  27. e.printStackTrace();
  28. }
  29. }
  30. }
  31. }

運行結果

[java] view plain copy
  1. put task@ 1
  2. put task@ 2
  3. start ..task@ 1
  4. put task@ 3
  5. start ..task@ 2
  6. put task@ 4
  7. put task@ 5
  8. put task@ 6
  9. put task@ 7
  10. start ..task@ 6
  11. put task@ 8
  12. put task@ 9
  13. start ..task@ 7
  14. put task@ 10
  15. finish task@ 2
  16. finish task@ 1
  17. start ..task@ 3
  18. start ..task@ 4
  19. finish task@ 7
  20. finish task@ 6
  21. start ..task@ 5
  22. finish task@ 3
  23. finish task@ 4
  24. finish task@ 5



ThreadPoolExecutor.DiscardOldestPolicy

[java] view plain copy
  1. import java.util.concurrent.ArrayBlockingQueue;
  2. import java.util.concurrent.LinkedBlockingQueue;
  3. import java.util.concurrent.ThreadPoolExecutor;
  4. import java.util.concurrent.TimeUnit;
  6. public class ThreadPool {
  8. //讓可執行程序休息一下
  9. private static int executePrograms = 0;
  10. private static int produceTaskMaxNumber = 10;
  12. public static void main(String[] args) {
  13. // 構造一個線程池
  14. ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3,
  15. TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(3),
  16. new ThreadPoolExecutor.CallerRunsPolicy());
  18. for (int i = 1; i <= produceTaskMaxNumber; i++) {
  19. try {
  20. String task = "task@ " + i;
  21. System.out.println("put " + task);
  22. threadPool.execute(new ThreadPoolTask(task));
  24. // 便於觀察，等待一段時間
  25. Thread.sleep(executePrograms);
  26. } catch (Exception e) {
  27. e.printStackTrace();
  28. }
  29. }
  30. }
  31. }

運行結果：

[java] view plain copy
  1. put task@ 1
  2. put task@ 2
  3. start ..task@ 1
  4. put task@ 3
  5. start ..task@ 2
  6. put task@ 4
  7. put task@ 5
  8. put task@ 6
  9. put task@ 7
  10. start ..task@ 6
  11. put task@ 8
  12. start ..task@ 7
  13. put task@ 9
  14. put task@ 10
  15. finish task@ 6
  16. finish task@ 7
  17. start ..task@ 8
  18. finish task@ 1
  19. start ..task@ 9
  20. finish task@ 2
  21. start ..task@ 10
  22. finish task@ 8
  23. finish task@ 9
  24. finish task@ 10



ThreadPoolExecutor中策略的選擇與工作隊列的選擇（java線程池）

相關推薦

ThreadPoolExecutor中策略的選擇與工作隊列的選擇（java線程池）

完全 系統 rod 大小 font 操作系統 重復 null 定義 工作原理 1、線程池剛創建時，裏面沒有一個線程。任務隊列是作為參數傳進來的。不過，就算隊列裏面有任務，線程池也不會馬上執行它們。 2、當調用 execute() 方法添加一個任務時，線程池會做如下判斷：

Java線程池的選擇

fix value lin AC 固定 OS pan syn 正常的 　　在java的concurrent.Executors主要提供兩種線程池：無固定線程數但有限制任務隊列的cachedThreadPool與有固定線程數但無任務隊列限制的fixedThreadPool，這

#資料結構與演算法學習筆記#劍指Offer35：數字在排序陣列中出現的次數 + 測試用例（Java、C/C++）

2018.10.24 今天是程式設計師節，但是程式設計師有啥節好過的，還是好好幹活吧~ 這道題也是一道效率題，根據實際情況的不同有兩個比較好用的方法： 方法一：平均複雜度o(n)。二分找到某一個K

Java 線程池的原理與實現

控制 try 所在 使用 urn str waiting media .info 這幾天主要是狂看源程序，在彌補了一些曾經知識空白的同一時候，也學會了不少新的知識（比方 NIO）。或者稱為新技術吧。 線程池就是當中之中的一個，一提到線程。我們會想到曾經《操作系統》的

Java線程池之ThreadPoolExecutor

一次 ray 線程的創建 代碼 記錄日誌 turn throwable 限制 dex 前言 　　線程池可以提高程序的並發性能（當然是合適的情況下），因為對於沒有線程的情況下，我們每一次提交任務都新建一個線程，這種方法存在不少缺陷： 1． 線程的創建和銷毀的開銷非常高，線程

Android中關於項目中對Thread的管理（不是線程池）

lan ole 問題 直接 per dma 之前 ron move 背景 項目中對於一些並不復雜的耗時操作，比如計算，不頻繁操作數據庫等，因為沒必要使用線程池，所以之前項目會直接使用new Thread的方式，時間一長，回頭再看，原來new Thread之處已經很多

ThreadPoolExecutor（線程池）源碼分析

情況 failed rac hashset 必須 ext 阻塞 throw ice 1. 常量和變量 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 高3位為線程池的運行

Java線程池的理論與實踐

java 架構 前段時間公司裏有個項目需要進行重構，目標是提高吞吐量和可用性，在這個過程中對原有的線程模型和處理邏輯進行了修改，發現有很多基礎的多線程的知識已經模糊不清，如底層線程的運行情況、現有的線程池的策略和邏輯、池中線程的健康狀況的監控等，這次重新回顧了一下，其中涉及大量java.util.co

Java線程池拒絕策略

線程池 reject log handler 資料 HA 相關 java線程池 csdn Java線程池拒絕策略 相關資料： 線程池的RejectedExecutionHandler（拒絕策略）：http://blog.csdn.net/jgteng/article/d

深入理解java線程池—ThreadPoolExecutor

advance gre sign tor 基於 大致 eui 博文 CP 幾句閑扯：首先，我想說java的線程池真的是很繞，以前一直都感覺新建幾個線程一直不退出到底是怎麽實現的，也就有了後來學習ThreadPoolExecutor源碼。學習源碼的過程中，最惡心的其實就是幾

Java線程池實現原理與技術

到來 res ble 的區別 數據庫連接 sca 講解 屬性 活動 本文將通過實現一個簡易的線程池理解線程池的原理，以及介紹JDK中自帶的線程池ThreadPoolExecutor和Executor框架。 1.無限制線程的缺陷 多線程的軟件設計方法確實可以最大限度地發揮

深入理解Java線程池：ThreadPoolExecutor

時間 生命 empty nec 通過 進制 以及 ram 獲取鎖 線程池介紹在web開發中，服務器需要接受並處理請求，所以會為一個請求來分配一個線程來進行處理。如果每次請求都新創建一個線程的話實現起來非常簡便，但是存在一個問題： 如果並發的請求數量非常多，但每個線程執行的時

Java線程池可用的隊列

許可 pri 工作隊列 locking 默認 多線程 ray handler cpu使用率 Java線程池ThreadPoolExecutor的構造器： public ThreadPoolExecutor(int corePoolSize,

Java線程池ThreadPoolExecutor&&Executors

同步 amp keep inter 拒絕 順序 fifo 重點 edt 一、先看看傳統的開啟線程。 new Thread(new Runnable() { @Override public void run() { } }).start

Java線程池的底層實現與使用

線程 了解 hand 通過 rem 高程 滿了 nbsp 被調用 正文 前言 　　在我們進行開發的時候，為了充分利用系統資源，我們通常會進行多線程開發，實現起來非常簡單，需要使用線程的時候就去創建一個線程（繼承Thread類、實現Runnable接口、使用Callabl

協同過濾中item-based與user-based選擇依據

 此文總結的不錯。 協同過濾是大家熟知的推薦演算法。 總的來說協同過濾又可以分為以下兩大類： Neighborhood-based：計算相似item 或user後進行推薦Model-based： 直接訓練模型預測Rating在Neighborhoold-based演算法

線程池工作隊列飽和策略

高負載 java線程池 sca ted 可能 內存耗盡 caller body 策略 線程池工作隊列飽和策略 Java線程池會將提交的任務先置於工作隊列中，在從工作隊列中獲取(SynchronousQueue直接由生產者提交給工作線程)。 那麽工作隊列就有兩種實現策略：

跟我學Java多線程——線程池與堵塞隊列

信號 線程的創建 margin cit rect weight offer 成功 rain 前言 上一篇文章中我們將ThreadPoolExecutor進行了深入的學習和介紹，實際上我們在項目中應用的時候非常少有直接應用ThreadPoolExecutor來

stl中的transform（）註意其與for_each的不同點（有無返回值）

bind2nd n) oid highlight tlist _each pre stream gin #include<iostream> using namespace std; #include"vector" #include"algorithm" #

Java - "JUC線程池" 線程狀態與拒絕策略源碼分析

int 簡單 his 類型 post 由於 lec java turn Java多線程系列--“JUC線程池”04之 線程池原理(三) 本章介紹線程池的生命周期。在"Java多線程系列--“基礎篇”01之 基本概念"中，我們介紹過，線程有5種狀態：新建狀態，就緒狀態，運