Spring Boot异步任务线程池配置与异常处理实战

Spring Boot中通过@Async注解可以快速实现异步方法调用，但默认配置使用SimpleAsyncTaskExecutor，每次调用都创建新线程，在高频场景下会导致系统资源耗尽。正确配置线程池并处理异步任务中的异常，是保证系统稳定性的关键。

线程池核心参数配置

ThreadPoolExecutor的参数配置需要根据业务场景的IO/CPU密集特性进行调整。配置不当会导致任务排队积压、OOM或吞吐量下降。

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    @Override
    @Bean("taskExecutor")
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心线程数：CPU密集型设为CPU核心数，IO密集型设为2*CPU核心数
        executor.setCorePoolSize(10);
        // 最大线程数：根据下游服务承载能力设置上限
        executor.setMaxPoolSize(50);
        // 队列容量：根据内存和容忍延迟设置，过大导致OOM和长延迟
        executor.setQueueCapacity(200);
        // 线程空闲超时：超过核心线程数的线程空闲60秒后回收
        executor.setKeepAliveSeconds(60);
        // 线程名前缀：便于日志排查
        executor.setThreadNamePrefix("async-task-");
        // 拒绝策略：
        // AbortPolicy(默认): 抛出RejectedExecutionException
        // CallerRunsPolicy: 由调用线程执行，起到限流作用
        // DiscardPolicy: 静默丢弃
        // DiscardOldestPolicy: 丢弃队列头部最老的任务
        executor.setRejectedExecutionHandler(
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
        // 等待所有任务完成再关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        // 最多等待30秒
        executor.setAwaitTerminationSeconds(30);
        executor.initialize();
        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new CustomAsyncExceptionHandler();
    }
}

拒绝策略的选择需要根据业务容忍度决定。CallerRunsPolicy是最安全的选择——当线程池和队列都满时，由提交任务的线程自己执行，自然降低提交速度起到限流效果。AbortPolicy适合不允许丢弃任务的场景，但需要调用方捕获异常并做重试。DiscardPolicy和DiscardOldestPolicy适合可容忍丢失的场景如日志上报、监控打点。

异步方法的正确使用方式

@Async注解的使用有一个常见陷阱：同类内部方法调用不会触发异步执行。因为Spring AOP基于代理实现，内部调用不经过代理对象。解决方法是将异步方法拆分到独立类中，或通过AopContext.currentProxy()获取代理对象调用。

// 错误示例：内部调用不生效
@Service
public class OrderService {

    public void processOrder(Long orderId) {
        // 直接调用同类方法，不走代理，实际是同步执行
        this.sendNotification(orderId);
    }

    @Async("taskExecutor")
    public void sendNotification(Long orderId) {
        // 这个方法不会异步执行
        emailSender.send(orderId);
    }
}

// 正确做法：拆分到独立Service
@Service
public class OrderService {

    @Autowired
    private NotificationService notificationService;

    public void processOrder(Long orderId) {
        // 通过注入的代理对象调用，异步生效
        notificationService.sendNotification(orderId);
    }
}

@Service
public class NotificationService {

    @Async("taskExecutor")
    public void sendNotification(Long orderId) {
        emailSender.send(orderId);
    }
}

异步任务异常处理

异步方法抛出的异常不会传播到调用方，@Async方法返回void时异常会被AsyncUncaughtExceptionHandler捕获。返回CompletableFuture时，异常封装在Future中，调用方通过future.get()或future.exceptionally()处理。

// 自定义异步异常处理器
public class CustomAsyncExceptionHandler
        implements AsyncUncaughtExceptionHandler {

    private static final Logger log =
        LoggerFactory.getLogger(CustomAsyncExceptionHandler.class);

    @Override
    public void handleUncaughtException(
            Throwable ex, Method method, Object... params) {

        log.error("异步任务异常 | method={} | params={} | error={}",
            method.getName(),
            Arrays.toString(params),
            ex.getMessage(), ex);

        // 根据异常类型做不同处理
        if (ex instanceof BusinessException) {
            // 业务异常：记录告警，不重试
            alertService.send("业务异常: " + ex.getMessage());
        } else if (ex instanceof HttpClientErrorException) {
            // HTTP调用异常：可考虑重试
            retryService.enqueue(method, params);
        } else {
            // 未知异常：触发告警
            alertService.send("异步任务未知异常: " + ex.getMessage());
        }
    }
}

// 返回CompletableFuture的异步方法
@Async("taskExecutor")
public CompletableFuture<Result> fetchDataAsync(String query) {
    try {
        Result result = apiClient.query(query);
        return CompletableFuture.completedFuture(result);
    } catch (Exception e) {
        CompletableFuture<Result> future = new CompletableFuture<>();
        future.completeExceptionally(e);
        return future;
    }
}

// 调用方处理异常
fetchDataAsync("test")
    .thenAccept(result -> handleResult(result))
    .exceptionally(ex -> {
        log.error("异步查询失败", ex);
        return null;
    });

线程池监控与动态调参

生产环境中线程池的运行状态需要持续监控。通过Spring Actuator暴露线程池指标，结合Prometheus+Grafana实现可视化监控。

@Component
public class ThreadPoolMonitor {

    private final ThreadPoolTaskExecutor executor;

    public ThreadPoolMonitor(@Qualifier("taskExecutor")
                             ThreadPoolTaskExecutor executor) {
        this.executor = executor;
    }

    // 每10秒采集一次线程池指标
    @Scheduled(fixedRate = 10000)
    public void collectMetrics() {
        ThreadPoolExecutor pool = executor.getThreadPoolExecutor();
        log.info(
            "线程池状态 | active={} | poolSize={} | " +
            "queueSize={} | completedTaskCount={} | " +
            "taskCount={}",
            pool.getActiveCount(),
            pool.getPoolSize(),
            pool.getQueue().size(),
            pool.getCompletedTaskCount(),
            pool.getTaskCount()
        );
    }
}

// 暴露指标到Micrometer
@Component
public class ThreadPoolMetrics {

    private final ThreadPoolTaskExecutor executor;

    public ThreadPoolMetrics(@Qualifier("taskExecutor")
                             ThreadPoolTaskExecutor executor) {
        this.executor = executor;
    }

    @Bean
    public MeterBinder threadPoolMetrics() {
        return registry -> {
            ThreadPoolExecutor pool = executor.getThreadPoolExecutor();
            Gauge.builder("async.pool.active", pool,
                ThreadPoolExecutor::getActiveCount)
                .description("活跃线程数")
                .register(registry);
            Gauge.builder("async.pool.queue.size", pool,
                p -> p.getQueue().size())
                .description("队列积压任务数")
                .register(registry);
            Gauge.builder("async.pool.completed", pool,
                ThreadPoolExecutor::getCompletedTaskCount)
                .description("已完成任务数")
                .register(registry);
        };
    }
}

当队列积压持续增长时，说明线程池处理能力不足。动态调参可以通过设置允许运行时修改corePoolSize来实现，ThreadPoolExecutor提供了setCorePoolSize和setMaximumPoolSize方法，可以在不重建线程池的情况下调整参数。

@RestController
@RequestMapping("/admin/threadpool")
public class ThreadPoolController {

    @Autowired
    @Qualifier("taskExecutor")
    private ThreadPoolTaskExecutor executor;

    @PutMapping("/resize")
    public Map<String, Object> resize(
            @RequestParam int coreSize,
            @RequestParam int maxSize) {

        ThreadPoolExecutor pool = executor.getThreadPoolExecutor();
        pool.setCorePoolSize(coreSize);
        pool.setMaximumPoolSize(maxSize);

        return Map.of(
            "corePoolSize", pool.getCorePoolSize(),
            "maxPoolSize", pool.getMaximumPoolSize(),
            "activeCount", pool.getActiveCount(),
            "queueSize", pool.getQueue().size()
        );
    }
}

动态调参接口需要做好权限控制，防止误操作将线程池大小设置为0导致所有异步任务被拒绝。微服务架构中，多个服务共用一套线程池配置规范，通过配置中心统一管理各服务的线程池参数，避免各服务自行调参导致不一致。消息中间件消费端的线程池配置尤其需要注意——消费者线程数不应超过下游服务的最大并发处理能力，否则在流量高峰时会压垮下游服务。