Java线程池拒绝策略原理及任务不丢失方案总结(最近实践)_Java

一、线程池拒绝策略的核心机制

java线程池（threadpoolexecutor）的拒绝策略在以下条件下触发：

线程池已满：
- 活跃线程数 ≥ maximumpoolsize。
- 任务队列（workqueue）已满（若为有界队列）。
线程池关闭：
- 调用shutdown()后提交新任务。

触发流程：

提交任务时，线程池通过execute()方法检查状态。
若线程池无法接受任务（如满载或关闭），调用rejectedexecutionhandler.rejectedexecution()。

二、四种内置拒绝策略及适用场景

策略	行为	适用场景	风险
abortpolicy	抛出`rejectedexecutionexception`	关键任务（如支付）	调用方需处理异常
callerrunspolicy	由提交任务的线程直接执行任务	非关键但需保证执行（如日志上报）	可能阻塞调用线程
discardpolicy	静默丢弃任务	可丢失任务（如监控数据）	任务丢失风险
discardoldestpolicy	丢弃队列中最旧任务，重试提交新任务	实时性要求高（如股票行情）	可能丢失重要任务

三、确保任务不丢失的解决方案

1. 自定义拒绝策略 + 持久化存储

核心思想：将拒绝的任务保存到外部存储（数据库/消息队列/文件），后续通过重试机制恢复执行。

实现步骤：

定义持久化任务实体：

@data
public class persistedtask {
    private string id;
    private string taskdata; // 序列化后的任务
    private int retrycount;
    private localdatetime createtime;
}

自定义拒绝策略：

public class persistencerejectpolicy implements rejectedexecutionhandler {
    private final taskrepository taskrepository;
    @override
    public void rejectedexecution(runnable r, threadpoolexecutor executor) {
        try {
            string taskdata = serializetask(r);
            taskrepository.save(new persistedtask(uuid.randomuuid().tostring(), taskdata, 0));
        } catch (exception e) {
            log.error("持久化任务失败", e);
        }
    }
    private string serializetask(runnable r) {
        return new gson().tojson(r);
    }
}

定时任务重试：

@scheduled(fixedrate = 5000)
public void retryrejectedtasks() {
    list<persistedtask> tasks = taskrepository.findpendingtasks();
    for (persistedtask task : tasks) {
        try {
            runnable r = deserializetask(task.gettaskdata());
            executor.execute(r);
            taskrepository.markascompleted(task.getid());
        } catch (exception e) {
            if (task.getretrycount() >= 3) {
                taskrepository.markasfailed(task.getid());
            } else {
                taskrepository.incrementretry(task.getid());
            }
        }
    }
}

2. 结合消息队列的异步处理

优势：解耦任务提交与执行，利用消息队列的持久化能力。

实现：

public class mqrejectpolicy implements rejectedexecutionhandler {
    private final kafkatemplate<string, string> kafkatemplate;
    @override
    public void rejectedexecution(runnable r, threadpoolexecutor executor) {
        kafkatemplate.send("rejected-tasks", serializetask(r));
    }
}

3. 线程池参数优化

队列选择：
- 有界队列（如arrayblockingqueue）：需合理设置容量（例如queuecapacity = maxthreads * 2）。
- 优先级队列（priorityblockingqueue）：适合需要排序的任务。
线程数配置：
- cpu密集型：corepoolsize = cpu核心数。
- io密集型：corepoolsize = 2 * cpu核心数。

4. 优雅关闭与任务完整性

public void shutdowngracefully(threadpoolexecutor executor) {
    executor.shutdown(); // 拒绝新任务
    try {
        if (!executor.awaittermination(60, timeunit.seconds)) {
            list<runnable> pendingtasks = executor.shutdownnow(); // 尝试停止正在执行的任务
            savependingtasks(pendingtasks); // 持久化未执行任务
        }
    } catch (interruptedexception e) {
        thread.currentthread().interrupt();
    }
}

四、关键场景实践

场景1：高并发订单处理

策略：abortpolicy + 数据库持久化 + 熔断机制。
配置：

threadpoolexecutor executor = new threadpoolexecutor(
    50, 200, 60, timeunit.seconds,
    new arrayblockingqueue<>(1000),
    new persistencerejectpolicy(taskrepository)
);

降级：当数据库连接池满时，触发熔断直接丢弃非关键订单。

场景2：实时数据分析

策略：discardoldestpolicy + kafka缓冲。
配置：

threadpoolexecutor executor = new threadpoolexecutor(
    10, 50, 30, timeunit.seconds,
    new priorityblockingqueue<>(100),
    new mqrejectpolicy(kafkatemplate)
);

五、风险与优化建议

持久化性能瓶颈：
- 解决方案：批量插入数据库，或使用redis list暂存任务。
任务重复执行：
- 解决方案：为任务添加唯一id，执行前检查是否已处理。
内存泄漏：
- 解决方案：定期清理emergencyqueue中的积压任务。
监控缺失：
- 解决方案：通过micrometer暴露以下指标：
  - threadpool.rejected.count：拒绝任务数。
  - threadpool.queue.size：队列堆积情况。