3、异步任务与线程池：@Async注解、TaskExecutor配置、异步异常处理策略

在分布式任务调度与DAG边界架构中，异步执行是解耦任务依赖、提升吞吐量的核心手段。Spring Boot通过@Async注解与TaskExecutor抽象，为开发者提供了声明式异步编程能力。本章深入剖析其底层原理、配置策略与异常处理机制。

3.1 @Async注解：声明式异步的底层原理

@Async注解的本质是Spring AOP代理机制。当Bean被标注@Async后，Spring容器会为其创建代理对象，拦截方法调用并提交至线程池异步执行。

3.2 TaskExecutor配置：线程池的精细化管理

Spring Boot默认使用SimpleAsyncTaskExecutor，该实现每次执行都创建新线程，不适合生产环境。推荐替换为ThreadPoolTaskExecutor，并配置核心参数。

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(8);
        executor.setMaxPoolSize(16);
        executor.setQueueCapacity(1000);
        executor.setKeepAliveSeconds(120);
        executor.setThreadNamePrefix("dag-async-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }

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

3.3 异步异常处理策略

异步任务的异常处理与同步调用有本质区别。异常不会直接传播到调用者线程，必须通过专门机制捕获。

3.3.1 异常处理分类

3.3.2 自定义异常处理器

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.getName(), Arrays.toString(params), ex.getMessage(), ex);
        // 可集成告警、重试队列等
    }
}

3.3.3 CompletableFuture链式异常处理

@Async
public CompletableFuture<String> processTask(String input) {
    if (input == null) {
        throw new IllegalArgumentException("输入不能为空");
    }
    return CompletableFuture.completedFuture("处理结果: " + input);
}

// 调用方
CompletableFuture<String> future = service.processTask(null);
future.exceptionally(ex -> {
    log.error("异步任务异常: {}", ex.getMessage());
    return "默认兜底结果";
}).thenAccept(result -> log.info("最终结果: {}", result));

3.4 生产级配置清单

• 线程池隔离：不同业务类型（IO密集型、CPU密集型）使用独立线程池，避免相互影响。
• 监控埋点：通过ThreadPoolTaskExecutor的getPoolSize()、getQueueSize()等指标暴露给Prometheus。
• 优雅关闭：在@PreDestroy中调用executor.shutdown()，等待已提交任务完成。
• 超时控制：使用CompletableFuture.orTimeout()设置任务超时，避免线程池阻塞。