接受失敗：“失敗是什么？沒有什么，只是更走近成功一步；成功是什么？就是走過了所有通向失敗的路，只剩下一條路，那就是成功的路?！边@句話很好地詮釋了如何看待失敗的問題，即每一次跌倒都是通往勝利道路上不可或缺的一部分。

創(chuàng)造機(jī)會：“不要等待機(jī)會，而要創(chuàng)造機(jī)會?！边@句話鼓勵人們主動出擊，不要總是期待外界給予完美的條件或時(shí)機(jī)，而是要勇于嘗試，創(chuàng)造屬于自己的機(jī)遇。

---

目錄

上一篇博客課后習(xí)題

編寫一個簡單的生產(chǎn)者-消費(fèi)者模式程序

實(shí)現(xiàn)一個多階段流水線作業(yè)

設(shè)計(jì)一個支持動態(tài)調(diào)整線程數(shù)量的線程池框架

分析并修復(fù)一段包含潛在死鎖風(fēng)險(xiǎn)的代碼片段

探索LockSupport.park()和unpark()的用法

第4章 線程池入門

4.1 ThreadPoolExecutor

4.1.1 創(chuàng)建線程池

4.1.2 關(guān)閉線程池

4.2 Executor接口

4.3 ExecutorService 接口

4.3.1 Callable返回任務(wù)執(zhí)行結(jié)果

4.3.2 .shutdown與shutdownNow

4.4 Executors 工具箱

4.5 線程工廠與線程組

4.6 線程池異常處理

4.7 本章習(xí)題

---

上一篇博客課后習(xí)題

編寫一個簡單的生產(chǎn)者-消費(fèi)者模式程序

根據(jù)提供的資料，我們可以創(chuàng)建一個使用java.util.concurrent包中工具來實(shí)現(xiàn)生產(chǎn)者-消費(fèi)者模式的例子。這里我們將采用BlockingQueue接口的一個具體實(shí)現(xiàn)如ArrayBlockingQueue作為緩沖區(qū)。該隊(duì)列允許我們利用其內(nèi)置的方法put()和take()來進(jìn)行阻塞操作，確保當(dāng)隊(duì)列滿時(shí)生產(chǎn)者會被阻塞，而當(dāng)隊(duì)列為空時(shí)消費(fèi)者也會被阻塞等待元素出現(xiàn)。下面是一個簡化的代碼示例：

import java.util.concurrent.*;public class ProducerConsumerExample {private static final int BUFFER_SIZE = 10;private static final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(BUFFER_SIZE);public static void main(String[] args) throws InterruptedException {Thread producer = new Thread(() -> {try {for (int i = 0; ; i++) {System.out.println("Producing item: " + i);queue.put(i); // 如果隊(duì)列已滿，則此方法會自動阻塞直到有空間可用System.out.println("Buffer size after production: " + queue.size());Thread.sleep(100); // 模擬生產(chǎn)時(shí)間}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});Thread consumer = new Thread(() -> {try {while (true) {Integer item = queue.take(); // 如果隊(duì)列為空，則此方法會自動阻塞直到有元素可取System.out.println("Consuming item: " + item);System.out.println("Buffer size after consumption: " + queue.size());Thread.sleep(150); // 模擬消費(fèi)時(shí)間}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});producer.start();consumer.start();producer.join();consumer.join();}
}

實(shí)現(xiàn)一個多階段流水線作業(yè)

對于多階段流水線作業(yè)，可以考慮使用java.util.concurrent.Phaser或自定義信號量機(jī)制來同步各個階段之間的執(zhí)行。每個階段的任務(wù)完成之后通知下一個階段開始工作。下面給出的是一個簡化版本，其中每個階段都是通過獨(dú)立線程完成，并且通過共享變量傳遞數(shù)據(jù)給下一級別處理。

class PipelineStage implements Runnable {private final Phaser phaser;private final String name;private final Object input;private final Consumer<Object> output;public PipelineStage(Phaser phaser, String name, Object input, Consumer<Object> output) {this.phaser = phaser;this.name = name;this.input = input;this.output = output;}@Overridepublic void run() {phaser.arriveAndAwaitAdvance(); // 等待所有前驅(qū)階段完成System.out.println(name + " processing...");// 模擬處理過程...try { Thread.sleep((long)(Math.random() * 500)); } catch (InterruptedException e) {}if (output != null) output.accept(input); // 將結(jié)果傳遞給下一階段phaser.arriveAndDeregister(); // 完成本階段后注銷}
}// 測試用例
public class MultiStagePipeline {public static void main(String[] args) throws InterruptedException {Phaser phaser = new Phaser();phaser.register(); // 注冊主線程// 創(chuàng)建多個階段并啟動它們PipelineStage stage1 = new PipelineStage(phaser, "Stage 1", "Input Data", null);PipelineStage stage2 = new PipelineStage(phaser, "Stage 2", stage1, null);PipelineStage stage3 = new PipelineStage(phaser, "Stage 3", stage2, result -> System.out.println("Final Output: " + result));new Thread(stage1).start();new Thread(stage2).start();new Thread(stage3).start();phaser.arriveAndAwaitAdvance(); // 等待所有階段完成}
}

設(shè)計(jì)一個支持動態(tài)調(diào)整線程數(shù)量的線程池框架

要構(gòu)建這樣一個線程池，我們需要考慮如何監(jiān)控當(dāng)前的工作負(fù)載并根據(jù)需要調(diào)整線程的數(shù)量。可以通過監(jiān)聽器模式或者輪詢的方式來檢測任務(wù)隊(duì)列的狀態(tài)，并相應(yīng)地增加或減少線程數(shù)以優(yōu)化資源利用率。此外，還需要提供API讓用戶能夠配置核心線程數(shù)、最大線程數(shù)及隊(duì)列容量等參數(shù)，并允許這些設(shè)置在運(yùn)行時(shí)動態(tài)更改。

import java.util.concurrent.*;
import java.util.function.Consumer;public class DynamicThreadPoolExecutor extends ThreadPoolExecutor {private final Consumer<DynamicThreadPoolExecutor> configChangeListener;public DynamicThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue, Consumer<DynamicThreadPoolExecutor> listener) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);this.configChangeListener = listener;}public synchronized void setCorePoolSize(int corePoolSize) {super.setCorePoolSize(corePoolSize);notifyConfigChange();}public synchronized void setMaximumPoolSize(int maximumPoolSize) {super.setMaximumPoolSize(maximumPoolSize);notifyConfigChange();}private void notifyConfigChange() {if (configChangeListener != null) {configChangeListener.accept(this);}}// 其他必要的方法...
}

分析并修復(fù)一段包含潛在死鎖風(fēng)險(xiǎn)的代碼片段

假設(shè)有一段代碼如下所示，它展示了兩個線程分別試圖獲取對方持有的鎖，從而可能導(dǎo)致死鎖的情況。

public class DeadlockExample {private static final Object lock1 = new Object();private static final Object lock2 = new Object();public static void main(String[] args) {Thread t1 = new Thread(() -> {synchronized (lock1) {System.out.println("Thread 1: Holding lock 1...");try { Thread.sleep(100); } catch (InterruptedException e) {}synchronized (lock2) {System.out.println("Thread 1: Holding lock 1 & 2...");}}});Thread t2 = new Thread(() -> {synchronized (lock2) {System.out.println("Thread 2: Holding lock 2...");try { Thread.sleep(100); } catch (InterruptedException e) {}synchronized (lock1) {System.out.println("Thread 2: Holding lock 2 & 1...");}}});t1.start();t2.start();}
}

為了避免這種情況的發(fā)生，我們應(yīng)該遵循一定的規(guī)則來保證所有的鎖請求都按照相同的順序進(jìn)行。例如，在上面的例子中，我們可以修改代碼使得兩個線程總是先嘗試獲得lock1再嘗試獲得lock2。這樣做可以防止形成循環(huán)等待鏈，進(jìn)而避免死鎖的發(fā)生。

探索LockSupport.park()和unpark()的用法

最后，關(guān)于LockSupport.park()和unpark()，這兩個函數(shù)提供了非常低級別的線程阻塞/喚醒功能。它們不是基于對象監(jiān)視器（monitor）而是直接作用于線程本身。這意味著你可以更靈活地控制哪個特定的線程應(yīng)該被阻塞或喚醒。通常情況下，unpark(Thread thread)會在park()之前調(diào)用來“預(yù)存”一次許可，這樣當(dāng)目標(biāo)線程調(diào)用park()時(shí)可以直接繼續(xù)執(zhí)行而不必等待其他線程顯式地調(diào)用unpark()。此外，值得注意的是，即使多次調(diào)用了unpark()，只要有一次對應(yīng)的park()調(diào)用就會消耗掉這個許可；如果再次調(diào)用park()而沒有新的unpark()調(diào)用，則線程將重新進(jìn)入阻塞狀態(tài)。

第4章 線程池入門

4.1 ThreadPoolExecutor

4.1.1 創(chuàng)建線程池

在Java中，ThreadPoolExecutor是實(shí)現(xiàn)線程池的核心類之一。它允許開發(fā)者自定義線程池的行為，包括但不限于核心線程數(shù)、最大線程數(shù)、空閑線程存活時(shí)間等參數(shù)。通過這種方式，可以創(chuàng)建更加靈活且適合特定應(yīng)用場景的線程池。

代碼示例：

import java.util.concurrent.*;public class CustomThreadPoolExample {public static void main(String[] args) {// 定義一個簡單的任務(wù)類class Task implements Runnable {private final String name;public Task(String name) {this.name = name;}@Overridepublic void run() {System.out.println("Executing task " + name);try {Thread.sleep(2000); // 模擬長時(shí)間運(yùn)行的任務(wù)} catch (InterruptedException e) {// 如果線程被中斷，則恢復(fù)中斷狀態(tài)Thread.currentThread().interrupt();System.err.println("Task " + name + " was interrupted.");}System.out.println("Task " + name + " completed.");}}// 創(chuàng)建自定義線程池int corePoolSize = 2; // 核心線程數(shù)int maximumPoolSize = 4; // 最大線程數(shù)long keepAliveTime = 5000; // 空閑線程存活時(shí)間（毫秒）TimeUnit unit = TimeUnit.MILLISECONDS;BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(10); // 任務(wù)隊(duì)列容量為10ThreadFactory threadFactory = Executors.defaultThreadFactory(); // 使用默認(rèn)線程工廠RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy(); // 拒絕策略ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);// 提交多個任務(wù)給線程池執(zhí)行for (int i = 1; i <= 6; i++) {executor.submit(new Task("Task-" + i));}// 關(guān)閉線程池...executor.shutdown();// 嘗試優(yōu)雅地關(guān)閉線程池，等待所有任務(wù)完成try {if (!executor.awaitTermination(800, TimeUnit.MILLISECONDS)) {System.out.println("Not all tasks have been completed within the timeout period.");}} catch (InterruptedException e) {Thread.currentThread().interrupt();}}
}

這段代碼展示了如何使用ThreadPoolExecutor來創(chuàng)建一個具有特定配置的線程池，并向其中提交若干個任務(wù)進(jìn)行異步處理。注意這里選擇了LinkedBlockingQueue作為任務(wù)隊(duì)列，并設(shè)置了合理的大小以避免無限制的增長；同時(shí)指定了CallerRunsPolicy作為拒絕策略，這意味著當(dāng)線程池?zé)o法接受新任務(wù)時(shí)，調(diào)用者所在的線程將會執(zhí)行該任務(wù)。

4.1.2 關(guān)閉線程池

正確地關(guān)閉線程池非常重要，因?yàn)樗粌H涉及到資源的有效釋放，還可能影響到應(yīng)用程序的整體性能和穩(wěn)定性。根據(jù)需求的不同，可以選擇調(diào)用shutdown()或shutdownNow()方法來停止線程池的工作。

• shutdown()：此方法會等待所有已提交的任務(wù)完成之后再終止線程池，但不再接受新的任務(wù)提交。
• shutdownNow()：嘗試立即停止所有正在執(zhí)行中的任務(wù)，并返回未開始執(zhí)行的任務(wù)列表。需要注意的是，這并不保證所有活動線程都能被成功中斷。

此外，還可以結(jié)合awaitTermination()一起使用，以便在線程池完全結(jié)束前讓程序等待一段時(shí)間。如果超出了指定的時(shí)間而線程池仍未關(guān)閉，則可以通過邏輯判斷是否需要采取進(jìn)一步措施。

代碼補(bǔ)充：

// 在main方法末尾添加以下代碼，確保線程池能夠正常關(guān)閉
if (!executor.isTerminated()) {System.out.println("Shutting down Executor Service");
}// 或者使用更激進(jìn)的方式，立即嘗試關(guān)閉所有任務(wù)
List<Runnable> unfinishedTasks = executor.shutdownNow();
if (!unfinishedTasks.isEmpty()) {System.out.println("Unfinished tasks: " + unfinishedTasks.size());
}

4.2 Executor接口

Executor接口是最基礎(chǔ)的執(zhí)行器接口，它提供了一個簡化了多線程編程的方式——將任務(wù)提交與任務(wù)執(zhí)行分離出來。具體來說，它僅包含一個名為execute(Runnable command)的方法，用于接收實(shí)現(xiàn)了Runnable接口的對象，并安排它們在一個合適的線程上運(yùn)行。

4.3 ExecutorService 接口

4.3.1 Callable返回任務(wù)執(zhí)行結(jié)果

ExecutorService擴(kuò)展了Executor接口的功能，增加了對Callable的支持，使得我們可以從非阻塞的任務(wù)中獲取返回值。Callable類似于Runnable，但它允許拋出受檢異常并且支持返回類型。為了得到Callable的結(jié)果，通常我們會使用submit()方法提交任務(wù)，并通過返回的Future對象查詢結(jié)果或取消任務(wù)。

代碼示例：

import java.util.concurrent.*;public class CallableExample {public static void main(String[] args) throws InterruptedException, ExecutionException {// 創(chuàng)建一個固定大小的線程池ExecutorService executor = Executors.newFixedThreadPool(2);// 定義一個有返回值的任務(wù)Callable<String> task = () -> {Thread.sleep(1000); // 模擬任務(wù)執(zhí)行時(shí)間return "Hello from Callable!";};// 提交任務(wù)并獲取Future對象Future<String> future = executor.submit(task);// 獲取任務(wù)執(zhí)行結(jié)果String result = future.get(); // 此處會阻塞直到任務(wù)完成System.out.println(result);// 關(guān)閉線程池executor.shutdown();}
}

4.3.2 .shutdown與shutdownNow

如前所述，shutdown()方法可以讓線程池進(jìn)入“優(yōu)雅關(guān)閉”狀態(tài)，即不再接受新的任務(wù)，但是會繼續(xù)處理已經(jīng)提交的任務(wù)直到它們?nèi)客瓿?。相比之?#xff0c;shutdownNow()則試圖立即停止所有正在進(jìn)行的任務(wù)，并清理掉尚未啟動的任務(wù)。

4.4 Executors 工具箱

Executors是一個靜態(tài)工具類，提供了幾種常用的線程池創(chuàng)建方法。這些方法簡化了線程池的初始化過程，讓用戶不必每次都手動設(shè)置復(fù)雜的構(gòu)造參數(shù)。常見的有：

• newCachedThreadPool()：創(chuàng)建一個可根據(jù)需要動態(tài)調(diào)整大小的線程池，適用于執(zhí)行大量短期生存期的任務(wù)。
• newFixedThreadPool(int nThreads)：創(chuàng)建一個固定大小的線程池，適合于控制并發(fā)級別的情況。
• newSingleThreadExecutor()：創(chuàng)建只有一個工作線程的線程池，確保所有任務(wù)按照順序依次執(zhí)行。
• newScheduledThreadPool(int corePoolSize)：創(chuàng)建一個支持定時(shí)調(diào)度功能的線程池。
• newWorkStealingPool(int parallelism)：創(chuàng)建一個工作竊取線程池，旨在提高多核處理器上的任務(wù)并行度。

代碼示例：

import java.util.concurrent.*;public class ExecutorsExample {public static void main(String[] args) {// 創(chuàng)建不同類型的線程池ExecutorService cachedPool = Executors.newCachedThreadPool();ExecutorService fixedPool = Executors.newFixedThreadPool(3);ExecutorService singlePool = Executors.newSingleThreadExecutor();ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(2);ExecutorService workStealingPool = Executors.newWorkStealingPool();// 提交任務(wù)給不同的線程池Runnable task = () -> System.out.println("Task executed by " + Thread.currentThread().getName());cachedPool.submit(task);fixedPool.submit(task);singlePool.submit(task);scheduledPool.schedule(task, 2, TimeUnit.SECONDS);workStealingPool.submit(task);// 關(guān)閉線程池cachedPool.shutdown();fixedPool.shutdown();singlePool.shutdown();scheduledPool.shutdown();workStealingPool.shutdown();}
}

4.5 線程工廠與線程組

這部分內(nèi)容涉及到了更底層的線程管理機(jī)制。線程工廠(ThreadFactory)接口允許我們自定義線程的創(chuàng)建方式，比如命名規(guī)則、優(yōu)先級設(shè)定等。而線程組(ThreadGroup)則是用來組織相關(guān)線程的一種結(jié)構(gòu)，在某些情況下可以幫助更好地管理和監(jiān)控線程集合。

代碼示例：

import java.util.concurrent.*;public class ThreadFactoryExample {public static void main(String[] args) {// 自定義線程工廠ThreadFactory customThreadFactory = runnable -> {Thread thread = new Thread(runnable, "CustomThread-" + Thread.activeCount());thread.setDaemon(true); // 設(shè)置為守護(hù)線程thread.setPriority(Thread.MAX_PRIORITY); // 設(shè)置最高優(yōu)先級return thread;};// 使用自定義線程工廠創(chuàng)建線程池ExecutorService executor = Executors.newFixedThreadPool(2, customThreadFactory);// 提交任務(wù)Runnable task = () -> System.out.println("Running in " + Thread.currentThread().getName());executor.submit(task);// 關(guān)閉線程池executor.shutdown();}
}

4.6 線程池異常處理

當(dāng)線程池中的任務(wù)拋出未捕獲的異常時(shí)，默認(rèn)情況下JVM會打印堆棧跟蹤信息并將線程標(biāo)記為失敗狀態(tài)。為了避免這種情況，可以為每個線程設(shè)置一個UncaughtExceptionHandler，或者直接利用Future.get()捕獲由Callable產(chǎn)生的異常。

代碼示例：

import java.util.concurrent.*;public class ExceptionHandlingExample {public static void main(String[] args) {// 創(chuàng)建線程池并設(shè)置未捕獲異常處理器ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);executor.setUncaughtExceptionHandler((t, e) -> System.err.println("Thread " + t.getName() + " threw exception: " + e.getMessage()));// 提交可能會拋出異常的任務(wù)executor.submit(() -> {throw new RuntimeException("Something went wrong!");});// 使用Future捕獲Callable產(chǎn)生的異常Future<?> future = executor.submit(() -> {throw new Exception("Another issue occurred.");});try {future.get(); // 此處會拋出ExecutionException包裝原始異常} catch (InterruptedException | ExecutionException e) {System.err.println("Caught exception from Future: " + e.getCause().getMessage());}// 關(guān)閉線程池executor.shutdown();}
}

4.7 本章習(xí)題

請參照章節(jié)開頭給出的例子和說明，嘗試編寫一些簡單的練習(xí)來加深理解，例如實(shí)現(xiàn)生產(chǎn)者-消費(fèi)者模式、構(gòu)建多階段流水線作業(yè)等。此外，也可以探索更多關(guān)于LockSupport.park()/unpark()的知識點(diǎn)，了解它們是如何幫助實(shí)現(xiàn)低級別的線程同步操作的。