🙈作者簡介：練習(xí)時長兩年半的Java up主
🙉個人主頁：程序員老茶
🙊 ps:點贊👍是免費的，卻可以讓寫博客的作者開心好久好久😎
📚系列專欄：Java全棧，計算機系列（火速更新中）
💭 格言：種一棵樹最好的時間是十年前，其次是現(xiàn)在
🏡動動小手，點個關(guān)注不迷路，感謝寶子們一鍵三連

課程名：Java

內(nèi)容/作用：知識點/設(shè)計/實驗/作業(yè)/練習(xí)

學(xué)習(xí)：JUC的線程池架構(gòu)

JUC的線程池架構(gòu)

本文主要介紹Java中如何使用java.util.concurrent包中的線程池（ExecutorService和ThreadPoolExecutor）來實現(xiàn)高并發(fā)、高可用的系統(tǒng)。我們將從線程池的基本概念、使用方法、核心組件以及自定義線程池四個方面進行闡述。

一、線程池的基本概念

線程池是一種管理線程的機制，它可以有效地控制線程的數(shù)量，避免大量線程之間的切換導(dǎo)致性能下降。線程池中的線程可以被復(fù)用，當一個任務(wù)完成后，線程不會被銷毀，而是被重新分配給新的任務(wù)。

二、線程池的使用方法

1. 創(chuàng)建線程池

在Java中，可以通過Executors工具類來創(chuàng)建不同類型的線程池。例如，創(chuàng)建一個固定大小的線程池：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class ThreadPoolDemo {public static void main(String[] args) {ExecutorService executorService = Executors.newFixedThreadPool(5);}
}

2. 提交任務(wù)到線程池

使用submit()方法將任務(wù)提交到線程池中執(zhí)行：

executorService.submit(new Runnable() {@Overridepublic void run() {System.out.println("任務(wù)執(zhí)行中...");}
});

3. 關(guān)閉線程池

當所有任務(wù)都執(zhí)行完畢后，需要關(guān)閉線程池以釋放資源：

executorService.shutdown();

三、線程池的核心組件

1. ThreadPoolExecutor

ThreadPoolExecutor是ExecutorService接口的實現(xiàn)類，它提供了更多的功能，如定時執(zhí)行、定期執(zhí)行、異常處理等。我們通常需要自己實現(xiàn)一個ThreadPoolExecutor來滿足業(yè)務(wù)需求。

2. BlockingQueue

BlockingQueue是一個阻塞隊列，用于存放待處理的任務(wù)。它是一個FIFO（先進先出）的隊列，當隊列滿時，新來的任務(wù)會等待；當隊列為空時，正在執(zhí)行的任務(wù)會等待。常用的阻塞隊列有ArrayBlockingQueue、LinkedBlockingQueue和SynchronousQueue。

3. 拒絕策略

當線程池無法處理新提交的任務(wù)時，需要采取一定的策略來處理這些任務(wù)。Java中的RejectedExecutionHandler接口提供了四種默認的拒絕策略：

1. CallerRunsPolicy：直接在調(diào)用者線程中運行任務(wù)。
2. AbortPolicy：拋出一個未檢查異常中斷任務(wù)。
3. DiscardPolicy：丟棄任務(wù)，不做任何處理。
4. DiscardOldestPolicy：丟棄隊列中最舊的任務(wù)，嘗試重新提交新任務(wù)。

我們可以根據(jù)實際需求自定義拒絕策略。例如，下面是一個自定義的拒絕策略示例：

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;public class CustomRejectedExecutionHandler implements RejectedExecutionHandler {private static final int MAX_CAPACITY = 100; // 最大容量private AtomicInteger rejectedCount = new AtomicInteger(); // 被拒絕的任務(wù)計數(shù)器private ArrayBlockingQueue<Runnable> taskQueue; // 任務(wù)隊列public CustomRejectedExecutionHandler() {taskQueue = new ArrayBlockingQueue<>(MAX_CAPACITY); // 初始化任務(wù)隊列}@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // 拒絕策略實現(xiàn)方法if (taskQueue.remainingCapacity() == 0) { // 如果隊列已滿，添加拒絕策略executor.setRejectedExecutionHandler(new RejectedExecutionHandler() { // 設(shè)置新的拒絕策略@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // 拒絕策略實現(xiàn)方法（AbortPolicy）System.err.println("Task queue is full, rejecting task: " + r); // 輸出錯誤信息并拋出異常（AbortPolicy）throw new RejectedExecutionException("Task queue is full", r); // 拋出RejectedExecutionException異常（AbortPolicy）}});} else { // 如果隊列未滿，將任務(wù)添加到隊列中，并增加計數(shù)器（CallerRunsPolicy）或直接拋出異常（DiscardPolicy/DiscardOldestPolicy）taskQueue.put(r); // 將任務(wù)添加到隊列中（非阻塞）rejectedCount.incrementAndGet(); // 被拒絕的任務(wù)計數(shù)器加1（CallerRunsPolicy/DiscardPolicy/DiscardOldestPolicy）或直接拋出異常（DiscardPolicy/DiscardOldestPolicy）}}
}

四、自定義線程池示例

下面我們通過一個示例來說明如何自定義線程池：假設(shè)我們需要一個支持定時執(zhí)行和周期性執(zhí)行的任務(wù)線程池，我們可以這樣實現(xiàn)：
要實現(xiàn)一個支持定時執(zhí)行和周期性執(zhí)行的任務(wù)線程池，我們可以創(chuàng)建一個自定義的線程池類，繼承自ThreadPoolExecutor，并重寫其中的方法。以下是一個簡單的示例：

import java.util.concurrent.*;public class CustomThreadPoolExecutor extends ThreadPoolExecutor {private final long keepAliveTime;private final TimeUnit unit;public CustomThreadPoolExecutor(int corePoolSize, long keepAliveTime, TimeUnit unit) {super(corePoolSize);this.keepAliveTime = keepAliveTime;this.unit = unit;}@Overrideprotected void beforeExecute(Thread t, Runnable r) {super.beforeExecute(t, r);// 在任務(wù)執(zhí)行前執(zhí)行的操作，例如記錄日志、初始化資源等}@Overrideprotected void afterExecute(Runnable r, Throwable t) {super.afterExecute(r, t);// 在任務(wù)執(zhí)行后執(zhí)行的操作，例如清理資源、記錄日志等}public void scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {super.scheduleAtFixedRate(command, initialDelay, period, unit);}public void scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {super.scheduleWithFixedDelay(command, initialDelay, delay, unit);}
}

在這個示例中，我們創(chuàng)建了一個名為CustomThreadPoolExecutor的自定義線程池類，它繼承自ThreadPoolExecutor。我們?yōu)檫@個類添加了兩個方法：scheduleAtFixedRate和scheduleWithFixedDelay，分別用于定時執(zhí)行和周期性執(zhí)行任務(wù)。這兩個方法內(nèi)部調(diào)用了父類的相應(yīng)方法來實現(xiàn)任務(wù)調(diào)度。

使用這個自定義線程池的示例代碼如下：

import java.util.concurrent.*;public class CustomThreadPoolExample {public static void main(String[] args) throws InterruptedException {CustomThreadPoolExecutor threadPool = new CustomThreadPoolExecutor(2, 10, TimeUnit.SECONDS);for (int i = 0; i < 5; i++) {final int taskId = i;threadPool.execute(() -> {System.out.println("Task " + taskId + " is running");try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("Task " + taskId + " is completed");});}// 關(guān)閉線程池（所有任務(wù)執(zhí)行完成后，線程池不再接受新任務(wù)）threadPool.shutdown();}
}

在這個示例中，我們創(chuàng)建了一個定時執(zhí)行任務(wù)的線程池，核心線程數(shù)為2，任務(wù)間隔時間為10秒。然后，我們提交了5個任務(wù)到線程池，每個任務(wù)打印一條開始和結(jié)束信息，并在運行過程中暫停1秒。由于線程池具有定時執(zhí)行功能，因此這些任務(wù)會按照設(shè)定的時間間隔依次執(zhí)行。最后，我們在所有任務(wù)執(zhí)行完成后關(guān)閉線程池。

在Java中，我們可以通過java.util.concurrent.Executors類創(chuàng)建線程池。然而，有時候我們需要創(chuàng)建一個具有特定功能的線程池，例如定時執(zhí)行任務(wù)、周期性執(zhí)行任務(wù)等。這時，我們可以創(chuàng)建一個自定義的線程池來實現(xiàn)這些功能。本文將介紹如何創(chuàng)建一個自定義線程池，并通過一個示例來演示其使用方法。

1. 創(chuàng)建自定義線程池

要創(chuàng)建一個自定義線程池，我們需要實現(xiàn)java.util.concurrent.ThreadPoolExecutor接口，并重寫其中的方法。以下是一個簡單的自定義線程池實現(xiàn)：

import java.util.concurrent.*;public class CustomThreadPoolExecutor extends ThreadPoolExecutor {private final int corePoolSize;private final BlockingQueue<Runnable> workQueue;private final long keepAliveTime;private final TimeUnit unit;public CustomThreadPoolExecutor(int corePoolSize, BlockingQueue<Runnable> workQueue, long keepAliveTime, TimeUnit unit) {this.corePoolSize = corePoolSize;this.workQueue = workQueue;this.keepAliveTime = keepAliveTime;this.unit = unit;}@Overrideprotected void beforeExecute(Thread t, Runnable r) {super.beforeExecute(t, r);// 在任務(wù)執(zhí)行前執(zhí)行的操作，例如記錄日志、初始化資源等}@Overrideprotected void afterExecute(Runnable r, Throwable t) {super.afterExecute(r, t);// 在任務(wù)執(zhí)行后執(zhí)行的操作，例如清理資源、記錄日志等}
}

2. 使用自定義線程池

創(chuàng)建好自定義線程池后，我們可以像使用普通的ThreadPoolExecutor一樣使用它。以下是一個使用自定義線程池的示例：

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;public class CustomThreadPoolExample {public static void main(String[] args) {// 創(chuàng)建一個定時執(zhí)行任務(wù)的線程池CustomThreadPoolExecutor threadPool = new CustomThreadPoolExecutor(2, new LinkedBlockingQueue<>(2), 30, TimeUnit.SECONDS);// 提交任務(wù)到線程池for (int i = 0; i < 5; i++) {final int taskId = i;threadPool.execute(() -> {System.out.println("Task " + taskId + " is running");try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("Task " + taskId + " is completed");});}// 關(guān)閉線程池（所有任務(wù)執(zhí)行完成后，線程池不再接受新任務(wù)）threadPool.shutdown();}
}

在這個示例中，我們創(chuàng)建了一個定時執(zhí)行任務(wù)的線程池，最大容量為2，任務(wù)隊列大小為2，空閑線程的存活時間為30秒。然后，我們提交了5個任務(wù)到線程池，每個任務(wù)打印一條開始和結(jié)束信息，并在運行過程中暫停1秒。由于線程池具有定時執(zhí)行功能，因此這些任務(wù)會按照設(shè)定的時間間隔依次執(zhí)行。最后，我們在所有任務(wù)執(zhí)行完成后關(guān)閉線程池。

往期專欄
Java全棧開發(fā)
數(shù)據(jù)結(jié)構(gòu)與算法
計算機組成原理
操作系統(tǒng)
數(shù)據(jù)庫系統(tǒng)
物聯(lián)網(wǎng)控制原理與技術(shù)