1. 程式人生 > 其它 >ThreadPoolExecuter執行緒池詳解

ThreadPoolExecuter執行緒池詳解

一、原始碼分析步驟
  1.掌握執行緒池工作狀態
  2.任務線上程池工作流程
  3.執行緒池任務等待佇列管理策略
  4.執行緒池任務拒絕策略
  5.執行緒池關閉策略
  6.執行緒池初始化策略
  7.執行緒池中執行緒數量擴容方案
二、執行緒池工作狀態
  1.工作狀態分類:
    1)Running:線上程池被建立時就處於【Running】狀態,在這種狀態下執行緒池可以接收新的任務,也可以任務佇列已經存在的任務來進行處理
    2)Shutdown:執行緒池不會再接收新的任務,此時只會對任務佇列中處於等待的任務進行處理
    3)Stop:執行緒池被外部的一段程式終止,此時執行緒池既不能接收新的任務,也不能對佇列等待的任務進行處理
    4)Tidying:執行緒池處於空閒狀態
    5)Terminated:執行緒池工作結束
  2.工作狀態轉換規律:
    1)Running -> Shutdown:顯示呼叫shutdown(); 隱式呼叫finalize(),這個方法內部回撥用shutdown()
    2)Running or Shutdown -> Stop:顯示呼叫shutdownNow()
    3)Shutdown -> Tidying:任務等待佇列中任務都執行完畢時,此時執行緒池活動的執行緒達到了0,執行緒池工作狀態自動轉變為Tidying
    4)Stop -> Tidying:任務等待佇列中任務都執行完畢時,此時執行緒池活動的執行緒達到了0,執行緒池工作狀態自動轉變為Tidying
    5)Tidying -> Terminated:顯示呼叫terminated()
三、通過執行緒池中屬性瞭解執行緒池工作細節
  1.private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    *** ctl儲存執行緒池【狀態】和【執行緒數量】
  2.private final BlockingQueue<Runnable> workQueue;
    *** workQueue是【任務阻塞佇列】,線上程池接收新任務後,如果發現【核心執行緒】都在工作,此時執行緒池就會將新任務新增到【任務阻塞佇列】
  3.private final ReentrantLock mainLock = new ReentrantLock();
    *** 執行緒池中來處理併發時使用的鎖ReentrantLock
  4.private int largestPoolSize;
    *** 記錄執行緒池在工作過程中出現過最大執行緒數。這個屬性與執行緒池可以管理最大執行緒數量,和執行緒數量沒有任何關係
  5.private long completedTaskCount;
    *** 執行緒池已經完成的任務數量
  6.private volatile ThreadFactory threadFactory;
    *** 執行緒池執行緒工廠,負責生成執行緒
  7.private volatile RejectedExecutionHandler handler;
    *** 執行緒池拒絕策略
      1) 無法處理任務,丟擲異常
      2) 放棄一個任務,讓空閒執行緒來處理當前任務
      3) 放棄當前任務
  8.private volatile long keepAliveTime;
    *** 存放執行緒的最大空閒時間。如果一個執行緒空閒時間達到了最大空閒值,此時執行緒池將負責銷燬掉當前執行緒
  9.private volatile int corePoolSize;
    *** 執行緒池核心執行緒數量
  10.private volatile int maximumPoolSize;
    *** 執行緒池可以存放最大執行緒數量
四、ThreadPoolExecuter原始碼分析----execute
  1.作用:用於將任務調給ThreadPoolExecuter進行處理
  2.流程:
    1).判斷當前核心執行緒數量<規定核心執行緒數量
      新增一個核心執行緒處理當前任務
    2).判斷當前核心執行緒數量==規定核心執行緒數量
      2.1 如果執行緒池工作狀態==Running(可以接收新任務)
      2.1.1 將任務成功新增到【任務佇列】
      2.1.1.1 第二次檢測執行緒池工作狀態 == Runnding
        建立一個執行緒處理當前任務
      2.1.1.2 第二次檢測執行緒池工作狀態 != Runnding
        將任務從任務佇列中刪除,呼叫拒絕策略
      2.1.2 將任務新增到【任務佇列】失敗
        嘗試建立一個新執行緒去處理這個任務,如果建立失敗就呼叫拒絕策略
      2.2 如果執行緒池工作狀態!=Running(不可以接收新任務)
五、ThreadPoolExecuter原始碼分析----addWorker
  1.作用:新增一個執行緒處理任務
  2.引數:private boolean addWorker(Runnable firstTask, boolean core)
     @Param Runnable firstTask ----> 當前處理任務
     @Param boolean core ----> ture:判斷執行緒數量<核心執行緒數量;false:判斷執行緒數量<執行緒池最大執行緒數量
  3.業務:addWorker方法程式碼分為兩塊
    1) 雙重迴圈:通過CAS試圖增加執行緒池數量
    2) 保證執行緒併發的安全情況下,實現將任務交給執行緒執行
  4.雙重迴圈:通過CAS試圖增加執行緒池數量
      runStateAtLeast(c, SHUTDOWN) && (runStateAtLeast(c, STOP) || firstTask != null || workQueue.isEmpty())
      當前狀態>=SHUTDOWN 並且 (當前狀態>=STOP 或 firstTask != null 或 workQueue為空)
      1) 執行緒池工作狀態是STOP,TIDYING TERMATED 返回false
      2) 執行緒池工作狀態是SHUTDOWN 並且任務是null 返回false
      3) 執行緒池工作狀態是SHUTDOWN 並且任務是空 返回false
六、ThreadPoolExecuter原始碼分析----shutDown
  1、作用:將執行緒池工作狀態轉變為【Shutdown】狀態,此時執行緒池不再接收新任務。將委派執行緒池已有執行緒對工作佇列中任務進行處理
  2、步驟:
    1) 加鎖
    2) 檢測執行緒(worker)許可權:
      當前執行緒(worker)是否具有關閉執行緒池許可權
      當前執行緒(worker)是否有能力中斷其它執行緒許可權
      如果沒有上述許可權,此時可能丟擲NullPointerException或者安全異常
    3) 通過原子性操作修改執行緒池工作狀態--> shutDown
    4) 將空閒執行緒進行終端(interrupt)
    5) 如果線上程池已經轉變為【Shutdown】狀態時,嘗試將執行緒池狀態直接轉變為【Terminated】
七、ThreadPoolExecuter原始碼分析----shutdownNow
  1、方法格式:public List<Runnable> shutdownNow()
  2、方法作用:shutdownNow方法執行時
    1) 將執行緒池工作狀態轉變為【stop】狀態
    2) 此時執行緒池不再接收新的任務
    3) 此時執行緒池要將所有正在執行的任務進行結束
    4) 此時執行緒池要將所有剩餘任務從任務佇列中清除
  3、返回值含義:
    List<Runnable>:返回從任務列表中清除的任務
  4、執行步驟:
    1) 加鎖
    2) 檢查當前執行緒(worker)許可權:
      檢測當前執行緒(worker)是否具備修改執行緒池工作狀態許可權
      檢測當前執行緒(worker)是否具備中斷其她執行緒許可權
      如果沒有上述許可權,此時可能丟擲NullPointerException或者安全異常
    3) 修改執行緒池工作狀態為【stop】
    4) 將所有執行緒進行中斷
    5) 將任務佇列中的任務進行清除
    6) 嘗試將執行緒池工作狀態直接轉變為【Terminated】
    7) 解鎖
    8) 將任務佇列中的任務返回
八、等待佇列管理策略
  1、佇列:執行緒池中任務管理模式,實際上就是一個數組儲存需要被處理的任務
  2、策略:
    1) 不放入任務佇列,直接交給核心執行緒執行 (執行執行緒數量 < 核心執行緒數量)
    2) 放入到任務佇列: (執行執行緒數量 >= 核心執行緒數量)
      情況1: 核心執行緒數量 == 最大執行緒數量,佇列沒有設定存放任務最大上限
      情況2: 核心執行緒數量 == 最大執行緒數量,佇列存放任務數量小於佇列允許存放最大任務數量
    3) 不放入任務佇列,建立一個新的執行緒去處理任務:(執行執行緒數量 >= 核心執行緒數量) (執行執行緒數量 < 最大執行緒數量)
    4) 不放入任務佇列,並呼叫拒絕策略阻止任務處理:(執行執行緒數量 >= 核心執行緒數量)
      情況1:任務佇列存放任務數量 == 任務佇列允許存放最大任務數量,此時執行執行緒數量 >= 執行緒池允許出現最大執行緒數量
      情況2:任務佇列存放任務數量 < 任務佇列允許存放最大任務數量,此時執行執行緒數量 >= 執行緒池允許出現最大執行緒數量
      情況3:線上程池工作狀態為【shutDown】或者呼叫shutDown方法或者呼叫shutdownNow方法
九、執行緒池阻塞佇列分類和選擇
  1、執行緒池中阻塞佇列:
    BolckingQueue workQueue
  2、執行緒池中阻塞佇列分類:
    1) 無界阻塞佇列
    2) 有界阻塞佇列
    3) 同步移交阻塞佇列
  3、無界阻塞佇列:
    不會限制儲存任務數量的上限,執行緒池中最常用的無界阻塞佇列----LinkedBlockingQueue
    隱患:如果當前任務執行時間相對較長,導致大量新任務堆積到佇列中,有可能造成服務端OOM問題
    不適用場景:在網際網路通訊過程中,如果QPS(單位時間裡的訪問量)較高並且傳送資料量較大,此時如果使用無界阻塞佇列來儲存任務,導致服務端出現熔斷現象
    Executore.newFixThreaPool預設使用的阻塞佇列----LinkedBlockingQueue
  4、有界阻塞佇列:
    設定佇列中所儲存的任務上限,如果達到了上限則需要通過執行緒池拒絕策略/飽和策略來拒絕接收這個任務
    基於FIFO原則來實現有界阻塞佇列----ArrayBlockingQueue
    基於任務優先順序原則實現有界阻塞佇列----PriorityBlockingQueue
  5、同步移交阻塞佇列:
    並不是一種真實的阻塞佇列
    執行緒池中最常用同步移交阻塞佇列----SynchronousQueue
    同步移交阻塞佇列接收到一個新任務要去執行緒池,此時必須存在一個執行緒來處理這個任務。
    線上程池發現執行執行緒數量 < 執行緒池核心執行緒數量,直接將任務交給【SynchronousQueue】或者執行執行緒數量 <執行緒池最大執行緒數量,執行緒池會建立一個新執行緒,然後將任務交給【SynchronousQueue】,由同步移交阻塞佇列將新任務交給新執行緒處理
十、執行緒池飽和策略
  1、介紹
    飽和策略也叫拒絕策略
    在新任務數量大於執行緒池中有界阻塞佇列上限時,此時可以通過飽和策略來對多出的任務進行處理,這些方案統稱飽和策略
    線上程池中飽和策略資料型別RejectHandler
  2、飽和策略分類:
    1) AbortPolicy方案
    2) DiscardPolicy方案
    3) DiscardOldestPolicy方案
    4) CallerRunsPolicy方案
  3、AbortPolicy方案:
    1) 是資料庫連線池預設的飽和方案
    2) 對於不能存放到阻塞佇列中的新任務會直接刪除並丟擲一個RejectedExecutionException異常,這個異常可以通過catch語句捕捉到
    3) 對應程式碼
      public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
        " rejected from " +
        e.toString());
      }
  4、DiscardPolicy方案:
    1) 與AbortPolicy方案相似,也是不處理不能存入到阻塞佇列中的新任務,只不過不會丟擲異常
    2) 對應程式碼
      public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
      }
  5、DiscardOldestPolicy方案:
    1) 在阻塞佇列數量已經達到了任務上限後,為了讓新任務進入到阻塞佇列中,DiscardOldestPolicy方案將阻塞佇列中最早的任務/最老的任務從阻塞佇列中刪除,將新的任務新增到阻塞佇列中
    2) 對應程式碼
      public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
          e.getQueue().poll();//彈出最早的任務/最老的任務
          e.execute(r);//新增新任務
        }
      }
  6、CallerRunsPolicy方案:
    1) 目的還是要處理新任務
    2) 不會使用執行緒池中執行緒來處理當前任務
    3) 將這個任務返回給提交任務的執行緒去執行
    4) 對應程式碼
      public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
          r.run();
        }
      }
十一、執行緒池關閉原則----執行緒中斷方案
  1、不應該手動中斷任何一個執行緒執行
  2、如果執行緒正在持有鎖,此時中斷執行緒執行,導致鎖不會釋放
  3、如果執行緒正在通過I/O流操作資料庫或者是文件,此時中斷執行緒執行,導致資料不一致
  4、所以JDK中將用於結束執行緒的方法stop方法設定為過時方法
  5、執行緒結束應該由執行緒自身來決定
十二、執行緒池關閉原則----interrupt方法
  1、interrupt方法本質上並不是結束一個執行緒,用於設定執行緒中狀態
  2、如果此時執行緒正處於阻塞佇列(wait,I/O等待),此時interrupt方法執行,會結束執行緒阻塞狀態並丟擲InterruptException異常,這個異常在catch中進行捕捉,進行處理後,執行緒自動結束
  3、如果此時執行緒正處於執行狀態,此時interrupt方法執行並不會阻止執行緒繼續執行。應該線上程執行中適當位置呼叫isInterrupt方法,來讓執行緒瞭解是否被設定中斷狀態,然後根據實際情況由執行緒決定市場還要繼續執行
  4、如果先執行interrupt方法。在將執行緒放入阻塞佇列中,此時依然會丟擲InterruptException異常
十三、執行緒池關閉原則----依賴條件
  1、如果執行緒池沒有繼續被物件引用,導致執行緒池嘗試關閉自己
  2、如果執行緒池沒有任何執行緒時,導致執行緒池關閉自己
十四、執行緒池關閉原則----shutdown與 shutdownNow
  1、shutdown方法:禁止執行緒池接收新任務
    要求執行緒池將阻塞佇列中的任務依次執行
    在任務執行完畢後,將執行緒依次銷燬
    在所有執行緒都銷燬完畢後,執行緒池關閉
  2、shutdownNow方法:禁止執行緒池接收新任務
    要求執行緒池將阻塞佇列任務進行依次刪除
    在任務刪除完畢後,將執行緒依次銷燬
    在所有執行緒都銷燬完畢後,執行緒池關閉
    ***shutdown與shutdownNow方法執行後,執行緒池並不會立刻關閉
    所以採用非同步通知方案,如果需要在這兩個方法執行後要求執行緒池立刻關閉,此時需要採用同步方案,因此需要呼叫terminated
十五、執行緒池關閉原則----shutdown與 shutdownNow執行特徵
  1、需要了解問題:
    1) 在shutdown與shutdownNow方法執行後,執行緒池是否還會接收新任務
    2) 在shutdown與shutdownNow方法執行後,阻塞佇列中的任務是否繼續執行
    3) 在shutdown與shutdownNow方法執行後,執行緒池中正在執行的任務會不會被中斷
  2、執行緒池關閉原則:
    1) 線上程池關閉時,是不會再接收新任務,此時對於傳遞過來新任務交給飽和原則(AbortPolicy[預設],DiscardPolicy,DiscardOldestPolicy,CallerRunsPolicy)進行處理
    2) 如果由shutdown方法通知執行緒池關閉,此時執行緒池會將阻塞佇列中所有任務執行完畢後才會關閉;如果由shutdownNow方法通知執行緒池關閉,此時執行緒池不會執行阻塞佇列中的任務,並且將阻塞佇列中的任務刪除
    3) 線上程池接收到關閉命令時,正在執行的執行緒是不會被中斷,可以繼續執行完畢
十六、執行緒池關閉原則----FixedThreadPool與CachedThreadPool沒有關閉執行緒池特徵
  1、FixedThreadPool:
    沒有設定執行緒超時策略,因此線上程池任務結束後,執行緒池中執行緒依然存在,如果不去手動關閉執行緒池,在多次使用之後會造成OOM【記憶體洩露】問題
  2、CachedThreadPool:
    預設執行緒超時策略,預設時間60s,因此CachedThreadPool在執行完成後即使沒有關閉執行緒池,也不會導致OOM問題
十七、執行緒池啟動原則
  1、線上程池物件建立完成後,不會自動執行
  2、線上程池建立時,想阻塞佇列新增任務,也不會導致執行緒池自動執行
  3、只有通過submit或者execute方法向執行緒池提交任務,才會促使執行緒池執行
    1) 建立一個全新執行緒獨立當前任務
      情況1:(執行執行緒數量 < 核心執行緒數量)新執行緒處理任務
      情況2:(執行緒池最大執行緒數量 >= 執行執行緒數量 >= 核心執行緒數量)
      同時(阻塞佇列以滿),新執行緒處理任務
    2) 將任務暫時存放到阻塞佇列中等待執行緒空閒時處理
      情況1:(阻塞佇列未滿)(執行執行緒數量 == 核心執行緒數量 < 執行緒池最大執行緒數量)將任務新增到阻塞佇列
      情況2:(阻塞佇列未滿)(執行執行緒數量 >= 核心執行緒數量 == 執行緒池最大執行緒數量) 將任務新增到阻塞佇列
      情況3:(阻塞佇列已滿)(執行執行緒數量 == 執行緒池最大執行緒數量) 但是執行緒池使用的是【DiscardOldestPolicy飽和策略】,將阻塞佇列最老任務刪除,新任務新增到阻塞佇列尾部
    3) 不處理當前任務:
      情況1:(阻塞佇列已滿)(執行執行緒數量 == 執行緒池最大執行緒數量)如果執行緒池採用飽和策略【AbortPolicy,DiscardPolicy】此時執行緒池將拒絕處理當前任務;如果採用【AbortPolicy飽和策略】,不僅不處理當前任務同時還會向上丟擲RejectExecuteException異常,如果採用【DiscardPolicy飽和策略】,只會不處理當前任務
      情況2:無論阻塞佇列是否已滿,執行執行緒數量是否大於等於執行緒池最大執行緒數量,只要執行緒池工作狀態為【shutdown】或者【stop】狀態,都會導致執行緒池不再接收新任務
    4) 執行緒池不會呼叫自己執行緒來處理當前任務,而是將任務交給提交任務的執行緒去處理
      情況1:(阻塞佇列已滿)(執行執行緒數量 >= 執行緒池最大執行緒數量) 執行緒池採用【CallerRunsPolicy飽和策略】,導致執行緒池不會呼叫自身的執行緒來處理當前任務,將新任務返回給發起請求的執行緒,由發起請求的執行緒處理
十八、執行緒池擴容策略----集合擴容
  集合 資料結構 載入因子 擴容因子
  ArrayList 陣列 0.5(元素個數超過ArrayList存放資料0.5倍) 0.5(擴容後是原來的1.5倍)
  Vector 陣列 0.5(元素個數超過Vector存放資料0.5倍) 0.5(擴容後是原來的1.5倍)
  HashSet HashMap 0.75(元素個數超過HashSet存放資料個數) 1(擴容後是原來的2倍)
  HashMap 散列表 0.75(元素個數超過HashMap存放資料個數) 1(擴容後是原來的2倍)
  HashTable 散列表 0.75(元素個數超過HashTable存放資料個數) 1+1(擴容後是原來的2倍+1)
十九、執行緒池擴容策略----執行緒擴容
  1、執行緒池採用HashSet儲存執行緒
  2、執行緒池在什麼情況下會新增執行緒
    情況1:(執行執行緒數量 < 核心執行緒數量)此時執行緒池會新增一個新執行緒,並存到HashSet中來處理當前任務
    情況2:(阻塞佇列已滿)(執行執行緒數量 < 執行緒池最大執行緒數量)此時執行緒池會新增一個新執行緒,並存到HashSet中來處理當前任務
二十、四種常見執行緒池----FixedThreadPool
  1、FixedThreadPool由Executros提供的一個ThreadPoolExecutor的例項
  2、FixedThreadPool是一個定長的執行緒池,執行緒池中執行緒池數量是固定不能增加的,核心執行緒數量與最大執行緒數量相等
  3、FixedThreadPool沒有設定執行緒空閒時間,因此執行緒工作完畢後是不會被銷燬,因此FixedThreadPool在執行完畢後,必須通過手動呼叫shutdown方法關閉執行緒池,避免出現OOM【記憶體洩露】問題
  4、FixedThreadPool使用的是LinkedBlockingQueue作為阻塞佇列,但是沒有設定佇列任務上限,因此FixedThreadPool可以無限存放任務,但是也帶來了OOM【記憶體洩露】問題
二十一、四種常見執行緒池----CachedThreadPool
  1、CachedThreadPool由Executros提供的一個ThreadPoolExecutor的例項
  2、CachedThreadPool被稱為快取執行緒池,如果執行緒池有空閒執行緒則複用執行緒處理任務,如果沒有空閒執行緒則新增執行緒來處理當前任務
  3、CachedThreadPool沒有核心執行緒,可以接收最大執行緒數量Integer.MAX_VALUE
  4、CachedThreadPool預設設定空閒時間60s,因此理論上來說CachedThreadPool在工作完畢後,可以不用手動呼叫shutdown也會被關閉
  5、CachedThreadPool使用的是SynchronousQueue<Runnable>作為阻塞佇列(同步移交阻塞佇列),因此不會將任務儲存到阻塞佇列,意味著第一插入的任務沒有執行完畢之前,下一個任務是不能執行的
二十二、四種常見執行緒池----SingleThreadPool
  1、SingleThreadPool由Executros提供的一個ThreadPoolExecutor的例項
  2、SingleThreadPool被稱為單一執行緒池
  3、SingleThreadPool中核心執行緒數量固定是1,最大執行緒數量固定1,在執行期間至始至終只使用1個執行緒來處理所有任務
  4、SingleThreadPool沒有設定執行緒空閒時間,因此在SingleThreadPool執行緒池工作完畢後,這個執行緒不會自動銷燬,因此必須通過手動呼叫shutdown方法關閉執行緒池
  5、SingleThreadPool使用的是LinkedBlockingQueue作為阻塞佇列,可以接收任意數量的任務
二十三、四種常見執行緒池----ScheduledThreadPool
*
*
*
二十四、SingleThreadPool與FixedThreadPool區別
  1、SingleThreadPool:
    1) 線上程池中只使用1個執行緒處理任務
    2) 能保證任務執行順序,先提交先執行
    3) 當執行緒執行時丟擲異常,SingleThreadPool會自動建立一個新執行緒來代替舊的執行緒
  2、FixedThreadPool:
    1) 指定執行緒池可以建立最大執行緒數量
    2) 如果執行緒池中執行緒數量達到最大執行緒數量,此時不會新增執行緒
    3) Executors.newFixedThreadPool(1),當這個唯一執行緒出現了異常,此時FixedThreadPool也會自動建立一個新執行緒來代替舊的執行緒