Java线程池的CallerRunsPolicy：一种在高负载下避免任务丢失的拒绝策略

大家好，今天我们来深入探讨Java线程池的一种重要的拒绝策略：CallerRunsPolicy。在并发编程中，线程池是管理和控制并发执行任务的核心工具。当提交给线程池的任务数量超过其处理能力时，就需要一种机制来处理这些过载的任务，避免系统崩溃或数据丢失。CallerRunsPolicy正是这样一种策略，它在应对高负载时，提供了一种优雅且实用的解决方案。

线程池基础回顾

在深入CallerRunsPolicy之前，我们先快速回顾一下Java线程池的基本概念。Java 通过 java.util.concurrent.ExecutorService 接口提供了线程池的抽象，常见的实现类包括 ThreadPoolExecutor 和 ScheduledThreadPoolExecutor。

ThreadPoolExecutor 允许我们配置线程池的核心参数，例如：

corePoolSize: 线程池中保持的最小线程数。即使线程处于空闲状态，也不会被销毁。
maximumPoolSize: 线程池中允许的最大线程数。
keepAliveTime: 当线程数大于核心线程数时，空闲线程在终止之前保持活动的最大时间。
unit: keepAliveTime 的时间单位。
workQueue: 用于保存等待执行的任务的阻塞队列。

一个典型的 ThreadPoolExecutor 的创建方式如下：

import java.util.concurrent.*;

public class ThreadPoolExample {

    public static void main(String[] args) {
        int corePoolSize = 5;
        int maximumPoolSize = 10;
        long keepAliveTime = 60;
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(100); // 使用有界队列
        RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy(); //设置拒绝策略

        ThreadPoolExecutor executorService = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, rejectedExecutionHandler);

        for (int i = 0; i < 200; i++) {
            final int taskNumber = i;
            executorService.execute(() -> {
                System.out.println("Task " + taskNumber + " is running in thread: " + Thread.currentThread().getName());
                try {
                    Thread.sleep(100); // 模拟任务执行时间
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        executorService.shutdown(); //不再接受新的任务
        try {
            executorService.awaitTermination(5, TimeUnit.MINUTES); //等待所有任务完成，最多等待5分钟
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("All tasks completed.");
    }
}

拒绝策略：应对任务过载的关键

当线程池的任务队列已满，并且线程池中的线程数量已达到 maximumPoolSize 时，再提交新的任务将会触发拒绝策略。RejectedExecutionHandler 接口定义了拒绝策略的行为。Java 提供了以下几种内置的拒绝策略：

拒绝策略	描述
`AbortPolicy` (默认)	直接抛出 `RejectedExecutionException` 异常，阻止新任务的提交。
`DiscardPolicy`	静默丢弃新提交的任务，不抛出任何异常，也不做任何处理。
`DiscardOldestPolicy`	丢弃任务队列中最旧的任务（即等待时间最长的任务），然后尝试再次提交新任务。
`CallerRunsPolicy`	由提交任务的线程（即调用 `execute()` 方法的线程）来执行被拒绝的任务。这是一种回退策略，它不会丢弃任务，而是将任务的执行责任交给调用者。
自定义 `RejectedExecutionHandler` 实现	允许开发者根据特定业务需求，实现自定义的拒绝策略。例如，可以将任务记录到日志中，或者将其放入持久化存储中，以便稍后重试。

CallerRunsPolicy 的工作原理

CallerRunsPolicy 的核心思想是：当线程池无法处理新提交的任务时，将任务的执行责任“返还”给提交任务的线程。也就是说，如果线程池拒绝了一个任务，那么调用 execute() 方法的线程会直接执行该任务。

这种策略有以下几个关键特点：

避免任务丢失： 与 DiscardPolicy 不同，CallerRunsPolicy 不会丢弃任务。它保证所有提交的任务最终都会被执行。
减缓任务提交速度： 由于提交任务的线程需要承担执行任务的责任，这会在一定程度上降低任务提交的速度。这种减速效应可以帮助系统从过载状态中恢复，防止进一步的恶化。
可能导致死锁： 如果提交任务的线程本身也依赖于线程池中的其他任务，那么 CallerRunsPolicy 可能会导致死锁。因为提交线程正在执行被拒绝的任务，而线程池中的其他线程可能正在等待该提交线程释放资源。
适用于非关键任务： CallerRunsPolicy 适用于那些即使延迟执行也不会对系统造成严重影响的任务。例如，日志记录、统计分析等。

代码示例：CallerRunsPolicy 的实际应用

让我们通过一个代码示例来演示 CallerRunsPolicy 的实际应用。

import java.util.concurrent.*;

public class CallerRunsPolicyExample {

    public static void main(String[] args) {
        int corePoolSize = 2;
        int maximumPoolSize = 4;
        long keepAliveTime = 60;
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(2); // 容量较小的队列，容易触发拒绝策略
        RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy();

        ThreadPoolExecutor executorService = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, rejectedExecutionHandler);

        for (int i = 0; i < 10; i++) {
            final int taskNumber = i;
            executorService.execute(() -> {
                System.out.println("Task " + taskNumber + " is running in thread: " + Thread.currentThread().getName());
                try {
                    Thread.sleep(100); // 模拟任务执行时间
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        executorService.shutdown(); //不再接受新的任务
        try {
            executorService.awaitTermination(5, TimeUnit.MINUTES); //等待所有任务完成，最多等待5分钟
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("All tasks completed.");
    }
}

在这个例子中，我们将任务队列的容量设置为 2，核心线程数设置为 2，最大线程数设置为 4。当提交的任务数量超过线程池的处理能力时，CallerRunsPolicy 会被触发。你会观察到，某些任务会由 main 线程执行，而不是由线程池中的线程执行。这表明 CallerRunsPolicy 正在发挥作用。

CallerRunsPolicy 的优缺点

下面我们总结一下 CallerRunsPolicy 的优缺点：

优点：

保证任务执行： 不会丢弃任务，确保所有提交的任务最终都会被执行。
减缓任务提交速度： 通过让提交线程执行任务，可以降低任务提交的速度，缓解系统压力。
简单易用： 使用起来非常简单，只需要在创建 ThreadPoolExecutor 时指定即可。

缺点：

可能导致死锁： 如果提交任务的线程依赖于线程池中的其他任务，可能会导致死锁。
影响调用线程的性能： 提交任务的线程需要承担执行任务的责任，这会影响其自身的性能。
任务执行顺序不确定： 被拒绝的任务可能不会按照提交的顺序执行。

何时使用 CallerRunsPolicy

CallerRunsPolicy 是一种非常有用的拒绝策略，但并非适用于所有场景。在以下情况下，可以考虑使用 CallerRunsPolicy：

任务丢失是不可接受的： 如果必须保证所有提交的任务最终都会被执行，那么 CallerRunsPolicy 是一个不错的选择。
系统可以容忍任务执行的延迟： CallerRunsPolicy 可能会导致任务执行的延迟，因此它适用于那些即使延迟执行也不会对系统造成严重影响的任务。
提交任务的线程不是关键线程： 如果提交任务的线程不是关键线程，那么它可以承担执行被拒绝任务的责任。
需要一种简单的回退机制： CallerRunsPolicy 提供了一种简单而有效的回退机制，可以帮助系统从过载状态中恢复。

不适用场景：

高并发、低延迟的场景： CallerRunsPolicy 会降低任务提交的速度，不适合对延迟非常敏感的场景。
提交任务的线程是关键线程： 如果提交任务的线程是关键线程，那么不应该让它承担执行被拒绝任务的责任。
存在死锁风险的场景： 如果提交任务的线程依赖于线程池中的其他任务，那么应该避免使用 CallerRunsPolicy，以防止死锁。

其他拒绝策略的比较

为了更好地理解 CallerRunsPolicy 的特点，我们将其与其他拒绝策略进行比较：

特性	`AbortPolicy`	`DiscardPolicy`	`DiscardOldestPolicy`	`CallerRunsPolicy`
任务丢失	是	是	是	否
异常抛出	是	否	否	否
减缓提交速度	否	否	否	是
死锁风险	低	低	低	高
适用场景	快速失败，不关心任务丢失	不关心任务丢失	丢弃旧任务，保证新任务	保证任务执行，可容忍延迟

自定义拒绝策略

除了使用 Java 提供的内置拒绝策略之外，我们还可以根据自己的业务需求，实现自定义的 RejectedExecutionHandler。例如，可以将任务记录到日志中，或者将其放入持久化存储中，以便稍后重试。

下面是一个自定义拒绝策略的示例：

import java.util.concurrent.*;

public class CustomRejectedExecutionHandler implements RejectedExecutionHandler {

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.err.println("Task rejected: " + r.toString() + ", Executor: " + executor.toString());
        // 可以将任务记录到日志中，或者将其放入持久化存储中
        // 例如：
        // Log.error("Task rejected: " + r.toString() + ", Executor: " + executor.toString());
        // taskQueue.offer(r); // 放入重试队列
    }
}

使用自定义拒绝策略：

RejectedExecutionHandler rejectedExecutionHandler = new CustomRejectedExecutionHandler();
ThreadPoolExecutor executorService = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, rejectedExecutionHandler);

实际案例分析：使用CallerRunsPolicy优化系统

假设我们有一个在线购物网站，用户在下单后需要发送一封确认邮件。发送邮件的任务可以提交到线程池中异步执行。但是，在高并发情况下，如果发送邮件的任务过多，可能会导致线程池过载。

在这种情况下，我们可以使用 CallerRunsPolicy 作为拒绝策略。这样，即使线程池过载，发送邮件的任务也不会被丢弃，而是由提交任务的线程（例如，处理用户下单请求的线程）来执行。虽然这会稍微增加下单请求的响应时间，但可以保证所有用户都能收到确认邮件。

代码示例：

import java.util.concurrent.*;

public class EmailSendingExample {

    private static final ThreadPoolExecutor emailExecutor = new ThreadPoolExecutor(
            5, // corePoolSize
            10, // maximumPoolSize
            60L, // keepAliveTime
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100), // workQueue
            new ThreadPoolExecutor.CallerRunsPolicy() // rejectedExecutionHandler
    );

    public static void sendConfirmationEmail(String userEmail, String orderDetails) {
        emailExecutor.execute(() -> {
            System.out.println("Sending confirmation email to: " + userEmail + " for order: " + orderDetails + " in thread: " + Thread.currentThread().getName());
            try {
                // 模拟发送邮件的过程
                Thread.sleep(500);
                System.out.println("Email sent successfully to: " + userEmail);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    public static void main(String[] args) {
        // 模拟用户下单请求
        for (int i = 0; i < 200; i++) {
            final int orderId = i;
            sendConfirmationEmail("user" + i + "@example.com", "Order ID: " + orderId);
        }

        emailExecutor.shutdown();
        try {
            emailExecutor.awaitTermination(5, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("All emails sent.");
    }
}

在这个例子中，sendConfirmationEmail 方法将发送邮件的任务提交到 emailExecutor 线程池中。如果线程池过载，CallerRunsPolicy 会被触发，由调用 sendConfirmationEmail 方法的线程来执行发送邮件的任务。

总结

CallerRunsPolicy 是一种简单而有效的拒绝策略，它通过将任务的执行责任返还给提交线程来避免任务丢失，并减缓任务提交速度。它适用于那些任务丢失是不可接受的，并且系统可以容忍任务执行延迟的场景。然而，需要注意的是，CallerRunsPolicy 可能会导致死锁，并且会影响调用线程的性能。在选择拒绝策略时，需要根据具体的业务需求和系统特点进行权衡。理解并合理运用 CallerRunsPolicy，能够帮助我们构建更加健壮和可靠的并发系统。

几句总结

CallerRunsPolicy避免任务丢失，但可能影响调用者性能。
权衡利弊，谨慎使用，确保系统整体稳定。
选择合适的拒绝策略，构建健壮并发系统。