一、序言

Java多線程編程線程池被廣泛使用，甚至成為了標(biāo)配。

線程池本質(zhì)是池化技術(shù)的應(yīng)用，和連接池類(lèi)似，創(chuàng)建連接與關(guān)閉連接屬于耗時(shí)操作，創(chuàng)建線程與銷(xiāo)毀線程也屬于重操作，為了提高效率，先提前創(chuàng)建好一批線程，當(dāng)有需要使用線程時(shí)從線程池取出，用完后放回線程池，這樣避免了頻繁創(chuàng)建與銷(xiāo)毀線程。

// 任務(wù)
Runnable runnable = () -> System.out.println(Thread.currentThread().getId());

在應(yīng)用中優(yōu)先選用線程池執(zhí)行異步任務(wù)，根據(jù)不同的場(chǎng)景選用不同的線程池，提高異步任務(wù)執(zhí)行效率。

1、普通執(zhí)行

new Thread(runnable).start();

2、線程池執(zhí)行

Executors.newSingleThreadExecutor().execute(runnable)

二、線程池基礎(chǔ)

1、核心參數(shù)

線程池的核心參數(shù)決定了池的類(lèi)型，進(jìn)而決定了池的特性。

2、參數(shù)與池的關(guān)系

Executors類(lèi)默認(rèn)創(chuàng)建線程池與參數(shù)對(duì)應(yīng)關(guān)系。

線程池對(duì)比：

根據(jù)使用場(chǎng)景選擇對(duì)應(yīng)的線程池。

1、通用對(duì)比

2、拓展對(duì)比

維護(hù)僅有一個(gè)線程的線程池有如下兩種方式，正常使用的情況下，二者差異不大；復(fù)雜使用環(huán)境下，二者存在細(xì)微的差異。用newSingleThreadExecutor方式創(chuàng)建的線程池在任何時(shí)刻至多只有一個(gè)線程，因此可以理解為用異步的方式執(zhí)行順序任務(wù)；后者初始化的時(shí)候也只有一個(gè)線程，使用過(guò)程中可能會(huì)出現(xiàn)最大線程數(shù)超過(guò)1的情況，這時(shí)要求線性執(zhí)行的任務(wù)會(huì)并行執(zhí)行，業(yè)務(wù)邏輯可能會(huì)出現(xiàn)問(wèn)題，與實(shí)際場(chǎng)景有關(guān)。

private final static ExecutorService executor = Executors.newSingleThreadExecutor();
private final static ExecutorService executor = Executors.newFixedThreadPool(1);

線程池原理：

線程池主要處理流程，任務(wù)提交之后是怎么執(zhí)行的。大致如下：

• 判斷核心線程池是否已滿(mǎn)，如果不是，則創(chuàng)建線程執(zhí)行任務(wù)
• 如果核心線程池滿(mǎn)了，判斷隊(duì)列是否滿(mǎn)了，如果隊(duì)列沒(méi)滿(mǎn)，將任務(wù)放在隊(duì)列中
• 如果隊(duì)列滿(mǎn)了，則判斷線程池是否已滿(mǎn)，如果沒(méi)滿(mǎn)，創(chuàng)建線程執(zhí)行任務(wù)
• 如果線程池也滿(mǎn)了，則按照拒絕策略對(duì)任務(wù)進(jìn)行處理

提交任務(wù)的方式：

往線程池中提交任務(wù)，主要有兩種方法：提交無(wú)返回值的任務(wù)和提交有返回值的任務(wù)。

3、無(wú)返回值任務(wù)

execute用于提交不需要返回結(jié)果的任務(wù)。

public static void main(String[] args) {
    ExecutorService executor = Executors.newFixedThreadPool(2);
    executor.execute(() -> System.out.println("hello"));
}

4、有返回值任務(wù)

submit()用于提交一個(gè)需要返回果的任務(wù)。

該方法返回一個(gè)Future對(duì)象，通過(guò)調(diào)用這個(gè)對(duì)象的get()方法，我們就能獲得返回結(jié)果。get()方法會(huì)一直阻塞，直到返回結(jié)果返回。

我們也可以使用它的重載方法get(long timeout, TimeUnit unit)，這個(gè)方法也會(huì)阻塞，但是在超時(shí)時(shí)間內(nèi)仍然沒(méi)有返回結(jié)果時(shí)，將拋出異常TimeoutException。

public static void main(String[] args) throws Exception {
    ExecutorService executor = Executors.newFixedThreadPool(2);
    Future<Long> future = executor.submit(() -> {
        System.out.println("task is executed");
        return System.currentTimeMillis();
    });
    System.out.println("task execute time is: " + future.get());
}

在提交任務(wù)時(shí)，如果無(wú)返回值任務(wù)，優(yōu)先使用execute。

關(guān)閉線程池：

在線程池使用完成之后，我們需要對(duì)線程池中的資源進(jìn)行釋放操作，這就涉及到關(guān)閉功能。我們可以調(diào)用線程池對(duì)象的shutdown()和shutdownNow()方法來(lái)關(guān)閉線程池。

這兩個(gè)方法都是關(guān)閉操作，又有什么不同呢？

• shutdown()會(huì)將線程池狀態(tài)置為SHUTDOWN，不再接受新的任務(wù)，同時(shí)會(huì)等待線程池中已有的任務(wù)執(zhí)行完成再結(jié)束。
• shutdownNow()會(huì)將線程池狀態(tài)置為SHUTDOWN，對(duì)所有線程執(zhí)行interrupt()操作，清空隊(duì)列，并將隊(duì)列中的任務(wù)返回回來(lái)。

另外，關(guān)閉線程池涉及到兩個(gè)返回boolean的方法，isShutdown()和isTerminated，分別表示是否關(guān)閉和是否終止。

三、Executors

Executors是一個(gè)線程池工廠，提供了很多的工廠方法，我們來(lái)看看它大概能創(chuàng)建哪些線程池。

// 創(chuàng)建單一線程的線程池
public static ExecutorService newSingleThreadExecutor();
// 創(chuàng)建固定數(shù)量的線程池
public static ExecutorService newFixedThreadPool(int nThreads);
// 創(chuàng)建帶緩存的線程池
public static ExecutorService newCachedThreadPool();
// 創(chuàng)建定時(shí)調(diào)度的線程池
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize);
// 創(chuàng)建流式（fork-join）線程池
public static ExecutorService newWorkStealingPool();

1、創(chuàng)建單一線程的線程池

任何時(shí)候線程池中至多只有一個(gè)線程，當(dāng)線程執(zhí)行異常終止時(shí)會(huì)自動(dòng)創(chuàng)建一個(gè)新線程替換。如果既有異步執(zhí)行任務(wù)的需求又希望任務(wù)得以順序執(zhí)行，那么此類(lèi)型線程池是首選。

若多個(gè)任務(wù)被提交到此線程池，那么會(huì)被緩存到隊(duì)列。當(dāng)線程空閑的時(shí)候，按照FIFO的方式進(jìn)行處理。

2、創(chuàng)建固定數(shù)量的線程池

創(chuàng)建核心線程與最大線程數(shù)相等的固定線程數(shù)的線程池，任何時(shí)刻至多有固定數(shù)目的線程，當(dāng)線程因異常而終止時(shí)則會(huì)自動(dòng)創(chuàng)建線程替換。

當(dāng)有新任務(wù)加入時(shí)，如果池內(nèi)線程均處于活躍狀態(tài)，則任務(wù)進(jìn)入等待隊(duì)列中，直到有空閑線程，隊(duì)列中的任務(wù)才會(huì)被順序執(zhí)行；如果池內(nèi)有非活躍線程，則任務(wù)可以立刻得以執(zhí)行。

• 如果線程的數(shù)量未達(dá)到指定數(shù)量，則創(chuàng)建線程來(lái)執(zhí)行任務(wù)
• 如果線程池的數(shù)量達(dá)到了指定數(shù)量，并且有線程是空閑的，則取出空閑線程執(zhí)行任務(wù)
• 如果沒(méi)有線程是空閑的，則將任務(wù)緩存到隊(duì)列（隊(duì)列長(zhǎng)度為Integer.MAX_VALUE）。當(dāng)線程空閑的時(shí)候，按照FIFO的方式進(jìn)行處理

3、創(chuàng)建可伸縮的線程池

這種方式創(chuàng)建的線程池，核心線程池的長(zhǎng)度為0，線程池最大長(zhǎng)度為Integer.MAX_VALUE。由于本身使用SynchronousQueue作為等待隊(duì)列的緣故，導(dǎo)致往隊(duì)列里面每插入一個(gè)元素，必須等待另一個(gè)線程從這個(gè)隊(duì)列刪除一個(gè)元素。

• 線程池可維護(hù)0到Integer.MAX_VALUE個(gè)線程資源，空閑線程默認(rèn)情況下超過(guò)60秒未使用則會(huì)被銷(xiāo)毀，長(zhǎng)期閑置的池占用較少的資源。
• 當(dāng)有新任務(wù)加入時(shí)，如果池中有空閑且尚未銷(xiāo)毀的線程，則將任務(wù)交給此線程執(zhí)行；如果沒(méi)有可用的線程，則創(chuàng)建一個(gè)新線程執(zhí)行任務(wù)并添加到池中。

4、創(chuàng)建定時(shí)調(diào)度的線程池

和上面3個(gè)工廠方法返回的線程池類(lèi)型有所不同，它返回的是ScheduledThreadPoolExecutor類(lèi)型的線程池。平時(shí)我們實(shí)現(xiàn)定時(shí)調(diào)度功能的時(shí)候，可能更多的是使用第三方類(lèi)庫(kù)，比如：quartz等。但是對(duì)于更底層的功能，我們?nèi)匀恍枰私狻?/p>

四、手動(dòng)創(chuàng)建線程池

理論上，我們可以通過(guò)Executors來(lái)創(chuàng)建線程池，這種方式非常簡(jiǎn)單。但正是因?yàn)楹?jiǎn)單，所以限制了線程池的功能。比如：無(wú)長(zhǎng)度限制的隊(duì)列，可能因?yàn)槿蝿?wù)堆積導(dǎo)致OOM，這是非常嚴(yán)重的bug，應(yīng)盡可能地避免。怎么避免？歸根結(jié)底，還是需要我們通過(guò)更底層的方式來(lái)創(chuàng)建線程池。

拋開(kāi)定時(shí)調(diào)度的線程池不管，我們看看ThreadPoolExecutor。它提供了好幾個(gè)構(gòu)造方法，但是最底層的構(gòu)造方法卻只有一個(gè)。那么，我們就從這個(gè)構(gòu)造方法著手分析。

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler);

這個(gè)構(gòu)造方法有7個(gè)參數(shù)，我們逐一來(lái)進(jìn)行分析。

• corePoolSize，線程池中的核心線程數(shù)
• maximumPoolSize，線程池中的最大線程數(shù)
• keepAliveTime，空閑時(shí)間，當(dāng)線程池?cái)?shù)量超過(guò)核心線程數(shù)時(shí)，多余的空閑線程存活的時(shí)間，即：這些線程多久被銷(xiāo)毀。
• unit，空閑時(shí)間的單位，可以是毫秒、秒、分鐘、小時(shí)和天，等等
• workQueue，等待隊(duì)列，線程池中的線程數(shù)超過(guò)核心線程數(shù)時(shí)，任務(wù)將放在等待隊(duì)列，它是一個(gè)BlockingQueue類(lèi)型的對(duì)象
• threadFactory，線程工廠，我們可以使用它來(lái)創(chuàng)建一個(gè)線程
• handler，拒絕策略，當(dāng)線程池和等待隊(duì)列都滿(mǎn)了之后，需要通過(guò)該對(duì)象的回調(diào)函數(shù)進(jìn)行回調(diào)處理

這些參數(shù)里面，基本類(lèi)型的參數(shù)都比較簡(jiǎn)單，我們不做進(jìn)一步的分析。我們更關(guān)心的是workQueue、threadFactory和handler，接下來(lái)我們將進(jìn)一步分析。

等待隊(duì)列-workQueue

等待隊(duì)列是BlockingQueue類(lèi)型的，理論上只要是它的子類(lèi)，我們都可以用來(lái)作為等待隊(duì)列。

同時(shí)，jdk內(nèi)部自帶一些阻塞隊(duì)列，我們來(lái)看看大概有哪些。

• ArrayBlockingQueue，隊(duì)列是有界的，基于數(shù)組實(shí)現(xiàn)的阻塞隊(duì)列
• LinkedBlockingQueue，隊(duì)列可以有界，也可以無(wú)界。基于鏈表實(shí)現(xiàn)的阻塞隊(duì)列
• SynchronousQueue，不存儲(chǔ)元素的阻塞隊(duì)列，每個(gè)插入操作必須等到另一個(gè)線程調(diào)用移除操作，否則插入操作將一直處于阻塞狀態(tài)。該隊(duì)列也是Executors.newCachedThreadPool()的默認(rèn)隊(duì)列
• PriorityBlockingQueue，帶優(yōu)先級(jí)的無(wú)界阻塞隊(duì)列

通常情況下，我們需要指定阻塞隊(duì)列的上界（比如1024）。另外，如果執(zhí)行的任務(wù)很多，我們可能需要將任務(wù)進(jìn)行分類(lèi)，然后將不同分類(lèi)的任務(wù)放到不同的線程池中執(zhí)行。

線程工廠-threadFactory

ThreadFactory是一個(gè)接口，只有一個(gè)方法。既然是線程工廠，那么我們就可以用它生產(chǎn)一個(gè)線程對(duì)象。來(lái)看看這個(gè)接口的定義。

public interface ThreadFactory {

    /**
     * Constructs a new {@code Thread}.  Implementations may also initialize
     * priority, name, daemon status, {@code ThreadGroup}, etc.
     *
     * @param r a runnable to be executed by new thread instance
     * @return constructed thread, or {@code null} if the request to
     *         create a thread is rejected
     */
    Thread newThread(Runnable r);
}

Executors的實(shí)現(xiàn)使用了默認(rèn)的線程工廠-DefaultThreadFactory。它的實(shí)現(xiàn)主要用于創(chuàng)建一個(gè)線程，線程的名字為pool-{poolNum}-thread-{threadNum}。

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

很多時(shí)候，我們需要自定義線程名字。我們只需要自己實(shí)現(xiàn)ThreadFactory，用于創(chuàng)建特定場(chǎng)景的線程即可。

拒絕策略-handler

所謂拒絕策略，就是當(dāng)線程池滿(mǎn)了、隊(duì)列也滿(mǎn)了的時(shí)候，我們對(duì)任務(wù)采取的措施。或者丟棄、或者執(zhí)行、或者其他…

jdk自帶4種拒絕策略，我們來(lái)看看。

• CallerRunsPolicy // 在調(diào)用者線程執(zhí)行
• AbortPolicy // 直接拋出RejectedExecutionException異常
• DiscardPolicy // 任務(wù)直接丟棄，不做任何處理
• DiscardOldestPolicy // 丟棄隊(duì)列里最舊的那個(gè)任務(wù)，再?lài)L試執(zhí)行當(dāng)前任務(wù)

這四種策略各有優(yōu)劣，比較常用的是DiscardPolicy，但是這種策略有一個(gè)弊端就是任務(wù)執(zhí)行的軌跡不會(huì)被記錄下來(lái)。所以，我們往往需要實(shí)現(xiàn)自定義的拒絕策略， 通過(guò)實(shí)現(xiàn)RejectedExecutionHandler接口的方式。

五、其它

1、配置線程池的參數(shù)

前面我們講到了手動(dòng)創(chuàng)建線程池涉及到的幾個(gè)參數(shù)，那么我們要如何設(shè)置這些參數(shù)才算是正確的應(yīng)用呢？實(shí)際上，需要根據(jù)任務(wù)的特性來(lái)分析。

• 任務(wù)的性質(zhì)：CPU密集型、IO密集型和混雜型
• 任務(wù)的優(yōu)先級(jí)：高中低
• 任務(wù)執(zhí)行的時(shí)間：長(zhǎng)中短
• 任務(wù)的依賴(lài)性：是否依賴(lài)數(shù)據(jù)庫(kù)或者其他系統(tǒng)資源

不同的性質(zhì)的任務(wù)，我們采取的配置將有所不同。在《Java并發(fā)編程實(shí)踐》中有相應(yīng)的計(jì)算公式。

通常來(lái)說(shuō)，如果任務(wù)屬于CPU密集型，那么我們可以將線程池?cái)?shù)量設(shè)置成CPU的個(gè)數(shù)，以減少線程切換帶來(lái)的開(kāi)銷(xiāo)。如果任務(wù)屬于IO密集型，我們可以將線程池?cái)?shù)量設(shè)置得更多一些，比如CPU個(gè)數(shù)*2。

PS：我們可以通過(guò)Runtime.getRuntime().availableProcessors()來(lái)獲取CPU的個(gè)數(shù)。

2、線程池監(jiān)控

如果系統(tǒng)中大量用到了線程池，那么我們有必要對(duì)線程池進(jìn)行監(jiān)控。利用監(jiān)控，我們能在問(wèn)題出現(xiàn)前提前感知到，也可以根據(jù)監(jiān)控信息來(lái)定位可能出現(xiàn)的問(wèn)題。

那么我們可以監(jiān)控哪些信息？又有哪些方法可用于我們的擴(kuò)展支持呢？

首先，ThreadPoolExecutor自帶了一些方法。

• long getTaskCount()，獲取已經(jīng)執(zhí)行或正在執(zhí)行的任務(wù)數(shù)
• long getCompletedTaskCount()，獲取已經(jīng)執(zhí)行的任務(wù)數(shù)
• int getLargestPoolSize()，獲取線程池曾經(jīng)創(chuàng)建過(guò)的最大線程數(shù)，根據(jù)這個(gè)參數(shù)，我們可以知道線程池是否滿(mǎn)過(guò)
• int getPoolSize()，獲取線程池線程數(shù)
• int getActiveCount()，獲取活躍線程數(shù)（正在執(zhí)行任務(wù)的線程數(shù)）

其次，ThreadPoolExecutor留給我們自行處理的方法有3個(gè)，它在ThreadPoolExecutor中為空實(shí)現(xiàn)（也就是什么都不做）。

• protected void beforeExecute(Thread t, Runnable r) // 任務(wù)執(zhí)行前被調(diào)用
• protected void afterExecute(Runnable r, Throwable t) // 任務(wù)執(zhí)行后被調(diào)用
• protected void terminated() // 線程池結(jié)束后被調(diào)用

六、總結(jié)

• 盡量使用手動(dòng)的方式創(chuàng)建線程池，避免使用Executors工廠類(lèi)
• 根據(jù)場(chǎng)景，合理設(shè)置線程池的各個(gè)參數(shù)，包括線程池?cái)?shù)量、隊(duì)列、線程工廠和拒絕策略

 到此這篇關(guān)于Java十分鐘入門(mén)多線程下篇的文章就介紹到這了,更多相關(guān)Java 多線程內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論