Java線程池Executor怎么使用

本文小編為大家詳細介紹“Java線程池Executor怎么使用”，內容詳細，步驟清晰，細節(jié)處理妥當，希望這篇“Java線程池Executor怎么使用”文章能幫助大家解決疑惑，下面跟著小編的思路慢慢深入，一起來學習新知識吧。

線程池類圖

我們最常使用的Executors實現(xiàn)創(chuàng)建線程池使用線程主要是用上述類圖中提供的類。在上邊的類圖中，包含了一個Executor框架，它是一個根據一組執(zhí)行策略的調用調度執(zhí)行和控制異步任務的框架，目的是提供一種將任務提交與任務如何運行分離開的機制。它包含了三個executor接口：

• Executor:運行新任務的簡單接口

• ExecutorService：擴展了Executor，添加了用來管理執(zhí)行器生命周期和任務生命周期的方法

• ScheduleExcutorService：擴展了ExecutorService，支持Future和定期執(zhí)行任務

線程池的好處

• 降低資源消耗-重用存在的線程，減少對象創(chuàng)建、消亡的開銷，性能好

• 提高響應速度 -可有效控制最大并發(fā)線程數，提高系統(tǒng)資源利用率，同時可以避免過多資源競爭，避免阻塞。當任務到達時，任務可不用等待線程創(chuàng)建就能立即執(zhí)行

• 提高線程的可管理性-提供定時執(zhí)行、定期執(zhí)行、單線程、并發(fā)數控制等功能。

new Thread的弊端

• 每次new Thread 新建對象，性能差

• 線程缺乏統(tǒng)一管理，可能無限制的新建線程，相互競爭，可能占用過多的系統(tǒng)資源導致死機或者OOM（out of memory 內存溢出），這種問題的原因不是因為單純的new一個Thread，而是可能因為程序的bug或者設計上的缺陷導致不斷new Thread造成的。

• 缺少更多功能，如更多執(zhí)行、定期執(zhí)行、線程中斷。

線程池核心類-ThreadPoolExecutor

參數說明：ThreadPoolExecutor一共有七個參數，這七個參數配合起來，構成了線程池強大的功能。

corePoolSize：核心線程數量

maximumPoolSize：線程最大線程數

workQueue：阻塞隊列，存儲等待執(zhí)行的任務，很重要，會對線程池運行過程產生重大影響

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

1、直接切換（SynchronusQueue）

2、無界隊列（LinkedBlockingQueue）能夠創(chuàng)建的最大線程數為corePoolSize,這時maximumPoolSize就不會起作用了。當線程池中所有的核心線程都是運行狀態(tài)的時候，新的任務提交就會放入等待隊列中。

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

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

unit：keepAliveTime的時間單位

threadFactory：線程工廠，用來創(chuàng)建線程，有一個默認的工場來創(chuàng)建線程，這樣新創(chuàng)建出來的線程有相同的優(yōu)先級，是非守護線程、設置好了名稱）

rejectHandler：當拒絕處理任務時(阻塞隊列滿)的策略（AbortPolicy默認策略直接拋出異常、CallerRunsPolicy用調用者所在的線程執(zhí)行任務、DiscardOldestPolicy丟棄隊列中最靠前的任務并執(zhí)行當前任務、DiscardPolicy直接丟棄當前任務）

corePoolSize、maximumPoolSize、workQueue 三者關系：如果運行的線程數小于corePoolSize的時候，直接創(chuàng)建新線程來處理任務。即使線程池中的其他線程是空閑的。如果運行中的線程數大于corePoolSize且小于maximumPoolSize時，那么只有當workQueue滿的時候才創(chuàng)建新的線程去處理任務。如果corePoolSize與maximumPoolSize是相同的，那么創(chuàng)建的線程池大小是固定的。這時有新任務提交，當workQueue未滿時，就把請求放入workQueue中。等待空線程從workQueue取出任務。如果workQueue此時也滿了，那么就使用另外的拒絕策略參數去執(zhí)行拒絕策略。

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

其他方法：

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

線程池生命周期：

• running：能接受新提交的任務，也能處理阻塞隊列中的任務

• shutdown：不能處理新的任務，但是能繼續(xù)處理阻塞隊列中任務

• stop：不能接收新的任務，也不處理隊列中的任務

• tidying：如果所有的任務都已經終止了，這時有效線程數為0

• terminated：最終狀態(tài)

使用Executors創(chuàng)建線程池

使用Executors可以創(chuàng)建四種線程池：分別對應上邊提到的四種線程池初始化方法

Executors.newCachedThreadPool

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

//創(chuàng)建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類型，該類型只包含基礎的線程池方法，但卻不包含線程監(jiān)控相關方法，因此在使用返回值為ExecutorService的線程池類型創(chuàng)建新線程時要考慮到具體情況。

Executors.newSingleThreadExecutor

newSingleThreadExecutor是單線程線程池，只有一個核心線程，用唯一的一個共用線程執(zhí)行任務，保證所有任務按指定順序執(zhí)行（FIFO、優(yōu)先級…）

//newSingleThreadExecutor創(chuàng)建線程池源碼
public static ExecutorService newSingleThreadExecutor() {
    /**
      *  corePoolSize : 1，核心線程池的數量為1

      *  maximumPoolSize : 1，只可以創(chuàng)建一個非核心線程

      *  keepAliveTime : 0L

      *  unit => 秒

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

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

Executors.newFixedThreadPool

定長線程池，核心線程數和最大線程數由用戶傳入，可以設置線程的最大并發(fā)數，超出在隊列等待

//newFixedThreadPool創(chuàng)建線程池源碼
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類似，唯一的區(qū)別就是核心線程數不同，并且由于使用的是LinkedBlockingQueue，在資源有限的時候容易引起OOM異常。

Executors.newScheduledThreadPool

定長線程池，核心線程數由用戶傳入，支持定時和周期任務執(zhí)行

//newScheduledThreadPool創(chuàng)建線程池源碼
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為自定義輸入，首先創(chuàng)建核心線程，核心線程滿了之后，因此最終會創(chuàng)建非核心線程來執(zhí)行任務。非核心線程使用后將被回收。因為Integer.MAX_VALUE非常大，可以認為是可以無限創(chuàng)建線程的，在資源有限的情況下容易引起OOM異常。因為使用的DelayedWorkQueue可以實現(xiàn)定時和周期任務。 ScheduledExecutorService提供了三種方法可以使用：

schedule：延遲后執(zhí)行任務 scheduleAtFixedRate：以指定的速率執(zhí)行任務 scheduleWithFixedDelay：以指定的延遲執(zhí)行任務 使用案例：

    public static void main(String[] args) {

        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);

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

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

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

總結

• FixedThreadPool和SingleThreadExecutor 允許的請求隊列長度為Integer.MAX_VALUE，可能會堆積大量的請求，從而引起OOM異常

• CachedThreadPool 和newScheduledThreadPool允許創(chuàng)建的線程數為Integer.MAX_VALUE，可能會創(chuàng)建大量的線程，從而引起OOM異常

這就是為什么禁止使用Executors去創(chuàng)建線程池，而是推薦自己去創(chuàng)建ThreadPoolExecutor的原因

如何定義線程池參數

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

• 在程序中捕獲RejectedExecutionException異常，在捕獲異常中對任務進行處理。針對默認拒絕策略

• 使用CallerRunsPolicy拒絕策略，該策略會將任務交給調用execute的線程執(zhí)行【一般為主線程】，此時主線程將在一段時間內不能提交任何任務，從而使工作線程處理正在執(zhí)行的任務。此時提交的線程將被保存在TCP隊列中，TCP隊列滿將會影響客戶端，這是一種平緩的性能降低

• 自定義拒絕策略，只需要實現(xiàn)RejectedExecutionHandler接口即可

• 如果任務不是特別重要，使用DiscardPolicy和DiscardOldestPolicy拒絕策略將任務丟棄也是可以的

如果使用Executors的靜態(tài)方法創(chuàng)建ThreadPoolExecutor對象，可以通過使用Semaphore對任務的執(zhí)行進行限流也可以避免出現(xiàn)OOM異常

讀到這里，這篇“Java線程池Executor怎么使用”文章已經介紹完畢，想要掌握這篇文章的知識點還需要大家自己動手實踐使用過才能領會，如果想了解更多相關內容的文章，歡迎關注億速云行業(yè)資訊頻道。