Java執行緒池的總結

• 2020 年 2 月 25 日
• 筆記

閱讀本文約需要10分鐘

今天跟大家分享Java執行緒池的知識。

1 為什麼需要執行緒池

在當今電腦的CPU計算速度非常快的情況下，為了能夠充分利用CPU性能提高程式運行效率我們在程式中使用了執行緒。但是在高並發情況下會頻繁的創建和銷毀執行緒，這樣就變相的阻礙了程式的執行速度，所以為了管理執行緒資源和減少執行緒創建以及銷毀的性能消耗就引入了執行緒池。

2 什麼場景下適合使用執行緒池

當伺服器接收到大量任務時，如果使用執行緒池可以大量減少執行緒的創建與銷毀次數，從而提升程式執行效率

在實際開發中，如果需要創建5個以上的執行緒，那麼就可以使用執行緒池來管理

3 執行緒池參數介紹以及特點

3.1 corePoolSize和maxPoolSize

corePoolSize：執行緒池在創建完時，裡面並沒有執行緒，只有當任務到來時再去創建執行緒。

maxPoolSize：執行緒池可能會在核心執行緒數的基礎上額外增加一些執行緒，但是執行緒數量的上限是maxPoolSize。比如第一天執行的任務非常多，第二天執行的任務非常少，但是有了maxPoolSize參數，就可以增強任務處理的靈活性。

3.2 添加執行緒的規則

當執行緒數量小於corePoolSize即使執行緒沒有在執行任務，也會創建新的執行緒。

如果執行緒數量等於（或大於）corePoolSize，但小於maxPoolSize則將任務放入隊列。

如果隊列已滿，並且執行緒數小於maxPoolSize，則創建新的執行緒運行任務。

如果隊列已滿，並且執行緒數大於或等於maxPoolSize，則拒絕該任務。

執行流程：

3.3 增減執行緒的特點

將corePoolSize和maxPoolSize設置為相同的值，那麼就會創建固定大小的執行緒池。

執行緒池希望保持更少的執行緒數，並且只有在負載變得很大時才會增加它。

如果將執行緒池的maxPoolSize參數設置為很大的值，例如Integer.MAX_VALUE，可以允許執行緒池容納任意數量的並發任務。

只有在隊列滿了的時候才會去創建大於corePoolSize的執行緒，所以如果使用了無界隊列（如：LinkedBlockingQueue）就不會創建到超過corePoolSize的執行緒數。

3.4 keepAliveTime

如果執行緒池當前的執行緒數大於corePoolSize,那麼如果多餘的執行緒的空閑時間大keepAliveTime，它們就會被終止。

keepAliveTime參數的使用可以減少執行緒數過多冗餘時的資源消耗。

3.5 threadFactory

新的執行緒由ThreadFactory創建，默認使用Executors.defaultThreadFactory()，創建出來的執行緒都在同一個執行緒組，擁有同樣的NORM_PRIORITY優先順序並且都不是守護執行緒。如果自己指定ThreadFactory，那麼就可以改變執行緒名、執行緒組、優先順序、是否是守護執行緒等。通常情況下直接使用defaultThreadFactory就行。

3.6 workQueue

直接交接（SynchronousQueue）：任務不多時，只需要用隊列進行簡單的任務中轉，這種隊列無法存儲任務，在使用這種隊列時，需要將maxPoolSize設置的大一點。

無界隊列（LinkedBlockingQueue）：如果使用無界隊列當作workQueue，將maxQueue設置的多大都沒有用，使用無界隊列的優點是可以防止流量突增，缺點是如果處理任務的速度跟不上提交任務的速度，這樣就會導致無界隊列中的任務越來越多，從而導致OOM異常。

有界隊列（ArrayBlockingQueue）：使用有界隊列可以設置隊列大小，讓執行緒池的maxPoolSize有意義。

4 執行緒池應該手動創建還是自動創建

手動創建更好，因為這樣可以讓我們更加了解執行緒池的運行規則，避免資源耗盡的風險。

4.1 直接調用JDK封裝好的執行緒池會帶來的問題

newFixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {        return new ThreadPoolExecutor(nThreads, nThreads,                                      0L, TimeUnit.MILLISECONDS,                                      new LinkedBlockingQueue<Runnable>());    }

newFixedThreadPool執行緒池通過傳入相同的corePoolSize和maxPoolSize可以保證執行緒數量固定，0L的keepAliveTime表示時刻被銷毀，workQueue使用的是無界隊列。

這樣潛在的問題就是當處理任務的速度趕不上任務提交的速度的時候，就可能會讓大量任務堆積在workQueue中，從而引發OOM異常。

4.2 演示newFixedThreadPool記憶體溢出問題

/**   * 演示newFixedThreadPool執行緒池OOM問題   */  public class FixedThreadPoolOOM {        private static ExecutorService executorService = Executors.newFixedThreadPool(1);        public static void main(String[] args) {          for (int i = 0; i < Integer.MAX_VALUE; i++) {              executorService.execute(new SubThread());          }      }  }    class SubThread implements Runnable {        @Override      public void run() {          try {              //延長任務時間              Thread.sleep(1000000000);          } catch (InterruptedException e) {              e.printStackTrace();          }      }  }

更改JVM參數

運行結果

4.3 newSingleThreadExecutor

public class SingleThreadExecutor {    public static void main(String[] args) {        ExecutorService executorService = Executors.newSingleThreadExecutor();        for (int i = 0; i < 1000; i++) {            executorService.execute(new Task());        }    }}

使用執行緒池列印執行緒名

查看newSingleThreadExecutor源碼

  public static ExecutorService newSingleThreadExecutor() {    return new FinalizableDelegatedExecutorService        (new ThreadPoolExecutor(1, 1,                                0L, TimeUnit.MILLISECONDS,                                new LinkedBlockingQueue<Runnable>()));}

從源碼可以看出newSingleThreadExecutor和newFixedThreadPool基本類似，不同的只是corePoolSize和maxPoolSize的值，所以newSingleThreadExecutor也存在記憶體溢出問題。

4.4 newCachedThreadPool

newCachedThreadPool也被稱為可快取執行緒池，它是一個無界執行緒池，具有自動回收多餘執行緒的功能。

  public class CachedThreadPool {    public static void main(String[] args) {        ExecutorService executorService = Executors.newCachedThreadPool();        for (int i = 0; i < 1000; i++) {            executorService.execute(new Task());        }    }}

  public static ExecutorService newCachedThreadPool() {    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,                                  60L, TimeUnit.SECONDS,                                  new SynchronousQueue<Runnable>());}

newCachedThreadPool的maxPoolSize設置的值為Integer.MAX_VALUE，所以可能會導致執行緒被無限創建，最終導致OOM異常。

4.5 newScheduledThreadPool

該執行緒池支援周期性任務的執行

public class ScheduledThreadPoolTest {    public static void main(String[] args) {        ScheduledExecutorService scheduledExecutorService =                Executors.newScheduledThreadPool(10);//        scheduledExecutorService.schedule(new Task(), 5, TimeUnit.SECONDS);        scheduledExecutorService.scheduleAtFixedRate(new Task(), 1, 3, TimeUnit.SECONDS);    }}  

public ScheduledThreadPoolExecutor(int corePoolSize) {    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,          new DelayedWorkQueue());}  

4.6 正確的創建執行緒池的方法

根據業務場景不同，自己設置執行緒池參數，例如記憶體有多大，自己取執行緒名字等。

4.7 執行緒池裡的執行緒數量設置多少比較合適？

CPU密集型（加密、計算hash等）：最佳執行緒數設置為CPU核心數的1——2倍。

耗時I/O型（讀寫資料庫、文件、網路讀寫等）：最佳執行緒數一般會大於CPU核心數很多倍，以JVM監控顯示繁忙情況為依據，保證執行緒空閑可以銜接上。參考Brain Goezt推薦的計算方法：執行緒數=CPU核心數 × (1+平均等待時間/平均工作時間)

5 對比執行緒池的特點

FixedThreadPool：通過手動傳入corePoolSize和maxPoolSize，以固定的執行緒數來執行任務

SingleThreadExecutor：corePoolSize和maxPoolSize默認都是1，全程只以1條執行緒執行任務

CachedThreadPool：它沒有需要維護的核心執行緒數，每當需要執行緒的時候就進行創建，因為它的執行緒存活時間是60秒，所以它也憑藉著這個參數實現了自動回收的功能。

ScheduledThreadPool：這個執行緒池可以執行定時任務，corePoolSize是通過手動傳入的，它的maxPoolSize為Integer.MAX_VALUE，並且具有自動回收執行緒的功能。

5.1 為什麼FixedThreadPool和SingleThreadExecutor的Queue是LinkedBlockingQueue？

因為這兩個執行緒池的核心執行緒數和最大執行緒數都是相同的，也就無法預估任務量，所以需要在自身進行改進，就使用了無界隊列。

5.2 為什麼CachedThreadPool使用的Queue是SynchronousQueue？

因為快取執行緒池的最大執行緒數是「無上限」的，每當任務來的時候直接創建執行緒進行執行就好了，所以不需要使用隊列來存儲任務。這樣避免了使用隊列進行任務的中轉，提高了執行效率。

5.3 為什麼ScheduledThreadPool使用延遲隊列DelayedWorkQueue？

因為ScheduledThreadPool是延遲任務執行緒池，所以使用延遲隊列有利於對執行任務的時間做延遲。

5.4 JDK1.8中加入的workStealingPool

• workStealingPool適用於執行產生子任務的環境，例如進行二叉樹的遍歷。
• workStealingPool具有竊取能力。
• 使用時最好不要加鎖，而且不保證執行順序。

6 停止執行緒池的正確方法

shutdown：調用了shutdown()方法不一定會立即停止，這個方法僅僅是初始整個關閉過程。因為執行緒池中的執行緒有可能正在運行，並且隊列中也有待處理的任務，不可能說停就停。所以每當調用該方法時，執行緒池會把正在執行的任務和隊列中等待的任務都執行完畢再關閉，並且在此期間如果接收到新的任務會被拒絕。

/**   * 演示關閉執行緒池   */  public class ShutDown {      public static void main(String[] args) throws InterruptedException {          ExecutorService executorService = Executors.newFixedThreadPool(10);          for (int i = 0; i < 1000; i++) {              executorService.execute(new ShutDownTask());          }          Thread.sleep(1500);          executorService.shutdown();            //再次提交任務          executorService.execute(new ShutDownTask());      }  }    class ShutDownTask implements Runnable {        @Override      public void run() {          try {              Thread.sleep(500);          } catch (InterruptedException e) {              e.printStackTrace();          }          System.out.println(Thread.currentThread().getName());      }  }

isShutdown：可以用於判斷執行緒池是否被shutdown了

/**   * 演示關閉執行緒池   */  public class ShutDown {      public static void main(String[] args) throws InterruptedException {          ExecutorService executorService = Executors.newFixedThreadPool(10);          for (int i = 0; i < 1000; i++) {              executorService.execute(new ShutDownTask());          }          Thread.sleep(1500);          System.out.println(executorService.isShutdown());          executorService.shutdown();          System.out.println(executorService.isShutdown());          //再次提交任務  //        executorService.execute(new ShutDownTask());      }  }    class ShutDownTask implements Runnable {        @Override      public void run() {          try {              Thread.sleep(500);          } catch (InterruptedException e) {              e.printStackTrace();          }          System.out.println(Thread.currentThread().getName());      }  }

isTerminated：可以判斷執行緒是否被完全終止了

/** * 演示關閉執行緒池 */public class ShutDown {    public static void main(String[] args) throws InterruptedException {        ExecutorService executorService = Executors.newFixedThreadPool(10);        for (int i = 0; i < 1000; i++) {            executorService.execute(new ShutDownTask());        }        Thread.sleep(1500);        System.out.println(executorService.isShutdown());        executorService.shutdown();        System.out.println(executorService.isShutdown());        System.out.println(executorService.isTerminated());        //再次提交任務//        executorService.execute(new ShutDownTask());    }}

將循環的次數改為100次，並且在第一次調用isTerminated方法的地方休眠10s

awaitTermination：傳入等待時間，等待時間達到時判斷是否停止了，主要用於檢測。

  //在3s後判斷執行緒池是否被終止，返回boolean值System.out.println(executorService.awaitTermination(3L, TimeUnit.SECONDS));

shutdownNow：調用了這個方法時，執行緒池會立即終止，並返回沒有被處理完的任務。如果需要繼續執行這裡的任務可以再次讓執行緒池執行這些返回的任務。

7 任務太多，怎麼拒絕？

7.1 拒絕的時機

當Executor關閉時，新提交的任務會被拒絕。

以及Executor對最大執行緒數和工作隊列容量使用有限邊界並且已經飽和時。

7.2 拒絕策略

AbortPolicy（中斷策略）：直接拋出異常進行拒絕

DiscardPolicy（丟棄策略）：不會得到通知，默默的拋棄掉任務

DiscardOldestPolicy（丟棄最老的）：由於隊列中存儲了很多任務，這個策略會丟棄在隊列中存在時間最久的任務。

CallerRunsPolicy：比如主執行緒給執行緒池提交任務，但是執行緒池已經滿了，在這種策略下會讓提交任務的執行緒去執行。

總結：第四種拒絕策略相對於前三種更加「機智」一些，可以避免前面三種策略產生的損失。在第四種策略下可以降低提交的速度，達到負回饋的效果。

8 使用鉤子為執行緒池加點料（可用於日誌記錄）

/**   * 演示每個任務執行的前後放鉤子函數   */  public class PauseableThreadPool extends ThreadPoolExecutor {        private boolean isPaused;      private final ReentrantLock lock = new ReentrantLock();      private Condition unPaused = lock.newCondition();        public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,                                 TimeUnit unit, BlockingQueue<Runnable> workQueue) {          super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);      }          public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,                                 TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {          super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);      }          public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,                                 TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {          super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);      }        public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,                                 TimeUnit unit, BlockingQueue<Runnable> workQueue,                                 ThreadFactory threadFactory, RejectedExecutionHandler handler) {          super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);      }        @Override      protected void beforeExecute(Thread t, Runnable r) {          super.beforeExecute(t, r);          lock.lock();          try {              while (isPaused) {                  unPaused.await();              }          } catch (InterruptedException e) {              e.printStackTrace();          } finally {              lock.unlock();          }      }        private void pause() {          lock.lock();          try {              isPaused = true;          } finally {              lock.unlock();          }      }        public void resume() {          lock.lock();          try {              isPaused = false;              //喚醒全部              unPaused.signalAll();          } finally {              lock.unlock();          }      }        public static void main(String[] args) throws InterruptedException {          PauseableThreadPool pauseableThreadPool = new PauseableThreadPool(10, 20, 10L,                  TimeUnit.SECONDS, new LinkedBlockingQueue<>());          Runnable runnable = new Runnable() {              @Override              public void run() {                  System.out.println(我被執行);                  try {                      Thread.sleep(10);                  } catch (InterruptedException e) {                      e.printStackTrace();                  }              }          };          for (int i = 0; i < 10000; i++) {              pauseableThreadPool.execute(runnable);          }          Thread.sleep(1500);          pauseableThreadPool.pause();          System.out.println(執行緒池被暫停了);          Thread.sleep(1500);          pauseableThreadPool.resume();          System.out.println(執行緒池被恢復了);      }  }

9 執行緒池實現原理

9.1 執行緒池組成部分

執行緒池管理器

工作執行緒

任務隊列

任務

9.2 Executor家族

Executor：它是一個頂層介面，其他介面以及類都i繼承或實現於它，包含以下方法：

void execute(Runnable command);

ExecutorService：它繼承於Executor，是Executor的子介面，在介面內部增加了一些新的方法，例如第6小節講到的幾個方法

Executors：這個類是一個工具類，裡面包含一些創建執行緒池的方法

9.3 執行緒池實現任務復用的原理

利用相同執行緒執行不同任務

源碼分析

public void execute(Runnable command) {      // 判斷任務是否為空，為空就拋出異常      if (command == null)          throw new NullPointerException();        int c = ctl.get();      // 如果當前執行緒數小於核心執行緒數，就增加Worker      if (workerCountOf(c) < corePoolSize) {          // command就是任務，點擊addWorker方法          // 第二個參數用於判斷當前執行緒數是否小於核心執行緒數          if (addWorker(command, true))              return;          c = ctl.get();      }      // 此時執行緒數大於等於核心執行緒數      // 判斷執行緒池是不是正在運行並將任務放到工作隊列中      if (isRunning(c) && workQueue.offer(command)) {          // 再次檢查執行緒狀態          int recheck = ctl.get();          // 如果執行緒不是正在運行的，就刪除掉任務並且拒絕          if (! isRunning(recheck) && remove(command))              reject(command);          else if (workerCountOf(recheck) == 0)   //這裡用於避免已經提交的任務沒有執行緒進行執行              addWorker(null, false);      }      // 如果任務無法添加或者大於最大執行緒數就拒絕任務      else if (!addWorker(command, false))          reject(command);  }

因為要查看的是Worker所以進入到addWorker()方法後點擊Worker類查看runWorker()方法

w = new Worker(firstTask);      private final class Worker      extends AbstractQueuedSynchronizer      implements Runnable      final void runWorker(Worker w) {      Thread wt = Thread.currentThread();      // 獲取到任務      Runnable task = w.firstTask;      w.firstTask = null;      w.unlock(); // allow interrupts      boolean completedAbruptly = true;      try {          // 只要任務不為空或者能夠獲取到任務就執行下面的方法          while (task != null || (task = getTask()) != null) {              w.lock();              if ((runStateAtLeast(ctl.get(), STOP) ||                   (Thread.interrupted() &&                    runStateAtLeast(ctl.get(), STOP))) &&                  !wt.isInterrupted())                  wt.interrupt();              try {                  beforeExecute(wt, task);                  Throwable thrown = null;                  try {                      // task是一個Runnable類型，調用run()方法就是運行執行緒                      task.run();                  } catch (RuntimeException x) {                      thrown = x; throw x;                  } catch (Error x) {                      thrown = x; throw x;                  } catch (Throwable x) {                      thrown = x; throw new Error(x);                  } finally {                      afterExecute(task, thrown);                  }              } finally {                  task = null;                  w.completedTasks++;                  w.unlock();              }          }          completedAbruptly = false;      } finally {          processWorkerExit(w, completedAbruptly);      }  }

總結：核心原理就是獲取到task,如果task不為空就調用run()方法，這樣就實現了執行緒的復用，達到讓相同的執行緒執行不同任務的目的。

10 執行緒池狀態

RUNNING：接受新任務並處理排隊任務

SHUTDOWN：不接受新的任務但是處理排隊任務

STOP：不接受新的任務，也不處理排隊的任務，並中斷正在執行的任務，就是調用shutdownNow()帶來的效果

TIDYING：中文意思是整潔，意思就是說任務都已經終止，workerCount為零時，執行緒會轉換到TIDYING狀態，並將運行terminate()鉤子方法

TERMINATED：terminate()運行完成

11 使用執行緒池的注意點

避免任務的堆積（堆積容易產生記憶體溢出）

避免執行緒數過多增加（快取執行緒池會導致執行緒數過度增加）

排查執行緒泄漏（執行緒已經執行完畢卻無法被回收）

參考來源：https://juejin.im/post/5e1b1fcce51d454d3046a3de