快捷導(dǎo)航

簡(jiǎn)單聊一聊Java線程池ThreadPoolExecutor

更新時(shí)間：2022年06月11日 09:56:21 作者：生命猿于運(yùn)動(dòng)

在使用線程池之后,開啟線程就變成了在線程池當(dāng)中找到一個(gè)空閑的線程,銷毀線程變成了歸還線程到線程池的過程,下面這篇文章主要給大家介紹了關(guān)于Java線程池ThreadPoolExecutor的相關(guān)資料,需要的朋友可以參考下

簡(jiǎn)介

ThreadPoolExecutor是一個(gè)實(shí)現(xiàn)ExecutorService接口的線程池，ExecutorService是主要用來處理多線程任務(wù)的一個(gè)接口，通常比較簡(jiǎn)單是用法是由Executors工廠類去創(chuàng)建。

線程池主要解決了兩個(gè)不同的問題：

在執(zhí)行大量異步任務(wù)時(shí)，為了能夠提高性能，通常會(huì)減少每個(gè)任務(wù)的調(diào)用開銷。
提供了一系列多線程任務(wù)的管理方法，便于多任務(wù)執(zhí)行時(shí)合理分配資源以及一些異常情況的處理。每個(gè)ThreadPoolExecutor還維護(hù)一些基本統(tǒng)計(jì)信息。例如：已完成任務(wù)的數(shù)量，當(dāng)前獲得線程數(shù)等。

參數(shù)說明

ThreadPoolExecutor提供了幾個(gè)核心參數(shù)，方便開發(fā)人員根據(jù)具體場(chǎng)景合理分配線程資源。

corePoolSize：核心線程數(shù)，在線程池創(chuàng)建時(shí)就已初始化好的n個(gè)核心線程，即使線程空閑著也會(huì)一直保留在線程池中不被銷毀，除非調(diào)用線程池方法設(shè)置了java.util.concurrent.ThreadPoolExecutor#allowCoreThreadTimeOut(true)（允許核心線程超時(shí)銷毀）。
maximumPoolSize：線程池允許存在最大線程數(shù)。
keepAliveTime：當(dāng)線程數(shù)大于核心線程數(shù)時(shí)，多余的線程在執(zhí)行任務(wù)結(jié)束后等待新任務(wù)的最大等待時(shí)間。
unit：TimeUnit類型，是keepAliveTime多余線程最大空余時(shí)間單位。
workQueue：必須指定一個(gè)阻塞隊(duì)列，在線程池執(zhí)行execute方法時(shí)新進(jìn)來的任務(wù)在執(zhí)行前都會(huì)保留到此隊(duì)列里進(jìn)入等待。
threadFactory：創(chuàng)建線程的工廠，默認(rèn)采用Executors.defaultThreadFactory()創(chuàng)建線程。
handler：拒絕策略，當(dāng)最大線程數(shù)已占滿，且隊(duì)列已滿，此時(shí)線程池將觸發(fā)拒絕策略，對(duì)新進(jìn)來的任務(wù)做拒絕處理，具體的處理方案在后面詳細(xì)分析（默認(rèn)使用java.util.concurrent.ThreadPoolExecutor.AbortPolicy直接拋出異常拒絕處理）。

注：maximumPoolSize如果大于corePoolSize，則多出的部分線程數(shù)只有在阻塞隊(duì)列workQueue占滿時(shí)才會(huì)創(chuàng)建核心線程之外的線程去執(zhí)行任務(wù)，如果我們?cè)O(shè)置的阻塞隊(duì)列為無界隊(duì)列（默認(rèn)大小為Integer.MAX_VALUE），則隊(duì)列永遠(yuǎn)無法占滿，就不會(huì)去創(chuàng)建額外的線程進(jìn)行工作，一般情況如果任務(wù)數(shù)足夠，那么也是在隊(duì)列大小還沒達(dá)到Integer.MAX_VALUE時(shí)就已經(jīng)出現(xiàn)內(nèi)存溢出了。Executors線程池工廠中的newFixedThreadPool()、newSingleThreadExecutor()方法就是使用了無界隊(duì)列LinkedBlockingQueue，防止內(nèi)存溢出在日常開發(fā)過程中一般是不建議直接去使用Executors去創(chuàng)建線程池。

如何創(chuàng)建線程池

上面我們提到的可以使用Executors工廠直接創(chuàng)建線程池，但是Executors提供的創(chuàng)建線程池都是不可控的，我們還是得按自己的業(yè)務(wù)做好分析自定義一個(gè)線程池。

以下是線程池創(chuàng)建的一個(gè)案例：

@Slf4j
@Configuration
public class ThreadPoolConfig {

    @Value("${threadPool.corePoolSize:8}")
    private int corePoolSize;

    @Value("${threadPool.maximumPoolSize:16}")
    private int maximumPoolSize;

    @Value("${threadPool.keepAliveTime:60}")
    private int keepAliveTime;

    @Value("${threadPool.queueSize:99999}")
    private int queueSize;

    @Bean
    public ThreadPoolExecutor testExecutor() {
        LinkedBlockingQueue queue = new LinkedBlockingQueue(queueSize);
        return new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, 
                TimeUnit.SECONDS, queue, getThreadFactory(), getRejectedExecutionHandler());
    }

    /**
     * 自定義線程池創(chuàng)建線程工廠，用于線程池創(chuàng)建線程的工廠
     * @return
     */
    private ThreadFactory getThreadFactory() {
        return new ThreadFactory() {

            @Override
            public Thread newThread(Runnable r) {
                log.info("===> Create new thread ...");
                return new Thread(r);
            }
        };
    }

    /**
     * 自定義拒絕策略，繼續(xù)往隊(duì)列里添加任務(wù)進(jìn)入等待
     * @return
     */
    private RejectedExecutionHandler getRejectedExecutionHandler() {
        return new RejectedExecutionHandler() {

            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                // 繼續(xù)往隊(duì)列里添加任務(wù)，這里只是一個(gè)案例，這種方式并不友好，會(huì)拋出隊(duì)列已滿的異常
                log.info("===> Handler runnable ......");
                executor.getQueue().add(r);
            }
        };
    }
}

application.properties配置文件

threadPool:
  corePoolSize: 8
  maximumPoolSize: 16
  keepAliveTime: 60
  # 為方便測(cè)試這里我們配置隊(duì)列數(shù)小一點(diǎn)
  queueSize: 99

由以上的線程池配置，我們寫一個(gè)demo測(cè)試一下：

截取部分運(yùn)行日志：

紅框我們可以看到執(zhí)行到了線程創(chuàng)建工廠部分代碼塊
藍(lán)色框日志我們可以看到largestPoolSize=11，這是由于我們配置的maximumPoolSize=16 > corePoolSize=8，我們demo執(zhí)行的是110個(gè)任務(wù)并發(fā)，隊(duì)列大小是99，由此分析得出（需要額外出創(chuàng)建線程數(shù) = 并發(fā)任務(wù)總數(shù)110 - 核心線程數(shù)8 - 隊(duì)列大小99 = 3），所以線程池在隊(duì)列已滿時(shí)會(huì)多創(chuàng)建3個(gè)線程用于執(zhí)行任務(wù)，在達(dá)到keepAliveTime配置的最大空閑時(shí)間后這3個(gè)線程即會(huì)自動(dòng)銷毀。

注：可能有的同學(xué)會(huì)想線程池使用后需要銷毀嗎？在這里補(bǔ)充一下，如果我們是作為局部變量創(chuàng)建出來的線程池（如：在執(zhí)行的方法內(nèi)使用Executors.newFixedThreadPool(10)創(chuàng)建臨時(shí)的線程池），這種情況我們用完就必須將它立即銷毀，否則主線程就會(huì)一直處于運(yùn)行狀態(tài)。如果是全局配置的線程池，那么就是為整個(gè)系統(tǒng)中諸多業(yè)務(wù)提供使用的，這種就不需要對(duì)線程池做銷毀，因?yàn)橐坏╀N毀了其他的任務(wù)就無法繼續(xù)使用該線程池執(zhí)行任務(wù)。

銷毀線程池主要有兩種方式：
- shutdown()：此方法對(duì)線程池做銷毀，線程池會(huì)優(yōu)先將剩余未完成的任務(wù)執(zhí)行完才會(huì)執(zhí)行銷毀任務(wù)。
- shutdownNow()：此方法會(huì)對(duì)線程池做立即銷毀，無論線程池中的任務(wù)是否執(zhí)行完成。

拒絕策略

通常我們?cè)谂渲煤糜邢揸?duì)列大小后，就會(huì)有可能出現(xiàn)隊(duì)列占滿的情況，這時(shí)候我們的拒絕策略就會(huì)起到作用，接下來我們就來分析一下RejectedExecutionHandler接口具體有哪一些實(shí)現(xiàn)方式：

AbortPolicy：線程池的默認(rèn)拒絕策略，在JDK提供的ThreadPoolExecutor線程池中有一個(gè)默認(rèn)線程池變量private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();作為默認(rèn)拒絕策略，查看如下圖源碼可知它就是直接拋出RejectedExecutionException異常，并且會(huì)直接丟棄當(dāng)前任務(wù)，如果擔(dān)心異常影響后續(xù)任務(wù)執(zhí)行開發(fā)人員需自行捕獲異常處理。

CallerRunsPolicy：只要在當(dāng)線程池未被銷毀的情況下，不丟棄任務(wù)直接使用主線程（調(diào)用線程池執(zhí)行的線程）執(zhí)行該任務(wù)。因?yàn)樵摬呗允怯芍骶€程直接執(zhí)行任務(wù)的，所以不建議在并發(fā)度高的情況下使用，建議在并發(fā)度較低且任務(wù)不允許失敗的情況下才使用此策略。

DiscardPolicy：直接丟棄當(dāng)前任務(wù)，不做任何處理。直接丟棄任務(wù)的情況下，開發(fā)人員也無法排查到哪些任務(wù)被丟棄掉，一般不建議使用，除非是無關(guān)緊要的任務(wù)即使丟棄也無所謂的。

DiscardOldestPolicy：在線程池未被銷毀的情況下，丟棄最早進(jìn)入隊(duì)列的一個(gè)任務(wù)（即最久未執(zhí)行的任務(wù)），然后再重新將此任務(wù)加入線程池，在此策略下需注意被丟棄的任務(wù)的重要性，如果任務(wù)不重要可直接丟棄。

自定義策略：在以上JDK提供的四種默認(rèn)拒絕策略之外，我們還可以通過自定義的方式來處理被拒絕的任務(wù)。如果擔(dān)心任務(wù)被拒絕或者被丟棄造成不可預(yù)估的問題，在時(shí)效性沒有太大要求的情況下我們可以先將任務(wù)內(nèi)容轉(zhuǎn)換成數(shù)據(jù)入庫(kù)做好日志記錄，后續(xù)可以使用定時(shí)任務(wù)或者通過MQ消息延遲處理。由以上的線程池配置Demo中的拒絕策略改造偽代碼如下：

private RejectedExecutionHandler getRejectedExecutionHandler() {
    return new RejectedExecutionHandler() {

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            // 偽代碼
            log.info("===> 可根據(jù)任務(wù)的重要性區(qū)分對(duì)待，將任務(wù)做轉(zhuǎn)換入庫(kù)延遲處理 ......");
        }
    };
}