JAVA 项目如何优雅关闭线程池？钩子函数与 shutdownGracefully 实战

大家好，今天我们来聊聊一个在Java并发编程中非常重要，但又容易被忽视的问题：如何优雅地关闭线程池。在实际的项目开发中，不恰当的线程池关闭方式可能会导致数据丢失、任务执行中断，甚至引发系统崩溃。因此，掌握优雅关闭线程池的技巧至关重要。

为什么需要优雅关闭线程池？

直接粗暴地关闭线程池，例如使用 executorService.shutdownNow() 可能会导致以下问题：

任务丢失： 正在执行的任务会被强制中断，未完成的任务可能直接被丢弃。
数据不一致： 任务可能正在更新数据库或文件系统，强制中断会导致数据损坏。
资源泄露： 线程可能持有锁或其他资源，强制中断会导致资源无法释放。

优雅关闭线程池的目标是确保所有已提交的任务都能完成执行，且不会再接受新的任务，从而避免上述问题。

Java 线程池关闭的几种方式

Java 提供了几种关闭线程池的方法，每种方法都有不同的特性和适用场景：

方法	说明	适用场景
`shutdown()`	启动线程池的关闭序列，拒绝接受新的任务，但会等待所有已提交的任务完成执行。	希望所有任务都完成，且不再接受新任务的场景。
`shutdownNow()`	尝试停止所有正在执行的任务，暂停处理正在等待的任务，并返回等待执行的任务列表。	希望尽快关闭线程池，即使可能导致任务丢失的场景。
`awaitTermination()`	阻塞当前线程，直到线程池中的所有任务都执行完毕，或者达到指定的超时时间。可以配合 `shutdown()` 使用，确保线程池完全关闭。	需要确保线程池完全关闭后才能继续执行后续操作的场景。
自定义关闭逻辑	通过自定义 `RejectedExecutionHandler` 和 `ThreadFactory`，可以实现更精细的控制，例如记录被拒绝的任务，或者在线程退出时执行特定的清理操作。	需要更灵活地控制线程池的行为，例如记录被拒绝的任务，或者在线程退出时执行特定的清理操作。

钩子函数（Shutdown Hook）

在 JVM 关闭时，可以注册一些钩子函数来执行清理操作。这些钩子函数会在 JVM 退出前被调用，可以用来优雅地关闭线程池。

public class ShutdownHookExample {

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(10);

        // 模拟提交任务
        for (int i = 0; i < 20; i++) {
            final int taskNumber = i;
            executorService.submit(() -> {
                try {
                    System.out.println("Task " + taskNumber + " started.");
                    Thread.sleep(1000); // 模拟耗时操作
                    System.out.println("Task " + taskNumber + " finished.");
                } catch (InterruptedException e) {
                    System.out.println("Task " + taskNumber + " interrupted.");
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 添加关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("Shutdown hook started.");
            executorService.shutdown(); // 拒绝接受新的任务
            try {
                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) { // 等待任务完成，最多等待 60 秒
                    System.out.println("ExecutorService did not terminate in the specified time.");
                    List<Runnable> droppedTasks = executorService.shutdownNow(); // 尝试中断所有正在执行的任务
                    System.out.println("Dropped tasks: " + droppedTasks.size());
                }
            } catch (InterruptedException e) {
                System.out.println("Interrupted while waiting for termination.");
                Thread.currentThread().interrupt();
            }
            System.out.println("Shutdown hook finished.");
        }));

        // 模拟程序运行一段时间后退出
        try {
            Thread.sleep(5000); // 模拟程序运行 5 秒
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("Main thread exiting.");
    }
}

代码解释：

创建线程池： 使用 Executors.newFixedThreadPool(10) 创建一个固定大小为 10 的线程池。
提交任务： 提交 20 个任务到线程池，每个任务模拟一个耗时 1 秒的操作。
添加关闭钩子： 使用 Runtime.getRuntime().addShutdownHook() 添加一个关闭钩子。这个钩子会在 JVM 关闭前被调用。
关闭钩子逻辑：
- 调用 executorService.shutdown() 拒绝接受新的任务。
- 调用 executorService.awaitTermination(60, TimeUnit.SECONDS) 等待所有已提交的任务完成执行，最多等待 60 秒。
- 如果超时，则调用 executorService.shutdownNow() 尝试中断所有正在执行的任务。
- 打印被丢弃的任务数量。
模拟程序运行： 主线程休眠 5 秒，模拟程序运行一段时间后退出。

运行结果分析：

运行上面的代码，可以看到以下输出：

首先，线程池开始执行任务。
5 秒后，主线程退出。
JVM 开始关闭，关闭钩子被调用。
关闭钩子首先调用 executorService.shutdown() 拒绝接受新的任务。
然后，它等待所有已提交的任务完成执行。
如果所有任务在 60 秒内完成，则程序正常退出。
如果超时，则关闭钩子会调用 executorService.shutdownNow() 尝试中断所有正在执行的任务，并打印被丢弃的任务数量。

优雅关闭线程池的最佳实践：shutdownGracefully 方法

为了更好地封装线程池的关闭逻辑，可以创建一个 shutdownGracefully 方法，该方法接受一个线程池作为参数，并执行优雅关闭操作。

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

public class ThreadPoolUtils {

    public static void shutdownGracefully(ExecutorService executorService, int timeout, TimeUnit unit) {
        System.out.println("Attempting to shutdown executor.");
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(timeout, unit)) {
                System.out.println("ExecutorService did not terminate in the specified time.");
                List<Runnable> droppedTasks = executorService.shutdownNow();
                System.out.println("Dropped tasks: " + droppedTasks.size());
            }
        } catch (InterruptedException e) {
            System.out.println("Interrupted while waiting for termination.");
            executorService.shutdownNow();
            Thread.currentThread().interrupt();
        }
        System.out.println("Executor shutdown complete.");
    }
}

代码解释：

shutdownGracefully 方法： 接受一个 ExecutorService 对象、一个超时时间和一个时间单位作为参数。
executorService.shutdown()： 首先调用 shutdown() 方法，拒绝接受新的任务。
executorService.awaitTermination()： 然后调用 awaitTermination() 方法，等待所有已提交的任务完成执行。
超时处理： 如果 awaitTermination() 方法超时，则调用 shutdownNow() 方法尝试中断所有正在执行的任务。
异常处理： 如果在等待过程中发生 InterruptedException 异常，则也调用 shutdownNow() 方法，并重新设置中断标志。

如何使用 shutdownGracefully 方法？

public class GracefulShutdownExample {

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(10);

        // 提交任务
        for (int i = 0; i < 20; i++) {
            final int taskNumber = i;
            executorService.submit(() -> {
                try {
                    System.out.println("Task " + taskNumber + " started.");
                    Thread.sleep(1000); // 模拟耗时操作
                    System.out.println("Task " + taskNumber + " finished.");
                } catch (InterruptedException e) {
                    System.out.println("Task " + taskNumber + " interrupted.");
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 添加关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("Shutdown hook started.");
            ThreadPoolUtils.shutdownGracefully(executorService, 60, TimeUnit.SECONDS);
            System.out.println("Shutdown hook finished.");
        }));

        // 模拟程序运行一段时间后退出
        try {
            Thread.sleep(5000); // 模拟程序运行 5 秒
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("Main thread exiting.");
    }
}

代码解释：

与之前的示例类似，首先创建一个线程池并提交一些任务。
在关闭钩子中，调用 ThreadPoolUtils.shutdownGracefully() 方法来优雅地关闭线程池。

优势：

代码复用： shutdownGracefully 方法可以被多个地方调用，避免代码重复。
易于维护： 关闭逻辑集中在一个地方，方便维护和修改。
更清晰的逻辑： 代码结构更清晰，易于理解。

更高级的技巧：自定义 RejectedExecutionHandler

除了使用 shutdownGracefully 方法，还可以通过自定义 RejectedExecutionHandler 来更精细地控制线程池的行为。RejectedExecutionHandler 接口定义了当线程池无法接受新任务时应该执行的操作。

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class CustomRejectedExecutionHandler implements RejectedExecutionHandler {

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println("Task " + r.toString() + " rejected from " + executor.toString());
        // 可以选择记录日志、重试任务或者执行其他自定义操作
    }
}

如何使用自定义 RejectedExecutionHandler？

在创建线程池时，可以将自定义的 RejectedExecutionHandler 传递给 ThreadPoolExecutor 的构造函数。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class RejectedExecutionHandlerExample {

    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(
                5, // corePoolSize
                10, // maximumPoolSize
                60, // keepAliveTime
                TimeUnit.SECONDS, // unit
                new ArrayBlockingQueue<>(5), // workQueue
                new CustomRejectedExecutionHandler() // rejectedExecutionHandler
        );

        // 提交任务
        for (int i = 0; i < 20; i++) {
            final int taskNumber = i;
            executorService.submit(() -> {
                try {
                    System.out.println("Task " + taskNumber + " started.");
                    Thread.sleep(1000); // 模拟耗时操作
                    System.out.println("Task " + taskNumber + " finished.");
                } catch (InterruptedException e) {
                    System.out.println("Task " + taskNumber + " interrupted.");
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 关闭线程池
        executorService.shutdown();
    }
}

代码解释：

创建线程池： 使用 ThreadPoolExecutor 的构造函数创建一个线程池，并将自定义的 CustomRejectedExecutionHandler 传递给它。
提交任务： 提交 20 个任务到线程池。由于线程池的队列大小为 5，最多同时运行 10 个任务，因此会有一些任务被拒绝。
自定义 RejectedExecutionHandler： 当线程池无法接受新任务时，CustomRejectedExecutionHandler 的 rejectedExecution() 方法会被调用，可以在这里执行自定义的操作，例如记录日志、重试任务或者执行其他自定义操作。

总结一下，优雅关闭线程池的关键步骤

使用 shutdown() 方法拒绝接受新的任务。
使用 awaitTermination() 方法等待所有已提交的任务完成执行。
如果 awaitTermination() 方法超时，则使用 shutdownNow() 方法尝试中断所有正在执行的任务。
使用钩子函数来确保线程池在 JVM 关闭前被正确关闭。
可以自定义 RejectedExecutionHandler 来更精细地控制线程池的行为。

钩子函数是保障，shutdownGracefully 是优雅，RejectedExecutionHandler 是补充

钩子函数提供了一种在JVM关闭时执行清理操作的保障机制，确保线程池能够被正确关闭。shutdownGracefully 方法封装了优雅关闭线程池的常用逻辑，提高了代码的可重用性和可维护性。自定义 RejectedExecutionHandler 则允许开发者根据具体需求，对线程池拒绝任务的行为进行更精细的控制，例如记录被拒绝的任务信息或执行重试策略。

希望通过今天的分享，大家能够更加深入地理解如何优雅地关闭线程池，并在实际项目中应用这些技巧，避免潜在的问题。谢谢大家！

JAVA 项目如何优雅关闭线程池？钩子函数与 shutdownGracefully 实战

发表回复 取消回复

发表回复取消回复