ThreadPoolExecutor參數(shù)說明

一、ThreadPoolExecutor核心參數(shù)說明

1、corePoolSize：核心線程數(shù)

* 核心線程會(huì)一直存活，及時(shí)沒有任務(wù)需要執(zhí)行

* 當(dāng)線程數(shù)小于核心線程數(shù)時(shí)，即使有線程空閑，線程池也會(huì)優(yōu)先創(chuàng)建新線程處理

* 設(shè)置allowCoreThreadTimeout=true（默認(rèn)false）時(shí)，核心線程會(huì)超時(shí)關(guān)閉

2、queueCapacity：任務(wù)隊(duì)列容量（阻塞隊(duì)列）

* 當(dāng)核心線程數(shù)達(dá)到最大時(shí)，新任務(wù)會(huì)放在隊(duì)列中排隊(duì)等待執(zhí)行

3、maxPoolSize：最大線程數(shù)

* 當(dāng)線程數(shù)>=corePoolSize，且任務(wù)隊(duì)列已滿時(shí)。線程池會(huì)創(chuàng)建新線程來處理任務(wù)

* 當(dāng)線程數(shù)=maxPoolSize，且任務(wù)隊(duì)列已滿時(shí)，線程池會(huì)拒絕處理任務(wù)而拋出異常

4、 keepAliveTime：線程空閑時(shí)間

* 當(dāng)線程空閑時(shí)間達(dá)到keepAliveTime時(shí)，線程會(huì)退出，直到線程數(shù)量=corePoolSize

* 如果allowCoreThreadTimeout=true，則會(huì)直到線程數(shù)量=0

5、allowCoreThreadTimeout：允許核心線程超時(shí)

6、rejectedExecutionHandler：任務(wù)拒絕處理器

* 兩種情況會(huì)拒絕處理任務(wù)：

• - 當(dāng)線程數(shù)已經(jīng)達(dá)到maxPoolSize，切隊(duì)列已滿，會(huì)拒絕新任務(wù)
• - 當(dāng)線程池被調(diào)用shutdown()后，會(huì)等待線程池里的任務(wù)執(zhí)行完畢，再shutdown。如果在調(diào)用shutdown()和線程池真正shutdown之間提交任務(wù)，會(huì)拒絕新任務(wù)

* 線程池會(huì)調(diào)用rejectedExecutionHandler來處理這個(gè)任務(wù)。如果沒有設(shè)置默認(rèn)是AbortPolicy，會(huì)拋出異常

* ThreadPoolExecutor類有幾個(gè)內(nèi)部實(shí)現(xiàn)類來處理這類情況：

• - AbortPolicy 丟棄任務(wù)，拋運(yùn)行時(shí)異常
• - CallerRunsPolicy 執(zhí)行任務(wù)
• - DiscardPolicy 忽視，什么都不會(huì)發(fā)生
• - DiscardOldestPolicy 從隊(duì)列中踢出最先進(jìn)入隊(duì)列（最后一個(gè)執(zhí)行）的任務(wù)

* 實(shí)現(xiàn)RejectedExecutionHandler接口，可自定義處理器

二、ThreadPoolExecutor執(zhí)行順序

線程池按以下行為執(zhí)行任務(wù)

1. 當(dāng)線程數(shù)小于核心線程數(shù)時(shí)，創(chuàng)建線程。

2. 當(dāng)線程數(shù)大于等于核心線程數(shù)，且任務(wù)隊(duì)列未滿時(shí)，將任務(wù)放入任務(wù)隊(duì)列。

3. 當(dāng)線程數(shù)大于等于核心線程數(shù)，且任務(wù)隊(duì)列已滿

• -1 若線程數(shù)小于最大線程數(shù)，創(chuàng)建線程
• -2 若線程數(shù)等于最大線程數(shù)，拋出異常，拒絕任務(wù)

三、ThreadPoolExecutor如何設(shè)置參數(shù)

1、默認(rèn)值

* corePoolSize=1 * queueCapacity=Integer.MAX_VALUE * maxPoolSize=Integer.MAX_VALUE * keepAliveTime=60s * allowCoreThreadTimeout=false * rejectedExecutionHandler=AbortPolicy()

2、如何來設(shè)置

* 需要根據(jù)幾個(gè)值來決定

• - tasks ：每秒的任務(wù)數(shù)，假設(shè)為1000
• - taskcost：每個(gè)任務(wù)花費(fèi)時(shí)間，假設(shè)為0.1s
• - responsetime：系統(tǒng)允許容忍的最大響應(yīng)時(shí)間，假設(shè)為1s

* 做幾個(gè)計(jì)算

- corePoolSize = 每秒需要多少個(gè)線程處理？

* 一顆CPU核心同一時(shí)刻只能執(zhí)行一個(gè)線程，然后操作系統(tǒng)切換上下文，核心開始執(zhí)行另一個(gè)線程的代碼，以此類推，超過cpu核心數(shù)，就會(huì)放入隊(duì)列，如果隊(duì)列也滿了，就另起一個(gè)新的線程執(zhí)行，所有推薦：corePoolSize = ((cpu核心數(shù) * 2) + 有效磁盤數(shù))，java可以使用Runtime.getRuntime().availableProcessors()獲取cpu核心數(shù)

- queueCapacity = (coreSizePool/taskcost)*responsetime

* 計(jì)算可得 queueCapacity = corePoolSize/0.1*1。意思是隊(duì)列里的線程可以等待1s，超過了的需要新開線程來執(zhí)行

* 切記不能設(shè)置為Integer.MAX_VALUE，這樣隊(duì)列會(huì)很大，線程數(shù)只會(huì)保持在corePoolSize大小，當(dāng)任務(wù)陡增時(shí)，不能新開線程來執(zhí)行，響應(yīng)時(shí)間會(huì)隨之陡增。

- maxPoolSize = (max(tasks)- queueCapacity)/(1/taskcost)

* 計(jì)算可得 maxPoolSize = (1000-corePoolSize)/10，即(每秒并發(fā)數(shù)-corePoolSize大小) / 10

* （最大任務(wù)數(shù)-隊(duì)列容量）/每個(gè)線程每秒處理能力 = 最大線程數(shù)

• - rejectedExecutionHandler：根據(jù)具體情況來決定，任務(wù)不重要可丟棄，任務(wù)重要?jiǎng)t要利用一些緩沖機(jī)制來處理
• - keepAliveTime和allowCoreThreadTimeout采用默認(rèn)通常能滿足

ThreadPoolExecutor參數(shù)allowCoreThreadTimeOut

ThreadPoolExecutor的執(zhí)行流程有一點(diǎn)可能被吐槽過，就是只有緩存隊(duì)列已經(jīng)滿了的時(shí)候才會(huì)使用到maxPoolSize創(chuàng)建新的線程.也就是說如果corePoolSize設(shè)為0的時(shí)候，要等到隊(duì)列滿了，才會(huì)創(chuàng)建線程去執(zhí)行任務(wù)

之前有被問到，希望沒有任務(wù)的時(shí)候線程池里的線程可以停掉?？赡軐?duì)性能和資源有過考慮的人都會(huì)想到這個(gè)問題吧

今天看JDK源碼的時(shí)候發(fā)現(xiàn)了ThreadPoolExecutor在1.6的時(shí)候已經(jīng)支持了

allowCoreThreadTimeOut參數(shù)就是為此設(shè)計(jì)的

    /**
     * Sets the policy governing whether core threads may time out and
     * terminate if no tasks arrive within the keep-alive time, being
     * replaced if needed when new tasks arrive. When false, core
     * threads are never terminated due to lack of incoming
     * tasks. When true, the same keep-alive policy applying to
     * non-core threads applies also to core threads. To avoid
     * continual thread replacement, the keep-alive time must be
     * greater than zero when setting {@code true}. This method
     * should in general be called before the pool is actively used.
     *
     * @param value {@code true} if should time out, else {@code false}
     * @throws IllegalArgumentException if value is {@code true}
     *         and the current keep-alive time is not greater than zero
     *
     * @since 1.6
     */
    public void allowCoreThreadTimeOut(boolean value) {
        if (value && keepAliveTime <= 0)
            throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
        if (value != allowCoreThreadTimeOut) {
            allowCoreThreadTimeOut = value;
            if (value)
                interruptIdleWorkers();
        }
    }

在ThreadPoolExecutor構(gòu)造函數(shù)的注釋上也有明確說明：corePoolSize 的數(shù)量會(huì)一直保持，即使這些線程是空閑的，除非設(shè)置了allowCoreThreadTimeOut

/**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters and default thread factory and rejected execution handler.
     * It may be more convenient to use one of the {@link Executors} factory
     * methods instead of this general purpose constructor.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue} is null
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

當(dāng)然，在Executors的靜態(tài)工廠里面的newCachedThreadPool提供了另外一種思路

coreSize為0
SynchronousQueue最多只能有一個(gè)任務(wù)在隊(duì)列里面

也就是說這個(gè)線程池的任務(wù)會(huì)被立即分配一個(gè)線程去處理，如果沒有空閑的線程會(huì)立即創(chuàng)建線程。

在空閑的時(shí)候，線程數(shù)量會(huì)減少直至為0，這一點(diǎn)倒是滿足了要求。可是隊(duì)列中最多只會(huì)緩存一個(gè)任務(wù)，當(dāng)任務(wù)的處理速度慢于任務(wù)進(jìn)入線程池的速度時(shí)，線程數(shù)量就會(huì)不斷膨脹。如果maxPoolSize設(shè)置成一個(gè)比較小的數(shù)字時(shí)，可能就會(huì)有大量任務(wù)被拒絕策略處理。

所以正如注釋中所說，newCachedThreadPool只適合于任務(wù)處理速度很快的場(chǎng)景下。比如做一些計(jì)算，不需要依賴其它服務(wù)

/**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available.  These pools will typically improve the performance
     * of programs that execute many short-lived asynchronous tasks.
     * Calls to {@code execute} will reuse previously constructed
     * threads if available. If no existing thread is available, a new
     * thread will be created and added to the pool. Threads that have
     * not been used for sixty seconds are terminated and removed from
     * the cache. Thus, a pool that remains idle for long enough will
     * not consume any resources. Note that pools with similar
     * properties but different details (for example, timeout parameters)
     * may be created using {@link ThreadPoolExecutor} constructors.
     *
     * @return the newly created thread pool
     */
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());

總結(jié)

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

最新評(píng)論