JAVA ThreadPoolExecutor 队列拒绝策略配置错误？RejectedExecutionHandler 应用技巧

JAVA ThreadPoolExecutor 队列拒绝策略配置错误？RejectedExecutionHandler 应用技巧

大家好，今天我们来深入探讨 ThreadPoolExecutor 的一个关键但容易被忽略的方面：拒绝策略（RejectedExecutionHandler）。配置不当的拒绝策略可能导致应用程序崩溃、任务丢失，甚至数据损坏。我们将剖析常见的配置错误，并分享一些实用的 RejectedExecutionHandler 应用技巧，帮助大家构建更健壮、更可靠的并发程序。

1. ThreadPoolExecutor 核心概念回顾

在深入拒绝策略之前，让我们快速回顾一下 ThreadPoolExecutor 的核心组件：

• 核心线程数 (corePoolSize): 线程池中始终保持的线程数量，即使它们处于空闲状态。
• 最大线程数 (maximumPoolSize): 线程池允许创建的最大线程数量。
• 保持活跃时间 (keepAliveTime): 当线程池中的线程数量超过核心线程数时，空闲线程在终止之前等待新任务的最长时间。
• 时间单位 (unit): keepAliveTime 的时间单位 (例如，秒、毫秒)。
• 工作队列 (workQueue): 用于保存等待执行的任务的队列。
• 线程工厂 (threadFactory): 用于创建新线程的工厂。
• 拒绝策略 (handler): 当工作队列已满且线程池中的线程数量达到最大线程数时，用于处理新提交的任务的策略。

ThreadPoolExecutor 的工作流程大致如下：

1. 当新任务提交到线程池时，如果线程池中的线程数量小于核心线程数，则创建新线程来执行任务。
2. 如果线程池中的线程数量大于或等于核心线程数，则将任务添加到工作队列中。
3. 如果工作队列已满，且线程池中的线程数量小于最大线程数，则创建新线程来执行任务。
4. 如果工作队列已满，且线程池中的线程数量达到最大线程数，则根据配置的拒绝策略来处理该任务。

2. 常见拒绝策略及配置错误

ThreadPoolExecutor 提供了四个默认的拒绝策略：

策略名称	描述	可能的配置错误	适用场景
AbortPolicy	抛出 RejectedExecutionException 异常。	将此策略用于不允许任务丢失的场景，但未对异常进行适当处理，导致应用程序崩溃。 没有良好的异常处理机制，比如try-catch，可能直接中断整个程序。	适用于要求任务必须执行，且允许应用程序抛出异常并进行适当处理的场景。例如，关键业务操作，必须保证成功或失败的明确通知。
CallerRunsPolicy	由提交任务的线程来执行该任务。	在高负载情况下，可能导致提交任务的线程（通常是 UI 线程或网络处理线程）被阻塞，影响系统响应。 如果任务执行时间过长，会进一步加剧阻塞，甚至造成死锁。	适用于可以容忍任务执行延迟，并且希望避免任务丢失的场景。例如，可以将一些非关键的任务交给提交任务的线程来执行，以降低线程池的压力。
DiscardPolicy	静默丢弃该任务，不抛出任何异常。	将此策略用于重要任务，导致数据丢失且难以追踪问题。 没有日志记录，导致问题发生时难以定位和诊断。 错误地认为该策略不会对系统产生任何影响，忽略了数据丢失的潜在风险。	适用于可以容忍任务丢失，且不希望抛出异常或阻塞提交任务的线程的场景。例如，可以丢弃一些不重要的日志或统计数据。
DiscardOldestPolicy	丢弃工作队列中最旧的任务，然后尝试重新提交该任务。	将此策略用于重要任务，导致重要任务被丢弃，数据一致性受到威胁。 没有监控机制，无法及时发现重要任务被丢弃的情况。 错误地认为该策略总是能保证最新的任务被执行，忽略了任务优先级和依赖关系。	适用于希望尽可能执行最新的任务，并且可以容忍丢弃旧任务的场景。例如，可以丢弃一些过期的缓存或状态信息。 需要注意的是，这种策略可能导致某些任务永远无法被执行。

配置错误通常源于以下几个方面：

• 盲目选择默认策略: 没有充分评估应用程序的需求，直接使用默认的 AbortPolicy，导致应用程序在任务被拒绝时崩溃。
• 忽略任务的重要性: 将 DiscardPolicy 或 DiscardOldestPolicy 用于重要任务，导致数据丢失或数据不一致。
• 未考虑线程上下文: 使用 CallerRunsPolicy 时，没有考虑到提交任务的线程的上下文，可能导致 UI 线程阻塞或网络处理线程超时。
• 缺乏监控和告警: 没有对拒绝策略的执行情况进行监控和告警，导致问题发生时无法及时发现和解决。

3. 自定义 RejectedExecutionHandler 的必要性

默认的拒绝策略在很多情况下无法满足实际需求。例如：

• 需要记录被拒绝的任务: 默认策略无法记录被拒绝的任务，不利于问题排查。
• 需要重试被拒绝的任务: 默认策略无法重试被拒绝的任务，可能导致任务丢失。
• 需要将任务持久化到数据库: 默认策略无法将任务持久化到数据库，可能导致数据丢失。
• 需要发送告警通知: 默认策略无法发送告警通知，可能导致问题被忽略。

因此，自定义 RejectedExecutionHandler 是非常必要的。

4. RejectedExecutionHandler 应用技巧及代码示例

下面我们将分享一些自定义 RejectedExecutionHandler 的应用技巧，并提供相应的代码示例。

4.1 记录被拒绝的任务

我们可以创建一个自定义的 RejectedExecutionHandler，用于记录被拒绝的任务信息，例如任务的名称、提交时间、执行参数等。

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

public class LoggingRejectedExecutionHandler implements RejectedExecutionHandler {

    private static final Logger logger = Logger.getLogger(LoggingRejectedExecutionHandler.class.getName());

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        logger.severe("Task " + r.toString() +
                " rejected from " +
                executor.toString());
        // 可以添加更多任务信息，例如任务的提交时间、执行参数等
    }
}

//使用示例
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Example {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5,
                10,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100),
                new LoggingRejectedExecutionHandler()
        );

        // 提交任务...
    }
}

4.2 重试被拒绝的任务

我们可以创建一个自定义的 RejectedExecutionHandler，用于将任务重新提交到线程池中。 但需要注意避免无限重试，导致系统资源耗尽。

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

public class RetryRejectedExecutionHandler implements RejectedExecutionHandler {

    private static final Logger logger = Logger.getLogger(RetryRejectedExecutionHandler.class.getName());
    private final int maxRetries;
    private final long retryIntervalMillis;

    public RetryRejectedExecutionHandler(int maxRetries, long retryIntervalMillis) {
        this.maxRetries = maxRetries;
        this.retryIntervalMillis = retryIntervalMillis;
    }

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        int retries = 0;
        while (retries < maxRetries) {
            try {
                Thread.sleep(retryIntervalMillis);
                executor.execute(r);
                logger.info("Task " + r.toString() + " re-submitted successfully.");
                return; // 成功提交，退出循环
            } catch (InterruptedException e) {
                logger.warning("Thread interrupted during retry: " + e.getMessage());
                Thread.currentThread().interrupt(); // 重新设置中断状态
                return; // 中断异常，退出循环
            } catch (Exception e) {
                logger.warning("Failed to re-submit task " + r.toString() + ": " + e.getMessage());
                retries++;
            }
        }
        logger.severe("Task " + r.toString() + " failed to re-submit after " + maxRetries + " retries.");
        // 如果重试多次仍然失败，可以抛出异常或执行其他操作
    }
}

// 使用示例
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Example {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5,
                10,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100),
                new RetryRejectedExecutionHandler(3, 1000) // 最大重试3次，每次间隔1秒
        );

        // 提交任务...
    }
}

4.3 将任务持久化到数据库

我们可以创建一个自定义的 RejectedExecutionHandler，用于将任务信息持久化到数据库中，以便后续重试或分析。

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.logging.Logger;

public class PersistenceRejectedExecutionHandler implements RejectedExecutionHandler {

    private static final Logger logger = Logger.getLogger(PersistenceRejectedExecutionHandler.class.getName());
    private final String jdbcUrl;
    private final String username;
    private final String password;

    public PersistenceRejectedExecutionHandler(String jdbcUrl, String username, String password) {
        this.jdbcUrl = jdbcUrl;
        this.username = username;
        this.password = password;
    }

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        try (Connection connection = DriverManager.getConnection(jdbcUrl, username, password);
             PreparedStatement statement = connection.prepareStatement("INSERT INTO rejected_tasks (task_name, submission_time) VALUES (?, ?)")) {
            statement.setString(1, r.toString());
            statement.setTimestamp(2, new java.sql.Timestamp(System.currentTimeMillis()));
            statement.executeUpdate();
            logger.info("Task " + r.toString() + " persisted to database.");
        } catch (SQLException e) {
            logger.severe("Failed to persist task " + r.toString() + " to database: " + e.getMessage());
            // 可以选择抛出异常或执行其他操作
        }
    }
}

// 使用示例
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Example {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5,
                10,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100),
                new PersistenceRejectedExecutionHandler("jdbc:mysql://localhost:3306/mydb", "user", "password")
        );

        // 提交任务...
    }
}

4.4 发送告警通知

我们可以创建一个自定义的 RejectedExecutionHandler，用于发送告警通知，例如发送邮件、短信或调用告警 API。

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

public class AlertingRejectedExecutionHandler implements RejectedExecutionHandler {

    private static final Logger logger = Logger.getLogger(AlertingRejectedExecutionHandler.class.getName());
    private final String alertRecipient; // 告警接收者，例如邮箱地址或手机号码

    public AlertingRejectedExecutionHandler(String alertRecipient) {
        this.alertRecipient = alertRecipient;
    }

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        String message = "Task " + r.toString() + " rejected from " + executor.toString();
        sendAlert(message);
        logger.warning(message + ". Alert sent to " + alertRecipient);
    }

    private void sendAlert(String message) {
        // 在这里实现发送告警通知的逻辑，例如发送邮件、短信或调用告警 API
        // 这只是一个示例，需要根据实际情况进行修改
        System.out.println("Sending alert to " + alertRecipient + ": " + message);
    }
}

// 使用示例
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Example {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5,
                10,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100),
                new AlertingRejectedExecutionHandler("[email protected]")
        );

        // 提交任务...
    }
}

4.5 延迟执行被拒绝的任务

我们可以创建一个自定义的 RejectedExecutionHandler，用于将任务延迟一段时间后重新提交到线程池中。 这可以用于处理一些瞬时性的负载高峰。

import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

public class DelayedRetryRejectedExecutionHandler implements RejectedExecutionHandler {

    private static final Logger logger = Logger.getLogger(DelayedRetryRejectedExecutionHandler.class.getName());
    private final long delayMillis;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); // 使用单独的调度线程池

    public DelayedRetryRejectedExecutionHandler(long delayMillis) {
        this.delayMillis = delayMillis;
    }

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        logger.info("Task " + r.toString() + " rejected, scheduling for delayed retry in " + delayMillis + "ms.");
        scheduler.schedule(() -> {
            try {
                executor.execute(r);
                logger.info("Task " + r.toString() + " re-submitted after delay.");
            } catch (Exception e) {
                logger.severe("Failed to re-submit task " + r.toString() + " after delay: " + e.getMessage());
                // 可以选择抛出异常或执行其他操作
            }
        }, delayMillis, TimeUnit.MILLISECONDS);
    }
}

// 使用示例
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Example {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5,
                10,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100),
                new DelayedRetryRejectedExecutionHandler(5000) // 延迟5秒后重试
        );

        // 提交任务...
    }
}

注意事项:

• 在自定义 RejectedExecutionHandler 时，需要考虑到线程安全问题，避免出现并发错误。
• 尽量避免在 rejectedExecution 方法中执行耗时操作，以免阻塞线程池的正常运行。
• 对拒绝策略的执行情况进行监控和告警，以便及时发现和解决问题。
• 根据实际需求选择合适的拒绝策略，避免盲目使用默认策略。

5. 选择合适的策略是关键

选择合适的拒绝策略需要仔细评估应用程序的需求和特点。以下是一些建议：

• 高可靠性系统: 如果任务不能丢失，可以考虑使用自定义的 RejectedExecutionHandler，将任务持久化到数据库或重试。 AbortPolicy配合良好的异常处理也是一个选择，但要确保异常处理逻辑的正确性。
• 高吞吐量系统: 如果可以容忍少量任务丢失，可以考虑使用 DiscardPolicy 或 DiscardOldestPolicy，以避免阻塞线程池的正常运行。
• 交互式系统: 避免使用 CallerRunsPolicy，以免阻塞 UI 线程或网络处理线程。 可以考虑使用自定义的 RejectedExecutionHandler，将任务放入优先级队列或延迟执行。
• 监控系统: 无论选择哪种拒绝策略，都应该对拒绝策略的执行情况进行监控和告警，以便及时发现和解决问题。

6. 工作队列的选择也很重要

ThreadPoolExecutor 的工作队列也会影响拒绝策略的触发。常见的队列类型包括：

• LinkedBlockingQueue: 一个无界队列，可以容纳无限数量的任务。 使用此队列时，maximumPoolSize 实际上不起作用，因为线程池会一直创建线程来处理队列中的任务，直到达到系统资源限制。
• ArrayBlockingQueue: 一个有界队列，可以容纳固定数量的任务。 当队列已满时，新提交的任务将被拒绝。
• SynchronousQueue: 一个不存储元素的队列。 每个插入操作必须等待一个相应的移除操作，反之亦然。 使用此队列时，线程池会尽可能创建线程来处理任务，直到达到 maximumPoolSize。

选择合适的队列类型需要根据应用程序的需求和特点进行权衡。

7. 总结：理解与灵活应用

ThreadPoolExecutor 的拒绝策略是一个重要的配置选项，配置不当可能导致应用程序出现各种问题。 通过理解不同的拒绝策略的特点，并根据实际需求自定义 RejectedExecutionHandler，可以构建更健壮、更可靠的并发程序。

我们需要根据实际场景，理解各种拒绝策略的优缺点，并灵活运用自定义 RejectedExecutionHandler，才能更好地应对各种并发场景。同时，工作队列的选择也至关重要，需要与拒绝策略配合使用，才能达到最佳效果。