Java執行緒池Executor用法詳解

2022-08-04 22:06:16

執行緒池類圖

我們最常使用的Executors實現建立執行緒池使用執行緒主要是用上述類圖中提供的類。在上邊的類圖中，包含了一個Executor框架，它是一個根據一組執行策略的呼叫排程執行和控制非同步任務的框架，目的是提供一種將任務提交與任務如何執行分離開的機制。它包含了三個executor介面：

Executor:執行新任務的簡單介面
ExecutorService：擴充套件了Executor，新增了用來管理執行器生命週期和任務生命週期的方法
ScheduleExcutorService：擴充套件了ExecutorService，支援Future和定期執行任務

執行緒池的好處

降低資源消耗-重用存在的執行緒，減少物件建立、消亡的開銷，效能好
提高響應速度 -可有效控制最大並行執行緒數，提高系統資源利用率，同時可以避免過多資源競爭，避免阻塞。當任務到達時，任務可不用等待執行緒建立就能立即執行
提高執行緒的可管理性-提供定時執行、定期執行、單執行緒、並行數控制等功能。

new Thread的弊端

每次new Thread 新建物件，效能差
執行緒缺乏統一管理，可能無限制的新建執行緒，相互競爭，可能佔用過多的系統資源導致宕機或者OOM（out of memory 記憶體溢位），這種問題的原因不是因為單純的new一個Thread，而是可能因為程式的bug或者設計上的缺陷導致不斷new Thread造成的。
缺少更多功能，如更多執行、定期執行、執行緒中斷。

執行緒池核心類-ThreadPoolExecutor

引數說明：ThreadPoolExecutor一共有七個引數，這七個引數配合起來，構成了執行緒池強大的功能。

corePoolSize：核心執行緒數量

maximumPoolSize：執行緒最大執行緒數

workQueue：阻塞佇列，儲存等待執行的任務，很重要，會對執行緒池執行過程產生重大影響

當我們提交一個新的任務到執行緒池，執行緒池會根據當前池中正在執行的執行緒數量來決定該任務的處理方式。處理方式有三種：

1、直接切換（SynchronusQueue）

2、無界佇列（LinkedBlockingQueue）能夠建立的最大執行緒數為corePoolSize,這時maximumPoolSize就不會起作用了。當執行緒池中所有的核心執行緒都是執行狀態的時候，新的任務提交就會放入等待佇列中。

3、有界佇列（ArrayBlockingQueue）最大maximumPoolSize，能夠降低資源消耗，但是這種方式使得執行緒池對執行緒排程變的更困難。因為執行緒池與佇列容量都是有限的。所以想讓執行緒池的吞吐率和處理任務達到一個合理的範圍，又想使我們的執行緒排程相對簡單，並且還儘可能降低資源的消耗，我們就需要合理的限制這兩個數量分配技巧： [如果想降低資源的消耗包括降低cpu使用率、作業系統資源的消耗、上下文切換的開銷等等，可以設定一個較大的佇列容量和較小的執行緒池容量，這樣會降低執行緒池的吞吐量。如果我們提交的任務經常發生阻塞，我們可以調整maximumPoolSize。如果我們的佇列容量較小，我們需要把執行緒池大小設定的大一些，這樣cpu的使用率相對來說會高一些。但是如果執行緒池的容量設定的過大，提高任務的數量過多的時候，並行量會增加，那麼執行緒之間的排程就是一個需要考慮的問題。這樣反而可能會降低處理任務的吞吐量。]

keepAliveTime：執行緒沒有任務執行時最多保持多久時間終止（當執行緒中的執行緒數量大於corePoolSize的時候，如果這時沒有新的任務提交核心執行緒外的執行緒不會立即銷燬，而是等待，直到超過keepAliveTime）

unit：keepAliveTime的時間單位

threadFactory：執行緒工廠，用來建立執行緒，有一個預設的工場來建立執行緒，這樣新建立出來的執行緒有相同的優先順序，是非守護執行緒、設定好了名稱）

rejectHandler：當拒絕處理任務時(阻塞佇列滿)的策略（AbortPolicy預設策略直接丟擲異常、CallerRunsPolicy用呼叫者所在的執行緒執行任務、DiscardOldestPolicy丟棄佇列中最靠前的任務並執行當前任務、DiscardPolicy直接丟棄當前任務）

corePoolSize、maximumPoolSize、workQueue 三者關係：如果執行的執行緒數小於corePoolSize的時候，直接建立新執行緒來處理任務。即使執行緒池中的其他執行緒是空閒的。如果執行中的執行緒數大於corePoolSize且小於maximumPoolSize時，那麼只有當workQueue滿的時候才建立新的執行緒去處理任務。如果corePoolSize與maximumPoolSize是相同的，那麼建立的執行緒池大小是固定的。這時有新任務提交，當workQueue未滿時，就把請求放入workQueue中。等待空執行緒從workQueue取出任務。如果workQueue此時也滿了，那麼就使用另外的拒絕策略引數去執行拒絕策略。

初始化方法：由七個引數組合成四個初始化方法

其他方法：

execute();	//提交任務，交給執行緒池執行	
submit();//提交任務，能夠返回執行結果 execute+Future
shutdown();//關閉執行緒池，等待任務都執行完
shutdownNow();//關閉執行緒池，不等待任務執行完
getTaskCount();//執行緒池已執行和未執行的任務總數
getCompleteTaskCount();//已完成的任務數量
getPoolSize();//執行緒池當前的執行緒數量
getActiveCount();//當前執行緒池中正在執行任務的執行緒數量

執行緒池生命週期：

running：能接受新提交的任務，也能處理阻塞佇列中的任務
shutdown：不能處理新的任務，但是能繼續處理阻塞佇列中任務
stop：不能接收新的任務，也不處理佇列中的任務
tidying：如果所有的任務都已經終止了，這時有效執行緒數為0
terminated：最終狀態

使用Executors建立執行緒池

使用Executors可以建立四種執行緒池：分別對應上邊提到的四種執行緒池初始化方法

Executors.newCachedThreadPool

newCachedThreadPool是一個根據需要建立新執行緒的執行緒池，當一個任務提交時，corePoolSize為0不建立核心執行緒，SynchronousQueue是一個不儲存元素的佇列，可以理解為隊裡永遠是滿的，因此最終會建立非核心執行緒來執行任務。對於非核心執行緒空閒60s時將被回收。因為Integer.MAX_VALUE非常大，可以認為是可以無限建立執行緒的，在資源有限的情況下容易引起OOM異常。

//建立newCachedThreadPool執行緒池原始碼
public static ExecutorService newCachedThreadPool() {
		/**
        *corePoolSize： 0，核心執行緒池的數量為0
		*maximumPoolSize：  Integer.MAX_VALUE，可以認為最大執行緒數是無限的
		*keepAliveTime： 60L
		*unit: 秒
		*workQueue: SynchronousQueue
        **/
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

使用案例：

public static void main(String[] args) {
    ExecutorService executor = Executors.newCachedThreadPool();
    for (int i = 0; i < 10; i++) {
        final int index = i;
        executor.execute(new Runnable() {
            @Override
            public void run() {
                log.info("task:{}",index);
            }
        });
    }
}

值得注意的一點是，newCachedThreadPool的返回值是ExecutorService型別，該型別只包含基礎的執行緒池方法，但卻不包含執行緒監控相關方法，因此在使用返回值為ExecutorService的執行緒池型別建立新執行緒時要考慮到具體情況。

Executors.newSingleThreadExecutor

newSingleThreadExecutor是單執行緒執行緒池，只有一個核心執行緒，用唯一的一個共用執行緒執行任務，保證所有任務按指定順序執行（FIFO、優先順序…）

//newSingleThreadExecutor建立執行緒池原始碼
public static ExecutorService newSingleThreadExecutor() {
    /**
      *  corePoolSize : 1，核心執行緒池的數量為1

      *  maximumPoolSize : 1，只可以建立一個非核心執行緒

      *  keepAliveTime : 0L

      *  unit => 秒

      *  workQueue => LinkedBlockingQueue
      **/
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

當一個任務提交時，首先會建立一個核心執行緒來執行任務，如果超過核心執行緒的數量，將會放入佇列中，因為LinkedBlockingQueue是長度為Integer.MAX_VALUE的佇列，可以認為是無界佇列，因此往佇列中可以插入無限多的任務，在資源有限的時候容易引起OOM異常，同時因為無界佇列，maximumPoolSize和keepAliveTime引數將無效，壓根就不會建立非核心執行緒。

Executors.newFixedThreadPool

定長執行緒池，核心執行緒數和最大執行緒數由使用者傳入，可以設定執行緒的最大並行數，超出在佇列等待

//newFixedThreadPool建立執行緒池原始碼
public static ExecutorService newFixedThreadPool(int nThreads) {
    	/**
          *  corePoolSize : 核心執行緒的數量為自定義輸入nThreads

          *  maximumPoolSize ： 最大執行緒的數量為自定義輸入nThreads

          *  keepAliveTime ： 0L

          *  unit ： 秒

          *  workQueue ： LinkedBlockingQueue
          **/
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

newFixedThreadPool和SingleThreadExecutor類似，唯一的區別就是核心執行緒數不同，並且由於使用的是LinkedBlockingQueue，在資源有限的時候容易引起OOM異常。

Executors.newScheduledThreadPool

定長執行緒池，核心執行緒數由使用者傳入，支援定時和週期任務執行

//newScheduledThreadPool建立執行緒池原始碼
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    /**
      *  corePoolSize : 核心執行緒的數量為自定義輸入corePoolSize

      *  maximumPoolSize ： 最大執行緒的數量為Integer.MAX_VALUE

      *  keepAliveTime ： 0L

      *  unit ： 納秒

      *  workQueue ： DelayedWorkQueue
      **/
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

當一個任務提交時，corePoolSize為自定義輸入，首先建立核心執行緒，核心執行緒滿了之後，因此最終會建立非核心執行緒來執行任務。非核心執行緒使用後將被回收。因為Integer.MAX_VALUE非常大，可以認為是可以無限建立執行緒的，在資源有限的情況下容易引起OOM異常。因為使用的DelayedWorkQueue可以實現定時和週期任務。 ScheduledExecutorService提供了三種方法可以使用：

schedule：延遲後執行任務 scheduleAtFixedRate：以指定的速率執行任務 scheduleWithFixedDelay：以指定的延遲執行任務使用案例：

    public static void main(String[] args) {

        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);

//        executorService.schedule(new Runnable() {
//            @Override
//            public void run() {
//                log.warn("schedule run");
//            }
//         //延遲3秒後執行
//        }, 3, TimeUnit.SECONDS);
        //        executorService.shutdown();

//        executorService.scheduleWithFixedDelay(new Runnable() {
//            @Override
//            public void run() {
//                log.warn("scheduleWithFixedDelay run");
//            }
//            //延遲一秒後每隔3秒執行
//        }, 1, 3, TimeUnit.SECONDS);
        
        executorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                log.warn("schedule run");
            }
            //延遲一秒後每隔3秒執行
        }, 1, 3, TimeUnit.SECONDS);

        /**
         * 定時器排程，不推薦使用，推薦ScheduledExecutorService排程
         */
//        Timer timer = new Timer();
//        timer.schedule(new TimerTask() {
//            @Override
//            public void run() {
//                log.warn("timer run");
//            }
//        //從當前時間每隔5秒執行
//        }, new Date(), 5 * 1000);
    }

總結

FixedThreadPool和SingleThreadExecutor 允許的請求佇列長度為Integer.MAX_VALUE，可能會堆積大量的請求，從而引起OOM異常
CachedThreadPool 和newScheduledThreadPool允許建立的執行緒數為Integer.MAX_VALUE，可能會建立大量的執行緒，從而引起OOM異常

這就是為什麼禁止使用Executors去建立執行緒池，而是推薦自己去建立ThreadPoolExecutor的原因

如何定義執行緒池引數

CPU密集型： 執行緒池的大小推薦為CPU數量 + 1，CPU數量可以根據Runtime.availableProcessors方法獲取 IO密集型： CPU數量 * CPU利用率 * (1 + 執行緒等待時間/執行緒CPU時間) 混合型： 將任務分為CPU密集型和IO密集型，然後分別使用不同的執行緒池去處理，從而使每個執行緒池可以根據各自的工作負載來調整 阻塞佇列： 推薦使用有界佇列，有界佇列有助於避免資源耗盡的情況發生 拒絕策略： 預設採用的是AbortPolicy拒絕策略，直接在程式中丟擲RejectedExecutionException異常【因為是執行時異常，不強制catch】，這種處理方式不夠優雅。處理拒絕策略有以下幾種比較推薦：

在程式中捕獲RejectedExecutionException異常，在捕獲異常中對任務進行處理。針對預設拒絕策略
使用CallerRunsPolicy拒絕策略，該策略會將任務交給呼叫execute的執行緒執行【一般為主執行緒】，此時主執行緒將在一段時間內不能提交任何任務，從而使工作執行緒處理正在執行的任務。此時提交的執行緒將被儲存在TCP佇列中，TCP佇列滿將會影響使用者端，這是一種平緩的效能降低
自定義拒絕策略，只需要實現RejectedExecutionHandler介面即可
如果任務不是特別重要，使用DiscardPolicy和DiscardOldestPolicy拒絕策略將任務丟棄也是可以的

如果使用Executors的靜態方法建立ThreadPoolExecutor物件，可以通過使用Semaphore對任務的執行進行限流也可以避免出現OOM異常

到此這篇關於Java執行緒池Executor用法詳解的文章就介紹到這了,更多相關Java執行緒池Executor內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！