前言

池化思想是一種空間換時(shí)間的思想，期望使用預(yù)先創(chuàng)建好的對(duì)象來(lái)減少頻繁創(chuàng)建對(duì)象的性能開(kāi)銷(xiāo)，同時(shí)還可以對(duì)對(duì)象進(jìn)行統(tǒng)一管理，減少對(duì)象使用成本。

java中有多種池化思想的應(yīng)用，例如：數(shù)據(jù)庫(kù)連接池、線程池、字符串常量池等。

為什么使用線程池

頻繁的開(kāi)啟線程或者停止線程，線程需要重新被cpu從就緒到運(yùn)行狀態(tài)調(diào)度，需要發(fā)生cpu的上下文切換，效率非常低。

線程池作用

• 降低線程創(chuàng)建和銷(xiāo)毀的開(kāi)銷(xiāo)：通過(guò)線程池重用已經(jīng)創(chuàng)建的線程，可以避免頻繁創(chuàng)建和銷(xiāo)毀線程所造成的內(nèi)存和CPU資源開(kāi)銷(xiāo)。
• 提高線程執(zhí)行效率：線程池中的線程是經(jīng)過(guò)優(yōu)化的，通常會(huì)采用更少的線程、更高效的調(diào)度算法、更快的執(zhí)行速度等方式，以提高線程的執(zhí)行效率。
• 降低線程間競(jìng)爭(zhēng)的激烈程度：線程池中的線程數(shù)量是有限的，可以減少線程間競(jìng)爭(zhēng)的激烈程度，從而降低CPU資源的消耗。
• 提高應(yīng)用程序的可伸縮性和健壯性：線程池可以控制線程的數(shù)量和執(zhí)行速度，可以更好地滿足應(yīng)用程序的需求，從而提高應(yīng)用程序的可伸縮性和健壯性。

ThreadPoolExecutor參數(shù)

• int corePoolSize: 核心線程數(shù)
• int maximumPoolSize: 最大線程數(shù)
• long keepAliveTime: 超出corePoolSize后創(chuàng)建的線程的存活時(shí)間
• TimeUnit unit: keepAliveTime的時(shí)間單位
• BlockingQueue workQueue: 任務(wù)隊(duì)列，存放待執(zhí)行任務(wù)
• ThreadFactory threadFactory: 創(chuàng)建線程的線程工廠
• RejectedExecutionHandler handler: 拒絕策略

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

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

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

• 1）若線程數(shù)小于最大線程數(shù)，創(chuàng)建線程
• 2）若線程數(shù)等于最大線程數(shù)，執(zhí)行拒絕策略

拒絕策略

當(dāng)線程池中的線程數(shù)量達(dá)到最大值或者任務(wù)隊(duì)列已滿時(shí)，如果再有新的任務(wù)提交給線程池，線程池會(huì)拒絕接受新的任務(wù)。這時(shí)，線程池會(huì)采用一定的拒絕策略來(lái)處理這些被拒絕的任務(wù)。

Java中提供了四種拒絕策略：

• AbortPolicy：默認(rèn)策略，直接拋出RejectedExecutionException異常，表示任務(wù)被拒絕執(zhí)行。
• CallerRunsPolicy：這個(gè)策略會(huì)使用當(dāng)前客戶端線程來(lái)執(zhí)行任務(wù)。如果客戶端線程不夠，或者存在線程被阻塞，則仍然會(huì)拋出RejectedExecutionException異常。
• DiscardPolicy：這個(gè)策略會(huì)直接丟棄被拒絕的任務(wù)，不會(huì)執(zhí)行任何操作。
• DiscardOldestPolicy：這個(gè)策略會(huì)丟棄隊(duì)列中等待時(shí)間最長(zhǎng)的任務(wù)，然后執(zhí)行當(dāng)前被提交的任務(wù)。如果隊(duì)列中沒(méi)有等待時(shí)間最長(zhǎng)的任務(wù)，則會(huì)使用CallerRunsPolicy策略來(lái)處理被拒絕的任務(wù)。

也可以自定義拒絕策略，實(shí)現(xiàn)RejectedExecutionHandler接口即可。

建議自定義實(shí)現(xiàn)拒絕策略，將任務(wù)持久化到db,后期在手動(dòng)補(bǔ)償。

線程池的創(chuàng)建方式

Executors為我們提供了四種新建線程池的方式：

newCachedThreadPool()可緩存線程池

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

線程池是創(chuàng)建一個(gè)核心線程數(shù)為0，最大線程為Inter.MAX_VALUE的線程池，線程池?cái)?shù)量不確定,有空閑線程則優(yōu)先使用,沒(méi)用則創(chuàng)建新的線程處理任務(wù),處理完放入線程池。

newFixedThreadPool(): 可定長(zhǎng)度，限制最大線程數(shù)

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

創(chuàng)建一個(gè)核心線程數(shù)跟最大線程數(shù)相同的線程池,線程池?cái)?shù)量大小不變,如果有任務(wù)放入隊(duì)列,等待空閑線程。

newScheduledThreadPool(): 可定時(shí)線程池\

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

創(chuàng)建一個(gè)沒(méi)有最大線程數(shù)限制的可以定時(shí)執(zhí)行線程池,還有創(chuàng)建一個(gè)只有單個(gè)線程的可以定時(shí)執(zhí)行線程池（Executors.newSingleThreadScheduledExecutor()）

newSingleThreadExecutor(): 單線程 線程池

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

池里只有一個(gè)線程

這四種底層都是基于ThreadPoolExecutor構(gòu)造函數(shù)封裝，且都采用的無(wú)界隊(duì)列，使用時(shí)需注意防止內(nèi)存溢出。

自定義線程名稱

可以通過(guò)自定義ThreadFactory來(lái)為線程池中的線程指定名稱。

ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("custom-thread-%d").build();

線程池的五種狀態(tài)

        // runState is stored in the high-order bits
        private static final int RUNNING    = -1 << COUNT_BITS;
        private static final int SHUTDOWN   =  0 << COUNT_BITS;
        private static final int STOP       =  1 << COUNT_BITS;
        private static final int TIDYING    =  2 << COUNT_BITS;
        private static final int TERMINATED =  3 << COUNT_BITS;

• RUNNING：線程池運(yùn)行狀態(tài)，此時(shí)線程池中的任務(wù)隊(duì)列可能有等待中的任務(wù)，但線程池會(huì)持續(xù)從隊(duì)列中取出任務(wù)執(zhí)行。
• SHUTDOWN：線程池關(guān)閉狀態(tài)，此時(shí)線程池會(huì)拒絕接受新的任務(wù)，但會(huì)執(zhí)行完已經(jīng)排隊(duì)的任務(wù)，不接受新任務(wù)并不意味著已經(jīng)排隊(duì)的任務(wù)必須執(zhí)行完。
• STOP：線程池強(qiáng)制停止?fàn)顟B(tài)，此時(shí)線程池不僅會(huì)拒絕接受新的任務(wù)，而且會(huì)中斷正在執(zhí)行的任務(wù)，并終止線程池。
• TIDYING：線程池在轉(zhuǎn)換為終止?fàn)顟B(tài)時(shí)的一種特殊狀態(tài)，此時(shí)線程池會(huì)執(zhí)行鉤子方法terminated()，并等待所有任務(wù)執(zhí)行完成。
• TERMINATED：線程池終止?fàn)顟B(tài)，此時(shí)線程池中的所有任務(wù)已經(jīng)執(zhí)行完成，線程池被徹底終止。

線程數(shù)設(shè)置

• 核心線程數(shù)：線程池中始終存在的線程數(shù)量。當(dāng)任務(wù)被提交到線程池時(shí)，如果當(dāng)前運(yùn)行的線程少于核心線程數(shù)，則會(huì)創(chuàng)建一個(gè)新的線程來(lái)執(zhí)行該任務(wù)，即使其他的核心線程正在空閑狀態(tài)。因此，核心線程數(shù)通常應(yīng)該設(shè)置為預(yù)期的并發(fā)數(shù)。
• 最大線程數(shù)：允許線程池中同時(shí)存在的最大線程數(shù)量。當(dāng)線程池中的線程數(shù)量達(dá)到最大線程數(shù)時(shí)，后續(xù)提交到線程池中的任務(wù)將被暫存到任務(wù)隊(duì)列中等待處理。因此，最大線程數(shù)不應(yīng)該設(shè)置過(guò)高，否則可能會(huì)導(dǎo)致系統(tǒng)資源緊張。

合理地設(shè)置核心線程數(shù)和最大線程數(shù)可以優(yōu)化線程池的性能和響應(yīng)時(shí)間。下面是一些設(shè)置建議：

• 核心線程數(shù) = CPU核心數(shù) + 1
• 最大線程數(shù) = 核心線程數(shù) * 2
• 如果任務(wù)執(zhí)行時(shí)間較長(zhǎng)，可以適當(dāng)增加最大線程數(shù)，以避免任務(wù)堆積在隊(duì)列中無(wú)法及時(shí)處理

具體的設(shè)置需要根據(jù)實(shí)際情況來(lái)考慮，如果線程池主要執(zhí)行的是I/O密集型任務(wù)，可以適當(dāng)增加核心線程數(shù)和最大線程數(shù)，以充分利用系統(tǒng)資源。如果線程池主要執(zhí)行的是CPU密集型任務(wù)，則需要根據(jù)系統(tǒng)的CPU核心數(shù)來(lái)設(shè)置核心線程數(shù)和最大線程數(shù)，避免過(guò)度消耗CPU資源。

springboot集成線程池

package com.fandf.common.config;  
import com.fandf.common.utils.CustomThreadPoolTaskExecutor;  
import lombok.Getter;  
import lombok.Setter;  
import org.springframework.beans.factory.annotation.Value;  
import org.springframework.context.annotation.Bean;  
import org.springframework.core.task.TaskExecutor;  
import org.springframework.scheduling.annotation.EnableAsync;  
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;  
import java.util.concurrent.ThreadPoolExecutor;  
/**  
* @author fandongfeng   
*/  
@EnableAsync(proxyTargetClass = true) 
@Configuration
public class DefaultAsycTaskConfig {  
    /**  
    * 線程池維護(hù)線程的最小數(shù)量.  
    */  
    @Value("${asyc-task.corePoolSize:10}")  
    private int corePoolSize;  
    /**  
    * 線程池維護(hù)線程的最大數(shù)量  
    */  
    @Value("${asyc-task.maxPoolSize:200}")  
    private int maxPoolSize;  
    /**  
    * 隊(duì)列最大長(zhǎng)度  
    */  
    @Value("${asyc-task.queueCapacity:10000}")  
    private int queueCapacity;  
    /**  
    * 線程池前綴  
    */  
    @Value("${asyc-task.threadNamePrefix:FdfExecutor-}")  
    private String threadNamePrefix;  
    @Bean  
    public TaskExecutor taskExecutor() {  
        ThreadPoolTaskExecutor executor = new CustomThreadPoolTaskExecutor();  
        executor.setCorePoolSize(corePoolSize);  
        executor.setMaxPoolSize(maxPoolSize);  
        executor.setQueueCapacity(queueCapacity);  
        executor.setThreadNamePrefix(threadNamePrefix);  
        /*  
        rejection-policy：當(dāng)pool已經(jīng)達(dá)到max size的時(shí)候，如何處理新任務(wù)  
        CALLER_RUNS：不在新線程中執(zhí)行任務(wù)，而是有調(diào)用者所在的線程來(lái)執(zhí)行  
        */  
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());  
        executor.initialize();  
        return executor;  
    }  
}

使用

@Service
public class MyService {
    @Async("taskExecutor")
    public void doSomething() {
        // 異步執(zhí)行的任務(wù)
    }
}

到此這篇關(guān)于Java并發(fā)編程之線程池實(shí)現(xiàn)原理詳解的文章就介紹到這了,更多相關(guān)Java線程池內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論