Java多线程与线程池技术详解（四）

接受失败：“失败是什么？没有什么，只是更走近成功一步；成功是什么？就是走过了所有通向失败的路，只剩下一条路，那就是成功的路。”这句话很好地诠释了如何看待失败的问题，即每一次跌倒都是通往胜利道路上不可或缺的一部分。

创造机会：“不要等待机会，而要创造机会。”这句话鼓励人们主动出击，不要总是期待外界给予完美的条件或时机，而是要勇于尝试，创造属于自己的机遇。

---

目录

上一篇博客课后习题

编写一个简单的生产者-消费者模式程序

实现一个多阶段流水线作业

设计一个支持动态调整线程数量的线程池框架

分析并修复一段包含潜在死锁风险的代码片段

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

第4章 线程池入门

4.1 ThreadPoolExecutor

4.1.1 创建线程池

4.1.2 关闭线程池

4.2 Executor接口

4.3 ExecutorService 接口

4.3.1 Callable返回任务执行结果

4.3.2 .shutdown与shutdownNow

4.4 Executors 工具箱

4.5 线程工厂与线程组

4.6 线程池异常处理

4.7 本章习题

---

上一篇博客课后习题

编写一个简单的生产者-消费者模式程序

根据提供的资料，我们可以创建一个使用java.util.concurrent包中工具来实现生产者-消费者模式的例子。这里我们将采用BlockingQueue接口的一个具体实现如ArrayBlockingQueue作为缓冲区。该队列允许我们利用其内置的方法put()和take()来进行阻塞操作，确保当队列满时生产者会被阻塞，而当队列为空时消费者也会被阻塞等待元素出现。下面是一个简化的代码示例：

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); // 如果队列已满，则此方法会自动阻塞直到有空间可用System.out.println("Buffer size after production: " + queue.size());Thread.sleep(100); // 模拟生产时间}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});Thread consumer = new Thread(() -> {try {while (true) {Integer item = queue.take(); // 如果队列为空，则此方法会自动阻塞直到有元素可取System.out.println("Consuming item: " + item);System.out.println("Buffer size after consumption: " + queue.size());Thread.sleep(150); // 模拟消费时间}} catch (InterruptedException e) {Thread.currentThread().interrupt();}});producer.start();consumer.start();producer.join();consumer.join();}
}

实现一个多阶段流水线作业

对于多阶段流水线作业，可以考虑使用java.util.concurrent.Phaser或自定义信号量机制来同步各个阶段之间的执行。每个阶段的任务完成之后通知下一个阶段开始工作。下面给出的是一个简化版本，其中每个阶段都是通过独立线程完成，并且通过共享变量传递数据给下一级别处理。

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(); // 等待所有前驱阶段完成System.out.println(name + " processing...");// 模拟处理过程...try { Thread.sleep((long)(Math.random() * 500)); } catch (InterruptedException e) {}if (output != null) output.accept(input); // 将结果传递给下一阶段phaser.arriveAndDeregister(); // 完成本阶段后注销}
}// 测试用例
public class MultiStagePipeline {public static void main(String[] args) throws InterruptedException {Phaser phaser = new Phaser();phaser.register(); // 注册主线程// 创建多个阶段并启动它们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(); // 等待所有阶段完成}
}

设计一个支持动态调整线程数量的线程池框架

要构建这样一个线程池，我们需要考虑如何监控当前的工作负载并根据需要调整线程的数量。可以通过监听器模式或者轮询的方式来检测任务队列的状态，并相应地增加或减少线程数以优化资源利用率。此外，还需要提供API让用户能够配置核心线程数、最大线程数及队列容量等参数，并允许这些设置在运行时动态更改。

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);}}// 其他必要的方法...
}

分析并修复一段包含潜在死锁风险的代码片段

假设有一段代码如下所示，它展示了两个线程分别试图获取对方持有的锁，从而可能导致死锁的情况。

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();}
}

为了避免这种情况的发生，我们应该遵循一定的规则来保证所有的锁请求都按照相同的顺序进行。例如，在上面的例子中，我们可以修改代码使得两个线程总是先尝试获得lock1再尝试获得lock2。这样做可以防止形成循环等待链，进而避免死锁的发生。

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

最后，关于LockSupport.park()和unpark()，这两个函数提供了非常低级别的线程阻塞/唤醒功能。它们不是基于对象监视器（monitor）而是直接作用于线程本身。这意味着你可以更灵活地控制哪个特定的线程应该被阻塞或唤醒。通常情况下，unpark(Thread thread)会在park()之前调用来“预存”一次许可，这样当目标线程调用park()时可以直接继续执行而不必等待其他线程显式地调用unpark()。此外，值得注意的是，即使多次调用了unpark()，只要有一次对应的park()调用就会消耗掉这个许可；如果再次调用park()而没有新的unpark()调用，则线程将重新进入阻塞状态。

第4章 线程池入门

4.1 ThreadPoolExecutor

4.1.1 创建线程池

在Java中，ThreadPoolExecutor是实现线程池的核心类之一。它允许开发者自定义线程池的行为，包括但不限于核心线程数、最大线程数、空闲线程存活时间等参数。通过这种方式，可以创建更加灵活且适合特定应用场景的线程池。

代码示例：

import java.util.concurrent.*;public class CustomThreadPoolExample {public static void main(String[] args) {// 定义一个简单的任务类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); // 模拟长时间运行的任务} catch (InterruptedException e) {// 如果线程被中断，则恢复中断状态Thread.currentThread().interrupt();System.err.println("Task " + name + " was interrupted.");}System.out.println("Task " + name + " completed.");}}// 创建自定义线程池int corePoolSize = 2; // 核心线程数int maximumPoolSize = 4; // 最大线程数long keepAliveTime = 5000; // 空闲线程存活时间（毫秒）TimeUnit unit = TimeUnit.MILLISECONDS;BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(10); // 任务队列容量为10ThreadFactory threadFactory = Executors.defaultThreadFactory(); // 使用默认线程工厂RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy(); // 拒绝策略ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);// 提交多个任务给线程池执行for (int i = 1; i <= 6; i++) {executor.submit(new Task("Task-" + i));}// 关闭线程池...executor.shutdown();// 尝试优雅地关闭线程池，等待所有任务完成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来创建一个具有特定配置的线程池，并向其中提交若干个任务进行异步处理。注意这里选择了LinkedBlockingQueue作为任务队列，并设置了合理的大小以避免无限制的增长；同时指定了CallerRunsPolicy作为拒绝策略，这意味着当线程池无法接受新任务时，调用者所在的线程将会执行该任务。

4.1.2 关闭线程池

正确地关闭线程池非常重要，因为它不仅涉及到资源的有效释放，还可能影响到应用程序的整体性能和稳定性。根据需求的不同，可以选择调用shutdown()或shutdownNow()方法来停止线程池的工作。

• shutdown()：此方法会等待所有已提交的任务完成之后再终止线程池，但不再接受新的任务提交。
• shutdownNow()：尝试立即停止所有正在执行中的任务，并返回未开始执行的任务列表。需要注意的是，这并不保证所有活动线程都能被成功中断。

此外，还可以结合awaitTermination()一起使用，以便在线程池完全结束前让程序等待一段时间。如果超出了指定的时间而线程池仍未关闭，则可以通过逻辑判断是否需要采取进一步措施。

代码补充：

// 在main方法末尾添加以下代码，确保线程池能够正常关闭
if (!executor.isTerminated()) {System.out.println("Shutting down Executor Service");
}// 或者使用更激进的方式，立即尝试关闭所有任务
List<Runnable> unfinishedTasks = executor.shutdownNow();
if (!unfinishedTasks.isEmpty()) {System.out.println("Unfinished tasks: " + unfinishedTasks.size());
}

4.2 Executor接口

Executor接口是最基础的执行器接口，它提供了一个简化了多线程编程的方式——将任务提交与任务执行分离出来。具体来说，它仅包含一个名为execute(Runnable command)的方法，用于接收实现了Runnable接口的对象，并安排它们在一个合适的线程上运行。

4.3 ExecutorService 接口

4.3.1 Callable返回任务执行结果

ExecutorService扩展了Executor接口的功能，增加了对Callable的支持，使得我们可以从非阻塞的任务中获取返回值。Callable类似于Runnable，但它允许抛出受检异常并且支持返回类型。为了得到Callable的结果，通常我们会使用submit()方法提交任务，并通过返回的Future对象查询结果或取消任务。

代码示例：

import java.util.concurrent.*;public class CallableExample {public static void main(String[] args) throws InterruptedException, ExecutionException {// 创建一个固定大小的线程池ExecutorService executor = Executors.newFixedThreadPool(2);// 定义一个有返回值的任务Callable<String> task = () -> {Thread.sleep(1000); // 模拟任务执行时间return "Hello from Callable!";};// 提交任务并获取Future对象Future<String> future = executor.submit(task);// 获取任务执行结果String result = future.get(); // 此处会阻塞直到任务完成System.out.println(result);// 关闭线程池executor.shutdown();}
}

4.3.2 .shutdown与shutdownNow

如前所述，shutdown()方法可以让线程池进入“优雅关闭”状态，即不再接受新的任务，但是会继续处理已经提交的任务直到它们全部完成。相比之下，shutdownNow()则试图立即停止所有正在进行的任务，并清理掉尚未启动的任务。

4.4 Executors 工具箱

Executors是一个静态工具类，提供了几种常用的线程池创建方法。这些方法简化了线程池的初始化过程，让用户不必每次都手动设置复杂的构造参数。常见的有：

• newCachedThreadPool()：创建一个可根据需要动态调整大小的线程池，适用于执行大量短期生存期的任务。
• newFixedThreadPool(int nThreads)：创建一个固定大小的线程池，适合于控制并发级别的情况。
• newSingleThreadExecutor()：创建只有一个工作线程的线程池，确保所有任务按照顺序依次执行。
• newScheduledThreadPool(int corePoolSize)：创建一个支持定时调度功能的线程池。
• newWorkStealingPool(int parallelism)：创建一个工作窃取线程池，旨在提高多核处理器上的任务并行度。

代码示例：

import java.util.concurrent.*;public class ExecutorsExample {public static void main(String[] args) {// 创建不同类型的线程池ExecutorService cachedPool = Executors.newCachedThreadPool();ExecutorService fixedPool = Executors.newFixedThreadPool(3);ExecutorService singlePool = Executors.newSingleThreadExecutor();ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(2);ExecutorService workStealingPool = Executors.newWorkStealingPool();// 提交任务给不同的线程池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);// 关闭线程池cachedPool.shutdown();fixedPool.shutdown();singlePool.shutdown();scheduledPool.shutdown();workStealingPool.shutdown();}
}

4.5 线程工厂与线程组

这部分内容涉及到了更底层的线程管理机制。线程工厂(ThreadFactory)接口允许我们自定义线程的创建方式，比如命名规则、优先级设定等。而线程组(ThreadGroup)则是用来组织相关线程的一种结构，在某些情况下可以帮助更好地管理和监控线程集合。

代码示例：

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); // 设置为守护线程thread.setPriority(Thread.MAX_PRIORITY); // 设置最高优先级return thread;};// 使用自定义线程工厂创建线程池ExecutorService executor = Executors.newFixedThreadPool(2, customThreadFactory);// 提交任务Runnable task = () -> System.out.println("Running in " + Thread.currentThread().getName());executor.submit(task);// 关闭线程池executor.shutdown();}
}

4.6 线程池异常处理

当线程池中的任务抛出未捕获的异常时，默认情况下JVM会打印堆栈跟踪信息并将线程标记为失败状态。为了避免这种情况，可以为每个线程设置一个UncaughtExceptionHandler，或者直接利用Future.get()捕获由Callable产生的异常。

代码示例：

import java.util.concurrent.*;public class ExceptionHandlingExample {public static void main(String[] args) {// 创建线程池并设置未捕获异常处理器ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);executor.setUncaughtExceptionHandler((t, e) -> System.err.println("Thread " + t.getName() + " threw exception: " + e.getMessage()));// 提交可能会抛出异常的任务executor.submit(() -> {throw new RuntimeException("Something went wrong!");});// 使用Future捕获Callable产生的异常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());}// 关闭线程池executor.shutdown();}
}

4.7 本章习题

请参照章节开头给出的例子和说明，尝试编写一些简单的练习来加深理解，例如实现生产者-消费者模式、构建多阶段流水线作业等。此外，也可以探索更多关于LockSupport.park()/unpark()的知识点，了解它们是如何帮助实现低级别的线程同步操作的。