一、线程池满的痛点

上周，一位做电商系统的朋友向我求助：他们的订单履约系统在高峰期经常丢失订单，导致大量用户投诉。

"我们使用了异步任务来处理订单，"朋友焦急地说，"但高峰期总是有任务被丢弃，订单状态一直处于'处理中'，用户频繁投诉。"

我查看了他们的代码，发现问题确实很严重：

• 系统使用 @Async 注解处理异步任务
• 线程池核心线程数 10，最大线程数 20
• 使用默认的 AbortPolicy 拒绝策略
• 高峰期任务提交速度超过处理速度
• 被拒绝的任务直接被丢弃，没有任何记录

更关键的是，他们根本不知道有多少任务被丢弃，是被丢弃了还是正在排队？

二、传统方案的局限性

1. 默认 AbortPolicy

当线程池满了，新任务会被直接拒绝并抛出 RejectedExecutionException。

@Bean(name = "taskExecutor")
public Executor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(10);
    executor.setMaxPoolSize(20);
    executor.setQueueCapacity(100);
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
    executor.initialize();
    return executor;
}

这种方案的问题：

• 任务丢失：被拒绝的任务直接丢失，没有补救措施
• 异常处理复杂：调用方需要捕获异常，增加复杂度
• 无法追溯：丢失的任务没有任何记录，难以排查
• 用户体验差：用户不知道任务是否被处理

2. CallerRunsPolicy

由调用线程执行被拒绝的任务。

executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

这种方案的问题：

• 可能拖慢主线程：调用线程被阻塞，影响主业务流程
• 线程安全风险：调用线程可能不是线程安全的
• 不可预测：性能变得不可预测
• 可能导致级联故障：主线程被阻塞，可能导致更多问题

3. DiscardPolicy

默默地丢弃被拒绝的任务，不抛出异常。

executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());

这种方案的问题：

• 任务静默丢失：任务被丢弃，没有任何记录
• 难以排查问题：不知道有多少任务被丢弃
• 可能遗漏重要任务：重要任务被丢弃后没有任何补救
• 不适合生产环境：无法接受的任务丢失

4. DiscardOldestPolicy

丢弃队列中最旧的任务，然后重试被拒绝的任务。

executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());

这种方案的问题：

• 新任务可能更重要：丢弃旧任务可能丢失重要信息
• 任务执行顺序不确定：执行顺序变得不可预测
• 仍然可能丢任务：如果新任务一直被拒绝，可能持续丢任务
• 不适合有顺序要求的场景：订单处理等场景不适用

三、终极方案：自定义拒绝策略 + 任务持久化

今天，我要和大家分享一个在实战中验证过的解决方案：自定义拒绝策略 + 任务持久化。

这套方案的核心思想是：

1. 自定义拒绝策略：当任务被拒绝时，不丢弃任务，而是持久化到数据库
2. 异步重试机制：后台线程定期从数据库读取待执行任务，重新提交到线程池
3. 任务状态追踪：记录任务的完整生命周期，包括提交、执行、完成、失败状态
4. 多维度告警：当待执行任务积压过多时，发送告警通知

四、方案详解

1. 核心原理

自定义拒绝策略的工作流程如下：

任务提交到线程池
    ↓
线程池已满，队列已满
    ↓
触发自定义拒绝策略
    ↓
将任务信息序列化为JSON
    ↓
持久化到数据库（待执行任务表）
    ↓
返回提交成功（不抛异常）
    ↓
后台重试线程定期扫描待执行任务
    ↓
将任务重新提交到线程池
    ↓
执行成功后更新任务状态

2. SpringBoot实现

（1）异步任务实体类

@Entity
@Table(name = "async_task")
@Data
public class AsyncTask {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(name = "task_type", nullable = false)
    private String taskType;

    @Column(name = "task_data", columnDefinition = "TEXT")
    private String taskData;

    @Column(name = "status", nullable = false)
    private Integer status;

    @Column(name = "retry_count")
    private Integer retryCount = 0;

    @Column(name = "max_retry")
    private Integer maxRetry = 3;

    @Column(name = "error_message", columnDefinition = "TEXT")
    private String errorMessage;

    @Column(name = "create_time")
    private LocalDateTime createTime;

    @Column(name = "update_time")
    private LocalDateTime updateTime;

    @Column(name = "execute_time")
    private LocalDateTime executeTime;

    @Column(name = "complete_time")
    private LocalDateTime completeTime;
}

（2）任务状态枚举

public enum TaskStatus {
    PENDING(0, "待执行"),
    RUNNING(1, "执行中"),
    COMPLETED(2, "已完成"),
    FAILED(3, "执行失败"),
    DISCARDED(4, "已丢弃");

    private final int code;
    private final String desc;

    TaskStatus(int code, String desc) {
        this.code = code;
        this.desc = desc;
    }
}

（3）任务持久化服务

@Service
@Slf4j
public class AsyncTaskPersistenceService {

    @Autowired
    private AsyncTaskRepository asyncTaskRepository;

    @Autowired
    private ObjectMapper objectMapper;

    public Long saveTask(String taskType, Runnable task, Map<String, Object> params) {
        try {
            AsyncTask asyncTask = new AsyncTask();
            asyncTask.setTaskType(taskType);
            asyncTask.setStatus(TaskStatus.PENDING.getCode());
            asyncTask.setTaskData(objectMapper.writeValueAsString(params));
            asyncTask.setRetryCount(0);
            asyncTask.setMaxRetry(3);
            asyncTask.setCreateTime(LocalDateTime.now());
            asyncTask.setUpdateTime(LocalDateTime.now());

            AsyncTask saved = asyncTaskRepository.save(asyncTask);
            log.info("Task persisted: type={}, id={}", taskType, saved.getId());
            return saved.getId();
        } catch (Exception e) {
            log.error("Failed to persist task", e);
            return null;
        }
    }

    public List<AsyncTask> getPendingTasks(int limit) {
        return asyncTaskRepository.findByStatusOrderByCreateTimeAsc(
            TaskStatus.PENDING.getCode(),
            PageRequest.of(0, limit)
        );
    }

    public void updateTaskStatus(Long taskId, TaskStatus status) {
        AsyncTask task = asyncTaskRepository.findById(taskId).orElse(null);
        if (task != null) {
            task.setStatus(status.getCode());
            task.setUpdateTime(LocalDateTime.now());
            if (status == TaskStatus.RUNNING) {
                task.setExecuteTime(LocalDateTime.now());
            } else if (status == TaskStatus.COMPLETED) {
                task.setCompleteTime(LocalDateTime.now());
            }
            asyncTaskRepository.save(task);
        }
    }

    public void incrementRetryCount(Long taskId) {
        AsyncTask task = asyncTaskRepository.findById(taskId).orElse(null);
        if (task != null) {
            task.setRetryCount(task.getRetryCount() + 1);
            task.setUpdateTime(LocalDateTime.now());
            asyncTaskRepository.save(task);
        }
    }

    public void markAsFailed(Long taskId, String errorMessage) {
        AsyncTask task = asyncTaskRepository.findById(taskId).orElse(null);
        if (task != null) {
            task.setStatus(TaskStatus.FAILED.getCode());
            task.setErrorMessage(errorMessage);
            task.setUpdateTime(LocalDateTime.now());
            asyncTaskRepository.save(task);
            log.warn("Task marked as failed: id={}, error={}", taskId, errorMessage);
        }
    }
}

（4）自定义拒绝策略

@Component
public class PersistenceRejectedExecutionHandler implements RejectedExecutionHandler {

    @Autowired
    private AsyncTaskPersistenceService persistenceService;

    @Autowired
    private AsyncTaskExecutorWrapper executorWrapper;

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (r instanceof PersistableRunnable) {
            PersistableRunnable persistableRunnable = (PersistableRunnable) r;

            Long taskId = persistenceService.saveTask(
                persistableRunnable.getTaskType(),
                r,
                persistableRunnable.getParams()
            );

            if (taskId != null) {
                log.warn("Task rejected and persisted: type={}, id={}",
                        persistableRunnable.getTaskType(), taskId);
            } else {
                log.error("Failed to persist rejected task: type={}",
                        persistableRunnable.getTaskType());
            }
        } else {
            log.error("Rejected task is not PersistableRunnable, task will be lost");
        }
    }
}

（5）可持久化任务包装器

@Data
public class PersistableRunnable implements Runnable {
    private final String taskType;
    private final Runnable originalTask;
    private final Map<String, Object> params;
    private final Long taskId;

    public PersistableRunnable(String taskType, Runnable originalTask,
                             Map<String, Object> params, Long taskId) {
        this.taskType = taskType;
        this.originalTask = originalTask;
        this.params = params;
        this.taskId = taskId;
    }

    @Override
    public void run() {
        try {
            originalTask.run();
        } catch (Exception e) {
            log.error("Task execution error: id={}", taskId, e);
            throw e;
        }
    }
}

（6）任务执行器包装器

@Component
public class AsyncTaskExecutorWrapper {

    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;

    @Autowired
    private AsyncTaskPersistenceService persistenceService;

    public void execute(String taskType, Runnable task, Map<String, Object> params) {
        PersistableRunnable persistableTask = new PersistableRunnable(
            taskType, task, params, null
        );
        taskExecutor.execute(persistableTask);
    }

    public void executeWithPersistence(String taskType, Runnable task,
                                      Map<String, Object> params, Long taskId) {
        PersistableRunnable persistableTask = new PersistableRunnable(
            taskType, task, params, taskId
        );
        taskExecutor.execute(persistableTask);
    }
}

（7）线程池配置

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean(name = "taskExecutor")
    public ThreadPoolTaskExecutor taskExecutor(
            PersistenceRejectedExecutionHandler rejectionHandler) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("async-");
        executor.setRejectedExecutionHandler(rejectionHandler);
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize();
        return executor;
    }
}

（8）任务重试服务

@Service
@Slf4j
public class AsyncTaskRetryService {

    @Autowired
    private AsyncTaskPersistenceService persistenceService;

    @Autowired
    private AsyncTaskExecutorWrapper executorWrapper;

    @Autowired
    private AlertService alertService;

    private static final int BATCH_SIZE = 50;
    private static final int MAX_RETRY = 3;
    private static final long RETRY_INTERVAL_MS = 5000;

    @Scheduled(fixedDelay = 5000)
    public void retryPendingTasks() {
        List<AsyncTask> pendingTasks = persistenceService.getPendingTasks(BATCH_SIZE);

        if (pendingTasks.isEmpty()) {
            return;
        }

        log.info("Found {} pending tasks to retry", pendingTasks.size());

        for (AsyncTask task : pendingTasks) {
            if (task.getRetryCount() >= MAX_RETRY) {
                persistenceService.markAsFailed(task.getId(),
                    "Exceeded max retry count: " + MAX_RETRY);
                continue;
            }

            try {
                executorWrapper.executeWithPersistence(
                    task.getTaskType(),
                    () -> executeTask(task),
                    parseTaskData(task.getTaskData()),
                    task.getId()
                );
                persistenceService.incrementRetryCount(task.getId());
            } catch (Exception e) {
                log.error("Failed to retry task: id={}", task.getId(), e);
            }
        }

        checkBacklogAlert(pendingTasks.size());
    }

    private void executeTask(AsyncTask task) {
        persistenceService.updateTaskStatus(task.getId(), TaskStatus.RUNNING);
        log.info("Executing persisted task: id={}, type={}", task.getId(), task.getTaskType());
    }

    private Map<String, Object> parseTaskData(String taskData) {
        if (taskData == null || taskData.isEmpty()) {
            return new HashMap<>();
        }
        try {
            return new ObjectMapper().readValue(taskData,
                new TypeReference<Map<String, Object>>() {});
        } catch (Exception e) {
            log.error("Failed to parse task data", e);
            return new HashMap<>();
        }
    }

    private void checkBacklogAlert(int pendingCount) {
        if (pendingCount > 100) {
            alertService.sendAlert("Async Task Backlog",
                String.format("Pending tasks count: %d", pendingCount));
        }
    }
}

（9）告警服务

@Service
@Slf4j
public class AlertService {

    @Value("${alert.enabled:false}")
    private boolean enabled;

    @Value("${alert.webhook:}")
    private String webhook;

    public void sendAlert(String subject, String content) {
        if (!enabled) {
            return;
        }
        log.warn("ALERT - {}: {}", subject, content);
    }
}

3. 使用示例

（1）定义异步任务

@Service
public class OrderService {

    @Autowired
    private AsyncTaskExecutorWrapper executorWrapper;

    public void processOrder(Long orderId) {
        Map<String, Object> params = new HashMap<>();
        params.put("orderId", orderId);

        executorWrapper.execute("order_process", () -> {
            doProcessOrder(orderId);
        }, params);
    }

    private void doProcessOrder(Long orderId) {
        // 订单处理逻辑
    }
}

（2）数据库表结构

CREATE TABLE `async_task` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `task_type` varchar(50) NOT NULL COMMENT '任务类型',
  `task_data` text COMMENT '任务参数JSON',
  `status` tinyint(4) NOT NULL DEFAULT '0' COMMENT '状态：0-待执行，1-执行中，2-已完成，3-执行失败，4-已丢弃',
  `retry_count` int(11) DEFAULT '0' COMMENT '重试次数',
  `max_retry` int(11) DEFAULT '3' COMMENT '最大重试次数',
  `error_message` text COMMENT '错误信息',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  `execute_time` datetime DEFAULT NULL COMMENT '执行时间',
  `complete_time` datetime DEFAULT NULL COMMENT '完成时间',
  PRIMARY KEY (`id`),
  KEY `idx_status` (`status`),
  KEY `idx_create_time` (`create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='异步任务表';

五、性能对比

1. 测试场景

• 线程池核心线程数：10
• 线程池最大线程数：20
• 队列容量：100
• 任务提交速率：500/秒
• 任务处理速率：200/秒

2. 测试结果

六、最佳实践

1. 线程池配置

• 合理设置队列容量：队列容量不宜过大，会导致内存占用过高
• 设置合理的线程数：根据任务特性和服务器资源合理配置
• 启用拒绝策略：一定要设置自定义的拒绝策略
• 优雅关闭：设置 waitForTasksToCompleteOnShutdown

2. 任务持久化

• 选择合适的存储：根据任务量选择数据库或消息队列
• 定期清理：清理已完成的任务，避免数据膨胀
• 监控积压：监控待执行任务数量，及时告警
• 设置重试上限：避免无限重试，消耗资源

3. 监控告警

• 监控队列长度：队列积压时及时告警
• 监控任务失败率：任务失败率过高时告警
• 监控重试次数：重试次数过多时告警
• 定期巡检：定期检查任务执行情况

4. 异常处理

• 任务内异常捕获：在任务内部捕获异常，避免任务直接失败
• 重试机制：支持任务重试，提高成功率
• 死信队列：重试多次仍失败的任务，进入死信队列
• 日志记录：详细记录任务执行日志，便于排查

七、总结与展望

方案总结

1. 任务不丢失：通过持久化确保任务不会丢失
2. 自动重试：后台线程自动重试被拒绝的任务
3. 状态追踪：记录任务的完整生命周期
4. 多维度告警：积压过多时及时告警
5. 灵活配置：支持自定义重试次数、超时时间等
6. 可扩展：易于扩展，支持更多的任务类型和存储方式

未来优化方向

1. 智能重试：根据任务类型和失败原因，智能调整重试策略
2. 优先级队列：支持任务优先级，确保重要任务优先执行
3. 分布式支持：支持多实例部署，任务统一管理
4. 可视化界面：提供Web界面，直观展示任务执行情况

技术价值

1. 保证任务不丢失：通过持久化机制，确保任务不会丢失
2. 提高系统稳定性：系统在高负载时仍能正常运行
3. 便于排查问题：详细的日志和状态记录，便于问题排查
4. 提升用户体验：用户提交的任务都能被正确处理

八、写在最后

异步任务线程池满是一个常见的问题，但通过自定义拒绝策略 + 任务持久化方案，我们可以在保证任务不丢失的同时，提供良好的用户体验和可追溯性。

当然，这套方案也不是银弹，它有以下局限性：

• 增加数据库压力：需要将任务持久化到数据库，增加数据库压力
• 实时性降低：被拒绝的任务需要等待重试，实时性略有降低
• 复杂度增加：系统复杂度增加，需要维护更多的组件

但对于需要保证任务不丢失的业务场景，这套方案已经足够解决问题，而且稳定可靠。

希望这篇文章能给你带来一些启发，帮助你在实际项目中更好地处理异步任务的线程池拒绝问题。

如果你在使用这套方案的过程中有其他经验或困惑，欢迎在评论区留言交流！

服务端技术精选，专注分享后端开发实战经验，让技术落地更简单。

如果你觉得这篇文章有用，欢迎点赞、在看、分享三连！