線(xiàn)程池是很常用的并發(fā)框架，幾乎所有需要異步和并發(fā)處理任務(wù)的程序都可用到線(xiàn)程池。
使用線(xiàn)程池的好處如下：

• 降低資源消耗：可重復(fù)利用已創(chuàng)建的線(xiàn)程池，降低創(chuàng)建和銷(xiāo)毀帶來(lái)的消耗；
• 提高響應(yīng)速度：任務(wù)到達(dá)時(shí)，可立即執(zhí)行，無(wú)需等待線(xiàn)程創(chuàng)建；
• 提高線(xiàn)程的可管理性：線(xiàn)程池可對(duì)線(xiàn)程統(tǒng)一分配、調(diào)優(yōu)和監(jiān)控。

原理

線(xiàn)程池的原理非常簡(jiǎn)單，這里用處理流程來(lái)概括：

• 線(xiàn)程池判斷核心池里的線(xiàn)程是否都在執(zhí)行任務(wù)，如果不是，創(chuàng)建一個(gè)新的線(xiàn)程來(lái)執(zhí)行任務(wù)；
• 如果核心線(xiàn)程池已滿(mǎn)，則將新任務(wù)存在工作隊(duì)列中；
• 如果工作隊(duì)列滿(mǎn)了，線(xiàn)程數(shù)量沒(méi)有達(dá)到線(xiàn)程池上限的前提下，新建一個(gè)線(xiàn)程來(lái)執(zhí)行任務(wù)；
• 線(xiàn)程數(shù)量達(dá)到上限，則觸發(fā)飽和策略來(lái)處理這個(gè)任務(wù)；

使用工作隊(duì)列，是為了盡可能降低線(xiàn)程創(chuàng)建的開(kāi)銷(xiāo)。工作隊(duì)列用阻塞隊(duì)列來(lái)實(shí)現(xiàn)。

阻塞隊(duì)列

阻塞隊(duì)列（BlockingQueue）是指支持阻塞的插入和移除元素的隊(duì)列。

• 阻塞的插入：當(dāng)隊(duì)列滿(mǎn)時(shí)，阻塞插入元素的線(xiàn)程，直到隊(duì)列不滿(mǎn)；
• 阻塞的移除：當(dāng)隊(duì)列為空，阻塞移除元素的線(xiàn)程，直到隊(duì)列不為空；

原理：使用通知者模式實(shí)現(xiàn)。當(dāng)生產(chǎn)者往滿(mǎn)的隊(duì)列中添加元素時(shí)，會(huì)阻塞生產(chǎn)者。消費(fèi)者移除元素時(shí)，會(huì)通知生產(chǎn)者當(dāng)前隊(duì)列可用。

阻塞隊(duì)列有以下三種類(lèi)型，分別是：

• 有界阻塞隊(duì)列：ArrayBlockingQueue（數(shù)組），LinkedBlockingQueue（鏈表）
• 無(wú)界阻塞隊(duì)列：LinkedTransferQueue（鏈表），PriorityBlockingQueue（支持優(yōu)先級(jí)排序），DelayQueue（支持延時(shí)獲取元素的無(wú)界阻塞隊(duì)列）
• 同步移交隊(duì)列：SynchronousQueue

有界阻塞隊(duì)列

主要包括ArrayBlockingQueue（數(shù)組），LinkedBlockingQueue（鏈表）兩種。有界隊(duì)列大小與線(xiàn)程數(shù)量大小相互配合，隊(duì)列容量大線(xiàn)程數(shù)量小時(shí)，可減少上下文切換降低cpu使用率，但是會(huì)降低吞吐量。

無(wú)界阻塞隊(duì)列

比較常用的是LinkedTransferQueue。FixedThreadPool就是用這個(gè)實(shí)現(xiàn)的。無(wú)界阻塞隊(duì)列要慎重使用，因?yàn)樵谀承┣闆r，可能會(huì)導(dǎo)致大量的任務(wù)堆積到隊(duì)列中，導(dǎo)致內(nèi)存飆升。

同步移交隊(duì)列

SynchronousQueue。不存儲(chǔ)元素的阻塞隊(duì)列，每一個(gè)put操作必須等待一個(gè)take操作，否則不能繼續(xù)添加元素。用于實(shí)現(xiàn)CachedThreadPool線(xiàn)程池。

各個(gè)線(xiàn)程池所使用的任務(wù)隊(duì)列映射關(guān)系如下：

線(xiàn)程池阻塞隊(duì)列

FixedThreadPoolLinkedBlockingQueueSingleThreadExecutorLinkedBlockingQueueCachedThreadExecutorSynchronousQueueScheduledThreadPoolExecutorLinkedBlockingQueue

實(shí)現(xiàn)類(lèi)分析

ThreadPoolExecutor是Java線(xiàn)程池的實(shí)現(xiàn)類(lèi)，是Executor接口派生出來(lái)的最核心的類(lèi)。依賴(lài)關(guān)系圖如下：

這里不得不提到Executor框架，該框架包含三大部分，如下：

• 任務(wù)。被執(zhí)行任務(wù)需要實(shí)現(xiàn)的接口：Runnable和Callable;
• 任務(wù)執(zhí)行。即上述核心接口Executor以及繼承而來(lái)的ExecutorService。ExecutorService派生出如下兩個(gè)類(lèi)：ThreadPoolExecutor：線(xiàn)程池核心實(shí)現(xiàn)類(lèi)；ScheduledThreadPoolExecutor：用來(lái)做定時(shí)任務(wù)；
• 異步計(jì)算的結(jié)果。接口Future和實(shí)現(xiàn)Future接口的FutureTask類(lèi)。線(xiàn)程池創(chuàng)建

new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, milliseconds, runnableTaskQueue, handler)

構(gòu)造方法如下：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

參數(shù)說(shuō)明：

• corePoolSize：核心池的線(xiàn)程數(shù)量；
• workQueue：用于保存任務(wù)的工作隊(duì)列；
• maximumPoolSize：最大線(xiàn)程池的大?。?/li>
• keepAliveTime：當(dāng)線(xiàn)程數(shù)量大于核心池線(xiàn)程數(shù)量時(shí)，keepAliveTime為多余的空閑線(xiàn)程等待新任務(wù)的最長(zhǎng)時(shí)間，超過(guò)這個(gè)時(shí)間，多余的線(xiàn)程會(huì)被終止；
• TimeUnit：keepAliveTime的單位；
• ThreadFactory：線(xiàn)程工廠(chǎng)，可以給線(xiàn)程設(shè)置名字；
• handler：飽和策略。當(dāng)隊(duì)列和線(xiàn)程池都滿(mǎn)了，會(huì)觸發(fā)飽和策略，來(lái)處理新提交的任務(wù)。飽和策略以下幾種：AbortPolicy：直接拋出異常；CallerRunsPolicy：只用調(diào)用者所在線(xiàn)程來(lái)運(yùn)行任務(wù)；DiscardOldestPolicy：丟棄最近一個(gè)任務(wù)并執(zhí)行當(dāng)前任務(wù)；DiscardPolicy：不處理，丟棄掉。

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

使用工具類(lèi)Executors可創(chuàng)建三種類(lèi)型的線(xiàn)程池：FixedThreadPool、SingleThreadExecutor、CachedThreadPool。本質(zhì)上也是調(diào)用上述構(gòu)造方法。理解了前文的參數(shù)解釋?zhuān)旅嫒N線(xiàn)程池也就容易理解了。

FixedThreadPool

可重用固定線(xiàn)程數(shù)的線(xiàn)程池。

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

工作流程如下：

• 如果當(dāng)前運(yùn)行的線(xiàn)程數(shù)少于corePoolSize，則創(chuàng)建新線(xiàn)程來(lái)執(zhí)行任務(wù)；
• 線(xiàn)程數(shù)等于corePoolSize之后，新任務(wù)加入LinkedBlockingQueue（無(wú)界阻塞隊(duì)列）。因?yàn)樽畲缶€(xiàn)程數(shù)maximumPoolSize參數(shù)值等于corePoolSize，不會(huì)產(chǎn)生多余線(xiàn)程；
• 線(xiàn)程執(zhí)行完任務(wù)之后會(huì)反復(fù)從LinkedBlockingQueue中獲取任務(wù)來(lái)執(zhí)行。

SingleThreadExecutor

單個(gè)worker線(xiàn)程的線(xiàn)程池

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

SingleThreadExecutor與FixedThreadPool的區(qū)別在于，maximumPoolSize和corePoolSize都設(shè)置成了1，其它參數(shù)都一樣。

• CachedThreadPool

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

CachedThreadPool將corePoolSize設(shè)置為0，maximumPoolSize設(shè)置為無(wú)限大，同時(shí)使用了一個(gè)沒(méi)有容量的工作隊(duì)列SynchronousQueue。這個(gè)線(xiàn)程池沒(méi)有固定的核心線(xiàn)程，而是根據(jù)需要?jiǎng)?chuàng)建新線(xiàn)程。

工作流程：

• 有新任務(wù)時(shí)，主線(xiàn)程執(zhí)行SynchronousQueue.offer操作，空閑線(xiàn)程執(zhí)行SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS)操作，配對(duì)成功則將任務(wù)交給空閑線(xiàn)程執(zhí)行；
• 當(dāng)沒(méi)有空閑線(xiàn)程時(shí)，上面的配對(duì)操作失敗，此時(shí)會(huì)創(chuàng)建一個(gè)新線(xiàn)程來(lái)執(zhí)行任務(wù)；
• 任務(wù)執(zhí)行完畢后，空閑線(xiàn)程會(huì)等待60秒。60秒內(nèi)如果有新任務(wù)，就立即執(zhí)行，否則時(shí)間一過(guò)線(xiàn)程就終止。

線(xiàn)程池關(guān)閉

調(diào)用shutdown或者shutdownNow方法可關(guān)閉線(xiàn)程池。原理是遍歷線(xiàn)程池中所有工作線(xiàn)程，調(diào)用interrupt方法來(lái)中斷線(xiàn)程。

• shutdown：將線(xiàn)程置為SHUTDOWN狀態(tài)，不能接受新的任務(wù)，等待所有任務(wù)執(zhí)行完畢；
• shutdownNow：將線(xiàn)程置為STOP狀態(tài)，不能接受新的任務(wù)，嘗試去終止正在執(zhí)行的惡任務(wù)；

這里涉及到ThreadPoolExecutor中定義的線(xiàn)程的五種狀態(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：接受新任務(wù)，處理任務(wù)；
• SHUTDOWN：不接受新任務(wù)，但會(huì)把隊(duì)列中任務(wù)處理完；
• STOP：不接受新任務(wù)，不處理隊(duì)列中的任務(wù)，并且終止正在處理的任務(wù)；
• TIDYING：正在執(zhí)行的任務(wù)和隊(duì)列都為空，進(jìn)入該狀態(tài)，將要執(zhí)行terminated()；
• TERMINATED：所有terminated()方法執(zhí)行完畢，線(xiàn)程池徹底終止。

當(dāng)隊(duì)列和正在執(zhí)行的任務(wù)都為空時(shí)，由SHUTDOWN轉(zhuǎn)化為T(mén)IDYING；當(dāng)正在執(zhí)行的任務(wù)為空，由STOP轉(zhuǎn)化為T(mén)IDYING。

本博客從線(xiàn)程池的原理介紹作為切入點(diǎn)，分析了線(xiàn)程池中尤為關(guān)鍵的組件：阻塞隊(duì)列。同時(shí)分析了線(xiàn)程池的核心實(shí)現(xiàn)類(lèi)ThreadPoolExecutor。以線(xiàn)程池的創(chuàng)建和關(guān)閉的思路，梳理了相關(guān)知識(shí)點(diǎn)，包括三種常用線(xiàn)程池介紹以及線(xiàn)程池五種狀態(tài)。

到此這篇關(guān)于Java 線(xiàn)程池全面總結(jié)與詳解的文章就介紹到這了,更多相關(guān)Java 線(xiàn)程池內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論