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

@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());
            }
        }
    }
}

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));
    }
}

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();
    }
}

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    50, 200, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(1000),
    new PersistenceRejectPolicy(taskRepository)
);

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10, 50, 30, TimeUnit.SECONDS,
    new PriorityBlockingQueue<>(100),
    new MqRejectPolicy(kafkaTemplate)
);