當(dāng)創(chuàng)建的線程數(shù)等于corePoolSize，會將任務(wù)加入阻塞隊列(BlockingQueue)，維護(hù)著等待執(zhí)行的Runnable對象。

阻塞隊列通常有如下類型：

• ArrayBlockingQueue ：一個由數(shù)組結(jié)構(gòu)組成的有界阻塞隊列。可以限定隊列的長度，接收到任務(wù)時，如果沒有達(dá)到corePoolSize的值，則新建線程(核心線程)執(zhí)行任務(wù)，如果達(dá)到了，則入隊等候，如果隊列已滿，則新建線程(非核心線程)執(zhí)行任務(wù)，又如果總線程數(shù)到了maximumPoolSize，并且隊列也滿了，則發(fā)生錯誤。
• LinkedBlockingQueue ：一個由鏈表結(jié)構(gòu)組成的有界阻塞隊列。這個隊列在接收到任務(wù)時，如果當(dāng)前線程數(shù)小于核心線程數(shù)，則新建線程(核心線程)處理任務(wù);如果當(dāng)前線程數(shù)等于核心線程數(shù)，則進(jìn)入隊列等待。由于這個隊列沒有最大值限制，即所有超過核心線程數(shù)的任務(wù)都將被添加到隊列中，這也就導(dǎo)致了maximumPoolSize的設(shè)定失效，因為總線程數(shù)永遠(yuǎn)不會超過corePoolSize。
• PriorityBlockingQueue ：一個支持優(yōu)先級排序的無界阻塞隊列。
• DelayQueue：一個使用優(yōu)先級隊列實現(xiàn)的無界阻塞隊列。隊列內(nèi)元素必須實現(xiàn)Delayed接口，這就意味著傳入的任務(wù)必須先實現(xiàn)Delayed接口。這個隊列在接收到任務(wù)時，首先先入隊，只有達(dá)到了指定的延時時間，才會執(zhí)行任務(wù)。
• SynchronousQueue：一個不存儲元素的阻塞隊列。這個隊列在接收到任務(wù)時，會直接提交給線程處理，而不保留它，如果所有線程都在工作就新建一個線程來處理這個任務(wù)。所以為了保證不出現(xiàn)【線程數(shù)達(dá)到了maximumPoolSize而不能新建線程】的錯誤，使用這個類型隊列時，maximumPoolSize一般指定成Integer.MAX_VALUE，即無限大。
• LinkedTransferQueue：一個由鏈表結(jié)構(gòu)組成的無界阻塞隊列。
• LinkedBlockingDeque：一個由鏈表結(jié)構(gòu)組成的雙向阻塞隊列。

明確的拒絕策略

當(dāng)任務(wù)處理不過來時，線程池開始執(zhí)行拒絕策略。

支持的拒絕策略：

• ThreadPoolExecutor.AbortPolicy: 丟棄任務(wù)并拋出RejectedExecutionException異常。(默認(rèn))
• ThreadPoolExecutor.DiscardPolicy：也是丟棄任務(wù)，但是不拋出異常。
• ThreadPoolExecutor.DiscardOldestPolicy：丟棄隊列最前面的任務(wù)，然后重新嘗試執(zhí)行任務(wù)。(重復(fù)此過程)
• ThreadPoolExecutor.CallerRunsPolicy：由調(diào)用線程處理該任務(wù)。

線程池關(guān)閉

• shutdown：將線程池狀態(tài)置為SHUTDOWN，并不會立即停止。停止接收外部submit的任務(wù)，內(nèi)部正在跑的任務(wù)和隊列里等待的任務(wù)，會執(zhí)行完后，才真正停止。
• shutdownNow：將線程池狀態(tài)置為STOP。企圖立即停止，事實上不一定，跟shutdown()一樣，先停止接收外部提交的任務(wù)，忽略隊列里等待的任務(wù)，嘗試將正在跑的任務(wù)interrupt中斷(如果線程未處于sleep、wait、condition、定時鎖狀態(tài)，interrupt無法中斷當(dāng)前線程)，返回未執(zhí)行的任務(wù)列表。
• awaitTermination(long timeOut, TimeUnit unit)當(dāng)前線程阻塞，直到等所有已提交的任務(wù)(包括正在跑的和隊列中等待的)執(zhí)行完或者等超時時間到或者線程被中斷，拋出InterruptedException，然后返回true(shutdown請求后所有任務(wù)執(zhí)行完畢)或false(已超時)。

Executors

Executors是一個幫助類，提供了創(chuàng)建幾種預(yù)配置線程池實例的方法：newSingleThreadExecutor、newFixedThreadPool、newCachedThreadPool等。

如果查看源碼就會發(fā)現(xiàn)，Executors本質(zhì)上就是實現(xiàn)了幾類默認(rèn)的ThreadPoolExecutor。而阿里巴巴開發(fā)手冊，不建議采用Executors默認(rèn)的，讓使用者直接通過ThreadPoolExecutor來創(chuàng)建。

Executors.newSingleThreadExecutor()

創(chuàng)建一個單線程的線程池。這個線程池只有一個線程在工作，也就是相當(dāng)于單線程串行執(zhí)行所有任務(wù)。如果這個唯一的線程因為異常結(jié)束，那么會有一個新的線程來替代它。此線程池保證所有任務(wù)的執(zhí)行順序按照任務(wù)的提交順序執(zhí)行。

new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()) 

該類型線程池的結(jié)構(gòu)圖：

newSingleThreadExecutor

該線程池的特點：

• 只會創(chuàng)建一條工作線程處理任務(wù);
• 采用的阻塞隊列為LinkedBlockingQueue;

Executors.newFixedThreadPool()

創(chuàng)建固定大小的線程池。每次提交一個任務(wù)就創(chuàng)建一個線程，直到線程達(dá)到線程池的最大大小。線程池的大小一旦達(dá)到最大值就會保持不變，如果某個線程因為執(zhí)行異常而結(jié)束，那么線程池會補(bǔ)充一個新線程。

new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); 

該類型線程池的結(jié)構(gòu)圖：

newFixedThreadPool

該線程池的特點：

• 固定大小;
• corePoolSize和maximunPoolSize都為用戶設(shè)定的線程數(shù)量nThreads;
• keepAliveTime為0，意味著一旦有多余的空閑線程，就會被立即停止掉;但這里keepAliveTime無效;
• 阻塞隊列采用了LinkedBlockingQueue，一個無界隊列;
• 由于阻塞隊列是一個無界隊列，因此永遠(yuǎn)不可能拒絕任務(wù);
• 由于采用了無界隊列，實際線程數(shù)量將永遠(yuǎn)維持在nThreads，因此maximumPoolSize和keepAliveTime將無效。

Executors.newCachedThreadPool()

創(chuàng)建一個可緩存的線程池。如果線程池的大小超過了處理任務(wù)所需要的線程，那么就會回收部分空閑(60秒不執(zhí)行任務(wù))的線程，當(dāng)任務(wù)數(shù)增加時，此線程池又可以智能的添加新線程來處理任務(wù)。此線程池不會對線程池大小做限制，線程池大小完全依賴于操作系統(tǒng)(或者說JVM)能夠創(chuàng)建的最大線程大小。

new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>()); 

該類型線程池的結(jié)構(gòu)圖：

newCachedThreadPool

• 該線程池的特點：
• 可以無限擴(kuò)大;
• 比較適合處理執(zhí)行時間比較小的任務(wù);
• corePoolSize為0，maximumPoolSize為無限大，意味著線程數(shù)量可以無限大;
• keepAliveTime為60S，意味著線程空閑時間超過60s就會被殺死;
• 采用SynchronousQueue裝等待的任務(wù)，這個阻塞隊列沒有存儲空間，這意味著只要有請求到來，就必須要找到一條工作線程處理它，如果當(dāng)前沒有空閑的線程，那么就會再創(chuàng)建一條新的線程。

Executors.newScheduledThreadPool()

創(chuàng)建一個定長線程池，支持定時及周期性任務(wù)執(zhí)行。

new ThreadPoolExecutor(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, 
              new DelayedWorkQueue()); 

該線程池類圖：

newScheduledThreadPool

該線程池的特點：

• 接收SchduledFutureTask類型的任務(wù)，有兩種提交任務(wù)的方式：scheduledAtFixedRate和scheduledWithFixedDelay。SchduledFutureTask接收的參數(shù)：
• time：任務(wù)開始的時間
• sequenceNumber：任務(wù)的序號
• period：任務(wù)執(zhí)行的時間間隔
• 采用DelayQueue存儲等待的任務(wù);
• DelayQueue內(nèi)部封裝了一個PriorityQueue，它會根據(jù)time的先后時間排序，若time相同則根據(jù)sequenceNumber排序;
• DelayQueue也是一個無界隊列;
• 工作線程執(zhí)行時，工作線程會從DelayQueue取已經(jīng)到期的任務(wù)去執(zhí)行;執(zhí)行結(jié)束后重新設(shè)置任務(wù)的到期時間，再次放回DelayQueue;

Executors.newWorkStealingPool()

JDK8引入，創(chuàng)建持有足夠線程的線程池支持給定的并行度，并通過使用多個隊列減少競爭。

public static ExecutorService newWorkStealingPool() { 
    return new ForkJoinPool(Runtime.getRuntime().availableProcessors(), 
        ForkJoinPool.defaultForkJoinWorkerThreadFactory, 
        null, true); 
} 

Executors方法的弊端

1)newFixedThreadPool 和 newSingleThreadExecutor：允許的請求隊列長度為Integer.MAX_VALUE，可能會堆積大量的請求，從而導(dǎo)致 OOM。2)newCachedThreadPool 和 newScheduledThreadPool：允許的創(chuàng)建線程數(shù)量為Integer.MAX_VALUE，可能會創(chuàng)建大量的線程，從而導(dǎo)致 OOM。

合理配置線程池大小

合理配置線程池，需要先分析任務(wù)特性，可以從以下角度來進(jìn)行分析：

• 任務(wù)的性質(zhì)：CPU密集型任務(wù)，IO密集型任務(wù)和混合型任務(wù)。
• 任務(wù)的優(yōu)先級：高，中和低。
• 任務(wù)的執(zhí)行時間：長，中和短。
• 任務(wù)的依賴性：是否依賴其他系統(tǒng)資源，如數(shù)據(jù)庫連接。

另外，還需要查看系統(tǒng)的內(nèi)核數(shù)：

Runtime.getRuntime().availableProcessors()); 

根據(jù)任務(wù)所需要的CPU和IO資源可以分為：

• CPU密集型任務(wù): 主要是執(zhí)行計算任務(wù)，響應(yīng)時間很快，CPU一直在運行。一般公式：線程數(shù) = CPU核數(shù) + 1。只有在真正的多核CPU上才能得到加速，優(yōu)點是不存在線程切換開銷，提高了CPU的利用率并減少了線程切換的效能損耗。
• IO密集型任務(wù)：主要是進(jìn)行IO操作，CPU并不是一直在執(zhí)行任務(wù)，IO操作(CPU空閑狀態(tài))的時間較長，應(yīng)配置盡可能多的線程，其中的線程在IO操作時，其他線程可以繼續(xù)利用CPU，從而提高CPU的利用率。一般公式：線程數(shù) = CPU核數(shù) * 2。

使用實例

任務(wù)實現(xiàn)類：

/** 
 * 任務(wù)實現(xiàn)線程 
 * @author sec 
 * @version 1.0 
 * @date 2021/10/30 
 **/ 
public class MyThread implements Runnable{ 
 
   private final Integer number; 
 
   public MyThread(int number){ 
      this.number = number; 
   } 
 
   public Integer getNumber() { 
      return number; 
   } 
 
   @Override 
   public void run() { 
      try { 
         // 業(yè)務(wù)處理 
         TimeUnit.SECONDS.sleep(1); 
         System.out.println("Hello! ThreadPoolExecutor - " + getNumber()); 
      } catch (InterruptedException e) { 
         e.printStackTrace(); 
      } 
   } 
} 

自定義阻塞提交的ThreadLocalExcutor：

/** 
 * 自定義阻塞提交的ThreadPoolExecutor 
 * @author sec 
 * @version 1.0 
 * @date 2021/10/30 
 **/ 
public class CustomBlockThreadPoolExecutor { 
 
   private ThreadPoolExecutor pool = null; 
 
   /** 
    * 線程池初始化方法 
    */ 
   public void init() { 
      // 核心線程池大小 
      int poolSize = 2; 
      // 最大線程池大小 
      int maxPoolSize = 4; 
      // 線程池中超過corePoolSize數(shù)目的空閑線程最大存活時間：30+單位TimeUnit 
      long keepAliveTime = 30L; 
      // ArrayBlockingQueue<Runnable> 阻塞隊列容量30 
      int arrayBlockingQueueSize = 30; 
      pool = new ThreadPoolExecutor(poolSize, maxPoolSize, keepAliveTime, 
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(arrayBlockingQueueSize), new CustomThreadFactory(), 
            new CustomRejectedExecutionHandler()); 
   } 
 
   /** 
    * 關(guān)閉線程池方法 
    */ 
   public void destroy() { 
      if (pool != null) { 
         pool.shutdownNow(); 
      } 
   } 
 
   public ExecutorService getCustomThreadPoolExecutor() { 
      return this.pool; 
   } 
 
   /** 
    * 自定義線程工廠類， 
    * 生成的線程名詞前綴、是否為守護(hù)線程以及優(yōu)先級等 
    */ 
   private static class CustomThreadFactory implements ThreadFactory { 
 
      private final AtomicInteger count = new AtomicInteger(0); 
 
      @Override 
      public Thread newThread(Runnable r) { 
         Thread t = new Thread(r); 
         String threadName = CustomBlockThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1); 
         t.setName(threadName); 
         return t; 
      } 
   } 
 
 
   /** 
    * 自定義拒絕策略對象 
    */ 
   private static class CustomRejectedExecutionHandler implements RejectedExecutionHandler { 
      @Override 
      public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { 
         // 核心改造點，將blockingqueue的offer改成put阻塞提交 
         try { 
            executor.getQueue().put(r); 
         } catch (InterruptedException e) { 
            e.printStackTrace(); 
         } 
      } 
   } 
 
   /** 
    * 當(dāng)提交任務(wù)被拒絕時，進(jìn)入拒絕機(jī)制，實現(xiàn)拒絕方法，把任務(wù)重新用阻塞提交方法put提交，實現(xiàn)阻塞提交任務(wù)功能，防止隊列過大，OOM 
    */ 
   public static void main(String[] args) { 
 
      CustomBlockThreadPoolExecutor executor = new CustomBlockThreadPoolExecutor(); 
 
      // 初始化 
      executor.init(); 
      ExecutorService pool = executor.getCustomThreadPoolExecutor(); 
      for (int i = 1; i < 51; i++) { 
         MyThread myThread = new MyThread(i); 
         System.out.println("提交第" + i + "個任務(wù)"); 
         pool.execute(myThread); 
      } 
 
      pool.shutdown(); 
      try { 
         // 阻塞，超時時間到或者線程被中斷 
         if (!pool.awaitTermination(60, TimeUnit.SECONDS)) { 
            // 立即關(guān)閉 
            executor.destroy(); 
         } 
      } catch (InterruptedException e) { 
         executor.destroy(); 
      } 
   } 
} 

小結(jié)

看似簡單的線程池創(chuàng)建，其中卻蘊(yùn)含著各類知識，融合貫通，根據(jù)具體場景采用具體的參數(shù)進(jìn)行設(shè)置才能夠達(dá)到最優(yōu)的效果。

總結(jié)一下就是：

• 用ThreadPoolExecutor自定義線程池，要看線程的用途。如果任務(wù)量不大，可以用無界隊列，如果任務(wù)量非常大，要用有界隊列，防止OOM;
• 如果任務(wù)量很大，且要求每個任務(wù)都處理成功，要對提交的任務(wù)進(jìn)行阻塞提交，重寫拒絕機(jī)制，改為阻塞提交。保證不拋棄一個任務(wù);
• 最大線程數(shù)一般設(shè)為2N+1最好，N是CPU核數(shù);
• 核心線程數(shù)，要根據(jù)任務(wù)是CPU密集型，還是IO密集型。同時，如果任務(wù)是一天跑一次，設(shè)置為0合適，因為跑完就停掉了;
• 如果要獲取任務(wù)執(zhí)行結(jié)果，用CompletionService，但是注意，獲取任務(wù)的結(jié)果要重新開一個線程獲取，如果在主線程獲取，就要等任務(wù)都提交后才獲取，就會阻塞大量任務(wù)結(jié)果，隊列過大OOM，所以最好異步開個線程獲取結(jié)果。

參考文章：

[1]https://www.jianshu.com/p/94852bd1a283

[2]https://blog.csdn.net/jek123456/article/details/90601351

[3]https://blog.csdn.net/z_s_z2016/article/details/81674893

[4]https://zhuanlan.zhihu.com/p/33264000 

[5]https://www.cnblogs.com/semi-sub/p/13021786.html