線程池是多線程編程中的核心概念，簡(jiǎn)單來(lái)說(shuō)就是一組可以執(zhí)行任務(wù)的空閑線程。

首先，我們了解一下多線程框架模型，明白為什么需要線程池。

線程是在一個(gè)進(jìn)程中可以執(zhí)行一系列指令的執(zhí)行環(huán)境，或稱(chēng)運(yùn)行程序。多線程編程指的是用多個(gè)線程并行執(zhí)行多個(gè)任務(wù)。當(dāng)然，JVM 對(duì)多線程有良好的支持。

盡管這帶來(lái)了諸多優(yōu)勢(shì)，首當(dāng)其沖的就是程序性能提高，但多線程編程也有缺點(diǎn) —— 增加了代碼復(fù)雜度、同步問(wèn)題、非預(yù)期結(jié)果和增加創(chuàng)建線程的開(kāi)銷(xiāo)。

在這篇文章中，我們來(lái)了解一下如何使用 Java 線程池來(lái)緩解這些問(wèn)題。

為什么使用線程池？

創(chuàng)建并開(kāi)啟一個(gè)線程開(kāi)銷(xiāo)很大。如果我們每次需要執(zhí)行任務(wù)時(shí)重復(fù)這個(gè)步驟，那將會(huì)是一筆巨大的性能開(kāi)銷(xiāo)，這也是我們希望通過(guò)多線程解決的問(wèn)題。

為了更好理解創(chuàng)建和開(kāi)啟一個(gè)線程的開(kāi)銷(xiāo)，讓我們來(lái)看一看 JVM 在后臺(tái)做了哪些事：

• 為線程棧分配內(nèi)存，保存每個(gè)線程方法調(diào)用的棧幀。
• 每個(gè)棧幀包括本地變量數(shù)組、返回值、操作棧和常量池
• 一些 JVM 支持本地方法，也將分配本地方法棧
• 每個(gè)線程獲得一個(gè)程序計(jì)數(shù)器，標(biāo)識(shí)處理器正在執(zhí)行哪條指令
• 系統(tǒng)創(chuàng)建本地線程，與 Java 線程對(duì)應(yīng)
• 和線程相關(guān)的描述符被添加到 JVM 內(nèi)部數(shù)據(jù)結(jié)構(gòu)
• 線程共享堆和方法區(qū)

當(dāng)然，這些步驟的具體細(xì)節(jié)取決于 JVM 和操作系統(tǒng)。

另外，更多的線程意味著更多工作量，系統(tǒng)需要調(diào)度和決定哪個(gè)線程接下來(lái)可以訪問(wèn)資源。

線程池通過(guò)減少需要的線程數(shù)量并管理線程生命周期，來(lái)幫助我們緩解性能問(wèn)題。

本質(zhì)上，線程在我們使用前一直保存在線程池中，在執(zhí)行完任務(wù)之后，線程會(huì)返回線程池等待下次使用。這種機(jī)制在執(zhí)行很多小任務(wù)的系統(tǒng)中十分有用。

Java 線程池

Java 通過(guò) executor 對(duì)象來(lái)實(shí)現(xiàn)自己的線程池模型�？梢允褂� executor 接口或其他線程池的實(shí)現(xiàn)，它們都允許細(xì)粒度的控制。

java.util.concurrent 包中有以下接口：

• Executor —— 執(zhí)行任務(wù)的簡(jiǎn)單接口
• ExecutorService —— 一個(gè)較復(fù)雜的接口，包含額外方法來(lái)管理任務(wù)和 executor 本身
• ScheduledExecutorService —— 擴(kuò)展自 ExecutorService，增加了執(zhí)行任務(wù)的調(diào)度方法

除了這些接口，這個(gè)包中也提供了 Executors 類(lèi)直接獲取實(shí)現(xiàn)了這些接口的 executor 實(shí)例

一般來(lái)說(shuō)，一個(gè) Java 線程池包含以下部分：

• 工作線程的池子，負(fù)責(zé)管理線程
• 線程工廠，負(fù)責(zé)創(chuàng)建新線程
• 等待執(zhí)行的任務(wù)隊(duì)列

在下面的章節(jié)，讓我們仔細(xì)看一看 Java 類(lèi)和接口如何為線程池提供支持。

Executors 類(lèi)和 Executor 接口

Executors 類(lèi)包含工廠方法創(chuàng)建不同類(lèi)型的線程池，Executor 是個(gè)簡(jiǎn)單的線程池接口，只有一個(gè) execute() 方法。

我們通過(guò)一個(gè)例子來(lái)結(jié)合使用這兩個(gè)類(lèi)(接口)，首先創(chuàng)建一個(gè)單線程的線程池，然后用它執(zhí)行一個(gè)簡(jiǎn)單的語(yǔ)句：

Executor executor = Executors.newSingleThreadExecutor();
executor.execute(() -> System.out.println("Single thread pool test"));

注意語(yǔ)句寫(xiě)成了 lambda 表達(dá)式，會(huì)被自動(dòng)推斷成 Runnable 類(lèi)型。

如果有工作線程可用，execute() 方法將執(zhí)行語(yǔ)句，否則就把 Runnable 任務(wù)放進(jìn)隊(duì)列，等待線程可用。

基本上，executor 代替了顯式創(chuàng)建和管理線程。

Executors 類(lèi)里的工廠方法可以創(chuàng)建很多類(lèi)型的線程池：

• newSingleThreadExecutor()：包含單個(gè)線程和無(wú)界隊(duì)列的線程池，同一時(shí)間只能執(zhí)行一個(gè)任務(wù)
• newFixedThreadPool()：包含固定數(shù)量線程并共享無(wú)界隊(duì)列的線程池；當(dāng)所有線程處于工作狀態(tài)，有新任務(wù)提交時(shí)，任務(wù)在隊(duì)列中等待，直到一個(gè)線程變?yōu)榭捎脿顟B(tài)
• newCachedThreadPool()：只有需要時(shí)創(chuàng)建新線程的線程池
• newWorkStealingThreadPool()：基于工作竊�。╳ork-stealing）算法的線程池，后面章節(jié)詳細(xì)說(shuō)明

接下來(lái)，讓我們看一下 ExecutorService 接口提供了哪些新功能

ExecutorService

創(chuàng)建 ExecutorService 方式之一便是通過(guò) Excutors 類(lèi)的工廠方法。

ExecutorService executor = Executors.newFixedThreadPool(10);

Besides the execute() method, this interface also defines a similar submit() method that can return a Future object:

除了 execute() 方法，接口也定義了相似的 submit() 方法，這個(gè)方法可以返回一個(gè) Future 對(duì)象。

Callable<Double> callableTask = () -> {
    return employeeService.calculateBonus(employee);
};
Future<Double> future = executor.submit(callableTask);
// execute other operations
try {
    if (future.isDone()) {
        double result = future.get();
    }
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}

從上面的例子可以看到，Future 接口可以返回 Callable 類(lèi)型任務(wù)的結(jié)果，而且能顯示任務(wù)的執(zhí)行狀態(tài)。

當(dāng)沒(méi)有任務(wù)等待執(zhí)行時(shí)，ExecutorService 并不會(huì)自動(dòng)銷(xiāo)毀，所以你可以使用 shutdown() 或 shutdownNow() 來(lái)顯式關(guān)閉它。

executor.shutdown();

ScheduledExecutorService

這是 ExecutorService 的一個(gè)子接口，增加了調(diào)度任務(wù)的方法。

ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);

schedule() 方法的參數(shù)指定執(zhí)行的方法、延時(shí)和 TimeUnit

Future<Double> future = executor.schedule(callableTask, 2, TimeUnit.MILLISECONDS);

另外，這個(gè)接口定義了其他兩個(gè)方法：

executor.scheduleAtFixedRate(
  () -> System.out.println("Fixed Rate Scheduled"), 2, 2000, TimeUnit.MILLISECONDS);

executor.scheduleWithFixedDelay(
  () -> System.out.println("Fixed Delay Scheduled"), 2, 2000, TimeUnit.MILLISECONDS);

scheduleAtFixedRate() 方法延時(shí) 2 毫秒執(zhí)行任務(wù)，然后每 2 秒重復(fù)一次。相似的，scheduleWithFixedDelay() 方法延時(shí) 2 毫秒后執(zhí)行第一次，然后在上一次執(zhí)行完成 2 秒后再次重復(fù)執(zhí)行。

在下面的章節(jié)，我們來(lái)看一下 ExecutorService 接口的兩個(gè)實(shí)現(xiàn)：ThreadPoolExecutor 和 ForkJoinPool。

ThreadPoolExecutor

這個(gè)線程池的實(shí)現(xiàn)增加了配置參數(shù)的能力。創(chuàng)建 ThreadPoolExecutor 對(duì)象最方便的方式就是通過(guò) Executors 工廠方法：

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);

這種情況下，線程池按照默認(rèn)值預(yù)配置了參數(shù)。線程數(shù)量由以下參數(shù)控制：

• corePoolSize 和 maximumPoolSize：表示線程數(shù)量的范圍
• keepAliveTime：決定了額外線程存活時(shí)間

我們深入了解一下這些參數(shù)如何使用。

當(dāng)一個(gè)任務(wù)被提交時(shí)，如果執(zhí)行中的線程數(shù)量小于 corePoolSize，一個(gè)新的線程被創(chuàng)建。如果運(yùn)行的線程數(shù)量大于 corePoolSize，但小于 maximumPoolSize，并且任務(wù)隊(duì)列已滿時(shí)，依然會(huì)創(chuàng)建新的線程。如果多于 corePoolSize 的線程空閑時(shí)間超過(guò) keepAliveTime，它們會(huì)被終止。

上面那個(gè)例子中，newFixedThreadPool() 方法創(chuàng)建的線程池，corePoolSize=maximumPoolSize=10 并且 keepAliveTime 為 0 秒。

如果你使用 newCachedThreadPool() 方法，創(chuàng)建的線程池 maximumPoolSize 為 Integer.MAX_VALUE，并且 keepAliveTime 為 60 秒。

ThreadPoolExecutor cachedPoolExecutor 
  = (ThreadPoolExecutor) Executors.newCachedThreadPool();

The parameters can also be set through a constructor or through setter methods:

這些參數(shù)也可以通過(guò)構(gòu)造函數(shù)或setter方法設(shè)置：

ThreadPoolExecutor executor = new ThreadPoolExecutor(
  4, 6, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()
);
executor.setMaximumPoolSize(8);

ThreadPoolExecutor 的一個(gè)子類(lèi)便是 ScheduledThreadPoolExecutor，它實(shí)現(xiàn)了 ScheduledExecutorService 接口。你可以通過(guò) newScheduledThreadPool() 工廠方法來(lái)創(chuàng)建這種類(lèi)型的線程池。

ScheduledThreadPoolExecutor executor 
  = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(5);

上面語(yǔ)句創(chuàng)建了一個(gè)線程池，corePoolSize 為 5，maximumPoolSize 無(wú)限制，keepAliveTime 為 0 秒。

ForkJoinPool

另一個(gè)線程池的實(shí)現(xiàn)是 ForkJoinPool 類(lèi)。它實(shí)現(xiàn)了 ExecutorService 接口，并且是 Java 7 中 fork/join 框架的重要組件。

fork/join 框架基于“工作竊取算法”。簡(jiǎn)而言之，意思就是執(zhí)行完任務(wù)的線程可以從其他運(yùn)行中的線程“竊取”工作。

ForkJoinPool 適用于任務(wù)創(chuàng)建子任務(wù)的情況，或者外部客戶端創(chuàng)建大量小任務(wù)到線程池。

這種線程池的工作流程如下：

• 創(chuàng)建 ForkJoinTask 子類(lèi)
• 根據(jù)某種條件將任務(wù)切分成子任務(wù)
• 調(diào)用執(zhí)行任務(wù)
• 將任務(wù)結(jié)果合并
• 實(shí)例化對(duì)象并添加到池中

創(chuàng)建一個(gè) ForkJoinTask，你可以選擇 RecursiveAction 或 RecursiveTask 這兩個(gè)子類(lèi)，后者有返回值。

我們來(lái)實(shí)現(xiàn)一個(gè)繼承 RecursiveTask 的類(lèi)，計(jì)算階乘，并把任務(wù)根據(jù)閾值劃分成子任務(wù)。

public class FactorialTask extends RecursiveTask<BigInteger> {
    private int start = 1;
    private int n;
    private static final int THRESHOLD = 20;

    // standard constructors

    @Override
    protected BigInteger compute() {
        if ((n - start) >= THRESHOLD) {
            return ForkJoinTask.invokeAll(createSubtasks())
              .stream()
              .map(ForkJoinTask::join)
              .reduce(BigInteger.ONE, BigInteger::multiply);
        } else {
            return calculate(start, n);
        }
    }
}

這個(gè)類(lèi)需要實(shí)現(xiàn)的主要方法就是重寫(xiě) compute() 方法，用于合并每個(gè)子任務(wù)的結(jié)果。

具體劃分任務(wù)邏輯在 createSubtasks() 方法中：

private Collection<FactorialTask> createSubtasks() {
    List<FactorialTask> dividedTasks = new ArrayList<>();
    int mid = (start + n) / 2;
    dividedTasks.add(new FactorialTask(start, mid));
    dividedTasks.add(new FactorialTask(mid + 1, n));
    return dividedTasks;
}

最后，calculate() 方法包含一定范圍內(nèi)的乘數(shù)。

private BigInteger calculate(int start, int n) {
    return IntStream.rangeClosed(start, n)
      .mapToObj(BigInteger::valueOf)
      .reduce(BigInteger.ONE, BigInteger::multiply);
}

接下來(lái)，任務(wù)可以添加到線程池：

ForkJoinPool pool = ForkJoinPool.commonPool();
BigInteger result = pool.invoke(new FactorialTask(100));

ThreadPoolExecutor 與 ForkJoinPool 對(duì)比

初看上去，似乎 fork/join 框架帶來(lái)性能提升。但是這取決于你所解決問(wèn)題的類(lèi)型。

當(dāng)選擇線程池時(shí)，非常重要的一點(diǎn)是牢記創(chuàng)建、管理線程以及線程間切換執(zhí)行會(huì)帶來(lái)的開(kāi)銷(xiāo)。

ThreadPoolExecutor 可以控制線程數(shù)量和每個(gè)線程執(zhí)行的任務(wù)。這很適合你需要在不同的線程上執(zhí)行少量巨大的任務(wù)。

相比較而言，ForkJoinPool 基于線程從其他線程“竊取”任務(wù)。正因如此，當(dāng)任務(wù)可以分割成小任務(wù)時(shí)可以提高效率。

為了實(shí)現(xiàn)工作竊取算法，fork/join 框架使用兩種隊(duì)列：

• 包含所有任務(wù)的主要隊(duì)列
• 每個(gè)線程的任務(wù)隊(duì)列

當(dāng)線程執(zhí)行完自己任務(wù)隊(duì)列中的任務(wù)，它們?cè)噲D從其他隊(duì)列獲取任務(wù)。為了使這一過(guò)程更加高效，線程任務(wù)隊(duì)列使用雙端隊(duì)列（double ended queue）數(shù)據(jù)結(jié)構(gòu)，一端與線程交互，另一端用于“竊取”任務(wù)。

來(lái)自The H Developer的圖很好的表現(xiàn)出了這一過(guò)程：

和這種模型相比，ThreadPoolExecutor 只使用一個(gè)主要隊(duì)列。

最后要注意的一點(diǎn) ForkJoinPool 只適用于任務(wù)可以創(chuàng)建子任務(wù)。否則它和 ThreadPoolExecutor 沒(méi)區(qū)別，甚至開(kāi)銷(xiāo)更大。

跟蹤線程池的執(zhí)行

現(xiàn)在我們對(duì) Java 線程池生態(tài)系統(tǒng)有了基本的了解，讓我們通過(guò)一個(gè)使用了線程池的應(yīng)用，來(lái)看一看執(zhí)行中到底發(fā)生了什么。

通過(guò)在 FactorialTask 的構(gòu)造函數(shù)和 calculate() 方法中加入日志語(yǔ)句，你可以看到下面調(diào)用序列：

13:07:33.123 [main] INFO ROOT - New FactorialTask Created
13:07:33.123 [main] INFO ROOT - New FactorialTask Created
13:07:33.123 [main] INFO ROOT - New FactorialTask Created
13:07:33.123 [main] INFO ROOT - New FactorialTask Created
13:07:33.123 [ForkJoinPool.commonPool-worker-1] INFO ROOT - New FactorialTask Created
13:07:33.123 [ForkJoinPool.commonPool-worker-1] INFO ROOT - New FactorialTask Created
13:07:33.123 [main] INFO ROOT - New FactorialTask Created
13:07:33.123 [main] INFO ROOT - New FactorialTask Created
13:07:33.123 [main] INFO ROOT - Calculate factorial from 1 to 13
13:07:33.123 [ForkJoinPool.commonPool-worker-1] INFO ROOT - New FactorialTask Created
13:07:33.123 [ForkJoinPool.commonPool-worker-2] INFO ROOT - New FactorialTask Created
13:07:33.123 [ForkJoinPool.commonPool-worker-1] INFO ROOT - New FactorialTask Created
13:07:33.123 [ForkJoinPool.commonPool-worker-2] INFO ROOT - New FactorialTask Created
13:07:33.123 [ForkJoinPool.commonPool-worker-1] INFO ROOT - Calculate factorial from 51 to 63
13:07:33.123 [ForkJoinPool.commonPool-worker-2] INFO ROOT - Calculate factorial from 76 to 88
13:07:33.123 [ForkJoinPool.commonPool-worker-3] INFO ROOT - Calculate factorial from 64 to 75
13:07:33.163 [ForkJoinPool.commonPool-worker-3] INFO ROOT - New FactorialTask Created
13:07:33.163 [main] INFO ROOT - Calculate factorial from 14 to 25
13:07:33.163 [ForkJoinPool.commonPool-worker-3] INFO ROOT - New FactorialTask Created
13:07:33.163 [ForkJoinPool.commonPool-worker-2] INFO ROOT - Calculate factorial from 89 to 100
13:07:33.163 [ForkJoinPool.commonPool-worker-3] INFO ROOT - Calculate factorial from 26 to 38
13:07:33.163 [ForkJoinPool.commonPool-worker-3] INFO ROOT - Calculate factorial from 39 to 50

你可以看到創(chuàng)建了很多任務(wù)，但只有 3 個(gè)工作線程 —— 所以任務(wù)通過(guò)線程池被可用線程處理。

也可以看到在放到執(zhí)行池之前，主線程中對(duì)象如何被創(chuàng)建。

使用 Prefix 這一類(lèi)可視化的日志工具是一個(gè)很棒的方式來(lái)探索和理解運(yùn)行時(shí)的線程池。

記錄線程池日志的核心便是保證在日志信息中方便辨識(shí)線程名字。Log4J2 通過(guò)使用布局能夠很好完成這種工作。

使用線程池的潛在風(fēng)險(xiǎn)

盡管線程池有巨大優(yōu)勢(shì)，你在使用中仍會(huì)遇到一些問(wèn)題，比如：

• 用的線程池過(guò)大或過(guò)�。喝绻�線程池包含太多線程，會(huì)明顯的影響應(yīng)用的性能；另一方面，線程池太小并不能帶來(lái)所期待的性能提升。
• 正如其他多線程情形一樣，死鎖也會(huì)發(fā)生。舉個(gè)例子，一個(gè)任務(wù)可能等待另一個(gè)任務(wù)完成，而后者并沒(méi)有可用線程處理執(zhí)行。所以說(shuō)避免任務(wù)之間的依賴(lài)是個(gè)好習(xí)慣。
• 等待執(zhí)行時(shí)間很長(zhǎng)的任務(wù)：為了避免長(zhǎng)時(shí)間阻塞線程，你可以指定最大等待時(shí)間，并決定過(guò)期任務(wù)是拒絕處理還是重新加入隊(duì)列。

為了降低風(fēng)險(xiǎn)，你必須根據(jù)要處理的任務(wù)，來(lái)謹(jǐn)慎選擇線程池的類(lèi)型和參數(shù)。對(duì)你的系統(tǒng)進(jìn)行壓力測(cè)試也是值得的，它可以幫你獲取真實(shí)環(huán)境下的系統(tǒng)行為數(shù)據(jù)。

結(jié)論

線程池有很大優(yōu)勢(shì)，簡(jiǎn)單來(lái)說(shuō)就是可以將任務(wù)的執(zhí)行從線程的創(chuàng)建和管理中分離。另外，如果使用得當(dāng)，它們可以極大提高應(yīng)用的性能。

如果你學(xué)會(huì)充分利用線程池，Java 生態(tài)系統(tǒng)好處便是其中有很多成熟穩(wěn)定的線程池實(shí)現(xiàn)。

原文鏈接： stackify 翻譯： ImportNew.com - 一杯哈希不加鹽
譯文鏈接： http://www.importnew.com/29212.html
[ 轉(zhuǎn)載請(qǐng)保留原文出處、譯者和譯文鏈接。]

標(biāo)簽： 代碼

版權(quán)申明：本站文章部分自網(wǎng)絡(luò)，如有侵權(quán)，請(qǐng)聯(lián)系：west999com@outlook.com
特別注意：本站所有轉(zhuǎn)載文章言論不代表本站觀點(diǎn)！
本站所提供的圖片等素材，版權(quán)歸原作者所有，如需使用，請(qǐng)與原作者聯(lián)系。