自定义线程池饱和策略：实现一个将任务暂存到数据库或发送到消息队列的容灾拒绝方案-Java面试题-PHP中文网

线程池饱和时应将拒绝任务转存至数据库或消息队列实现容灾兜底：数据库适合强一致、需人工干预场景；mq适合高吞吐、多消费者场景；推荐组合策略——优先mq，失败再db，最后日志告警。

自定义线程池饱和策略：实现一个将任务暂存到数据库或发送到消息队列的容灾拒绝方案

当线程池饱和（即线程数已达 maximumPoolSize 且任务队列已满）时，直接丢弃或抛异常可能引发业务数据丢失或用户体验中断。若任务具备“可延迟执行”“需持久化保障”特性（如订单异步通知、用户行为埋点、风控结果落库），推荐将拒绝任务转存至外部可靠介质——比如数据库表或消息队列，实现容灾兜底。

为什么选数据库或消息队列做容灾存储

二者本质都是提供持久化 + 异步重试能力的中间载体：

数据库：适合任务结构固定、需强一致性、后续要人工干预或关联查询的场景（如“支付结果回调失败待重试表”）；写入可控、事务支持好，但高并发写入需注意连接池与索引优化。
消息队列（如 Kafka/RocketMQ）：适合吞吐量大、天然支持削峰填谷、需多消费者协同处理的场景；具备重试、死信、顺序等机制，但引入了额外中间件依赖和网络开销。

实现一个带数据库落库的自定义拒绝策略

核心思路：在 rejectedExecution 中将任务序列化并插入数据库，不阻塞主线程（建议异步写库或使用连接池+超时控制）：

public class DbFallbackPolicy implements RejectedExecutionHandler {
    private final JdbcTemplate jdbcTemplate; // Spring JDBC 模板
    private final String insertSql = "INSERT INTO task_fallback (task_class, task_data, create_time) VALUES (?, ?, ?)";

    public DbFallbackPolicy(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (executor.isShutdown()) return;

        try {
            // 简单序列化：仅记录类名和 toString()（生产中建议用 JSON 或 Protobuf）
            String taskClass = r.getClass().getName();
            String taskData = r.toString(); // 或使用 Jackson.writeValueAsString(r)

            jdbcTemplate.update(insertSql, taskClass, taskData, new Date());
            System.out.println("任务已降级落库：" + taskClass);

        } catch (Exception e) {
            // 落库失败也不应影响主流程，可降级为日志告警
            System.err.println("任务落库失败，将尝试本地日志备份：" + e.getMessage());
            // 可选：写入本地文件或发告警
        }
    }
}

配套建议：

建表字段至少包含：id、task_class、task_data（TEXT）、status（如 'pending'/'failed'/'success'）、retry_count、create_time、next_retry_time；
搭配一个定时任务或独立消费者服务，定期扫描 status = 'pending' 的记录，反序列化后重新提交到线程池或调用业务逻辑；
对 task_data 做长度限制或压缩，避免 BLOB 字段撑爆数据库。

实现一个发送到消息队列的拒绝策略（以 RocketMQ 为例）

利用 MQ 的可靠投递和重试能力，更贴近“解耦+弹性”的设计原则：

public class MqFallbackPolicy implements RejectedExecutionHandler {
    private final DefaultMQProducer producer;
    private final String topic = "TASK_FALLBACK_TOPIC";

    public MqFallbackPolicy(DefaultMQProducer producer) {
        this.producer = producer;
    }

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (executor.isShutdown()) return;

        try {
            // 构造消息体（JSON 格式更通用）
            String json = new ObjectMapper().writeValueAsString(Map.of(
                "taskClass", r.getClass().getName(),
                "taskData", r.toString(),
                "timestamp", System.currentTimeMillis()
            ));

            Message msg = new Message(topic, json.getBytes(StandardCharsets.UTF_8));
            producer.send(msg); // 生产者需已启动且有重试配置
            System.out.println("任务已发送至MQ容灾队列");

        } catch (Exception e) {
            System.err.println("MQ发送失败，触发本地降级处理：" + e.getMessage());
            // 降级方案：写日志、发企业微信告警、或 fallback 到 DB 表
        }
    }
}

关键注意事项：