线程池拒绝策略：优雅处理过载请求

在现代软件开发中，线程池（ThreadPool）是一个常用的并发处理工具。它能有效地管理和复用线程资源，提升系统的性能和响应速度。然而，当线程池达到最大容量时，会出现任务无法提交的情况，这时需要处理这种情况的机制称为“拒绝策略”（Rejection Policy）。本文将详细讲述 Java 中的线程池拒绝策略，帮助读者理解其原理和应用场景。

什么是线程池拒绝策略？

线程池拒绝策略是指当线程池中的任务数量达到上限时，新提交的任务如何处理的一种策略。Java 提供了几种内置的拒绝策略，开发者也可以自定义策略。

内置的拒绝策略

在Java中，java.util.concurrent.ThreadPoolExecutor类提供了四种预定义的拒绝策略，它们分别是：

1. AbortPolicy：这是默认的拒绝策略。当新任务到来但无法被处理时，会抛出一个RejectedExecutionException异常，并关闭调用者提供的Runnable或Callable对象的Future，如果适用的话。
2. CallerRunsPolicy：这个策略不会丢弃任务，也不会创建新的线程，而是将任务回退给调用者，由调用者的线程直接执行该任务。这可能会导致调用者的线程阻塞，直到任务完成。
3. DiscardOldestPolicy：如果无法执行新任务，则会从任务队列中移除最老的任务，然后尝试再次提交新任务。这种策略可以防止队列无限制地增长，但是可能会导致某些任务永远无法被执行。
4. DiscardPolicy：简单地丢弃无法处理的任务，不抛出异常，也不通知调用者。这种策略可以避免系统因异常而崩溃，但同时也意味着某些重要任务可能被静默地丢弃，这在某些场景下可能是不可接受的。
   每种策略都有其适用场景，选择哪种策略取决于应用程序的具体需求和对失败任务的容忍度。例如，在资源有限且需要保证系统稳定性的环境中，DiscardPolicy或DiscardOldestPolicy可能更为合适；而在需要确保每个任务都被处理的情况下，CallerRunsPolicy可能是更好的选择。

为了设置线程池的拒绝策略，可以通过ThreadPoolExecutor构造函数中的RejectedExecutionHandler参数来指定，或者在创建线程池后通过setRejectedExecutionHandler()方法来修改。

代码示例

创建一个简单的线程池

import java.util.concurrent.*;public class ThreadPoolExample {private static final int CORE_POOL_SIZE = 2;private static final int MAX_POOL_SIZE = 4;private static final long KEEP_ALIVE_TIME = 10L;public static void main(String[] args) {// 使用 ArrayBlockingQueue 作为任务队列，容量为 2BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(2);// 创建线程池ThreadPoolExecutor executor = new ThreadPoolExecutor(CORE_POOL_SIZE,MAX_POOL_SIZE,KEEP_ALIVE_TIME,TimeUnit.SECONDS,queue);// 提交任务for (int i = 0; i < 10; i++) {final int taskNumber = i + 1;executor.submit(() -> {try {System.out.println("Executing task " + taskNumber);Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}// 关闭线程池executor.shutdown();}
}

使用 AbortPolicy

import java.util.concurrent.*;public class AbortPolicyExample {public static void main(String[] args) {BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(2);ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 10L, TimeUnit.SECONDS, queue,new ThreadPoolExecutor.AbortPolicy() // 使用 AbortPolicy);for (int i = 0; i < 10; i++) {final int taskNumber = i + 1;try {executor.submit(() -> {try {System.out.println("Executing task " + taskNumber);Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});} catch (RejectedExecutionException e) {System.err.println("Task " + taskNumber + " was rejected");}}executor.shutdown();}
}

使用 CallerRunsPolicy

import java.util.concurrent.*;public class CallerRunsPolicyExample {public static void main(String[] args) {BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(2);ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 10L, TimeUnit.SECONDS, queue,new ThreadPoolExecutor.CallerRunsPolicy() // 使用 CallerRunsPolicy);for (int i = 0; i < 10; i++) {final int taskNumber = i + 1;executor.submit(() -> {System.out.println("Executing task " + taskNumber);try {Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();}
}

使用 DiscardPolicy

import java.util.concurrent.*;public class DiscardPolicyExample {public static void main(String[] args) {BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(2);ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 10L, TimeUnit.SECONDS, queue,new ThreadPoolExecutor.DiscardPolicy() // 使用 DiscardPolicy);for (int i = 0; i < 10; i++) {final int taskNumber = i + 1;executor.submit(() -> {System.out.println("Executing task " + taskNumber);try {Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();}
}

使用 DiscardOldestPolicy

import java.util.concurrent.*;public class DiscardOldestPolicyExample {public static void main(String[] args) {BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(2);ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 10L, TimeUnit.SECONDS, queue,new ThreadPoolExecutor.DiscardOldestPolicy() // 使用 DiscardOldestPolicy);for (int i = 0; i < 10; i++) {final int taskNumber = i + 1;executor.submit(() -> {System.out.println("Executing task " + taskNumber);try {Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();}
}

通俗易懂地理解这些拒绝策略

为了更好地理解这些拒绝策略，我们可以将其类比于生活中的场景：

1. AbortPolicy：就像餐厅已满员，不再接待新客人，并告知客人“已经客满，请去别处”。
2. CallerRunsPolicy：就像餐厅忙不过来时，老板自己上阵服务客人，保证所有客人都能被服务到。
3. DiscardPolicy：就像餐厅已满员，直接不理会新来的客人，不告知任何信息。
4. DiscardOldestPolicy：就像餐厅已满员，把最早来但还没点菜的客人请走，以便接待新来的客人。
   自定义拒绝策略

自定义拒绝策略

除了内置的拒绝策略，开发者还可以根据实际需求自定义拒绝策略。例如，记录日志、发送通知等。

import java.util.concurrent.*;public class CustomRejectionPolicyExample {public static void main(String[] args) {BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(2);ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 10L, TimeUnit.SECONDS, queue,new RejectedExecutionHandler() {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {System.err.println("Task " + r.toString() + " was rejected");// 这里可以添加更多处理逻辑，比如记录日志、发送通知等}});for (int i = 0; i < 10; i++) {final int taskNumber = i + 1;executor.submit(() -> {System.out.println("Executing task " + taskNumber);try {Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}executor.shutdown();}
}

总结

线程池拒绝策略概念
线程池拒绝策略是在线程池达到最大容量且任务队列满负荷时，用于处理额外任务的方法。这在并发编程中至关重要，因为它帮助系统在高负载情况下保持稳定性和可预测性。

内置拒绝策略详解
Java的ThreadPoolExecutor提供了四种主要的内置拒绝策略：

1. AbortPolicy：默认策略，当无法执行新任务时，抛出RejectedExecutionException，并关闭相关Future，适用于需要立即反馈不可执行情况的场景。
2. CallerRunsPolicy：调用者线程将直接执行任务，避免了任务的丢失，但可能导致调用者线程阻塞。
3. DiscardOldestPolicy：移除队列中最旧的任务，为新任务腾出空间，适用于队列长度控制和优先级管理。
4. DiscardPolicy：简单地丢弃新任务而不抛出异常，适用于对任务丢失容忍度较高的场景，以避免系统因异常处理而加重负担。

代码示例

• 创建线程池：通过ThreadPoolExecutor构造函数指定核心线程数、最大线程数、空闲线程存活时间、任务队列类型和容量以及拒绝策略。
• 使用不同策略：示例代码展示了如何分别使用AbortPolicy、CallerRunsPolicy、DiscardPolicy和DiscardOldestPolicy，以及它们在任务过载时的行为差异。

生活场景类比
为了加深理解，上述策略被类比于餐厅管理场景，形象说明了不同策略下的处理方式。

自定义拒绝策略
开发者可以根据具体需求实现RejectedExecutionHandler接口来自定义拒绝策略，例如记录日志、发送警报邮件或执行其他业务逻辑。

应用场景指导

• 在资源有限的环境下，选择DiscardPolicy或DiscardOldestPolicy以维持系统稳定性。
• 当每个任务都需要被处理时，使用CallerRunsPolicy确保任务的执行，尽管可能会牺牲调用者线程的响应能力。

结论
合理选择和配置线程池的拒绝策略对于构建高效、健壮的并发系统至关重要。开发者应根据系统需求和预期行为精心挑选策略，以达到最佳的并发处理效果。