一、序言

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

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

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

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

1、普通執(zhí)行

new Thread(runnable).start();

2、線程池執(zhí)行

Executors.newSingleThreadExecutor().execute(runnable)

二、線程池基礎(chǔ)

1、核心參數(shù)

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

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

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

線程池對比：

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

1、通用對比

2、拓展對比

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

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

線程池原理：

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

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

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

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

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

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

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

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

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

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

我們也可以使用它的重載方法get(long timeout, TimeUnit unit)，這個方法也會阻塞，但是在超時時間內(nèi)仍然沒有返回結(jié)果時，將拋出異常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ù)時，如果無返回值任務(wù)，優(yōu)先使用execute。

關(guān)閉線程池：

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

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

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

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

三、Executors

Executors是一個線程池工廠，提供了很多的工廠方法，我們來看看它大概能創(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)建定時調(diào)度的線程池
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize);
// 創(chuàng)建流式（fork-join）線程池
public static ExecutorService newWorkStealingPool();

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

等待隊列-workQueue

等待隊列是BlockingQueue類型的，理論上只要是它的子類，我們都可以用來作為等待隊列。

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

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

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

線程工廠-threadFactory

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

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的實現(xiàn)使用了默認(rèn)的線程工廠-DefaultThreadFactory。它的實現(xiàn)主要用于創(chuàng)建一個線程，線程的名字為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;
    }
}

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

拒絕策略-handler

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

jdk自帶4種拒絕策略，我們來看看。

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

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

五、其它

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

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

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

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

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

PS：我們可以通過Runtime.getRuntime().availableProcessors()來獲取CPU的個數(shù)。

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

如果系統(tǒng)中大量用到了線程池，那么我們有必要對線程池進(jìn)行監(jiān)控。利用監(jiān)控，我們能在問題出現(xiàn)前提前感知到，也可以根據(jù)監(jiān)控信息來定位可能出現(xià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)建過的最大線程數(shù)，根據(jù)這個參數(shù)，我們可以知道線程池是否滿過
• int getPoolSize()，獲取線程池線程數(shù)
• int getActiveCount()，獲取活躍線程數(shù)（正在執(zhí)行任務(wù)的線程數(shù)）

其次，ThreadPoolExecutor留給我們自行處理的方法有3個，它在ThreadPoolExecutor中為空實現(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é)

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

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

最新評論