一、异步编程基础概念

1.1 同步 vs 异步

通俗理解：同步就像在银行柜台排队办理业务，必须等前面的人办完才能轮到你；异步则像取号后可以坐着玩手机，等叫号时再去办理。

1.2 为什么要使用异步

• 提高吞吐量：服务器能同时处理更多请求
• 增强用户体验：避免用户长时间等待
• 资源优化：合理利用系统资源，避免阻塞主线程
• 解耦：将耗时操作与主流程分离

1.3 Java中的异步编程方式

// 1. 传统Thread方式
new Thread(() -> {
    // 异步任务
}).start();

// 2. Future模式
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<String> future = executor.submit(() -> {
    Thread.sleep(1000);
    return "Result";
});

// 3. CompletableFuture (Java8+)
CompletableFuture.supplyAsync(() -> {
    // 异步任务
    return "Result";
}).thenAccept(result -> {
    // 处理结果
});

// 4. Spring @Async (本文重点)
@Async
public void asyncMethod() {
    // 异步方法体
}

二、@Async基础使用

2.1 启用@Async支持

步骤1：在Spring Boot主类或配置类上添加@EnableAsync

@SpringBootApplication
@EnableAsync  // 启用异步支持
public class AsyncApplication {
    public static void main(String[] args) {
        SpringApplication.run(AsyncApplication.class, args);
    }
}

步骤2：创建异步服务类

@Service
public class EmailService {
    
    // 无返回值异步方法
    @Async
    public void sendEmail(String to, String content) {
        // 模拟邮件发送耗时
        try {
            Thread.sleep(3000);
            System.out.println("邮件已发送至: " + to + ", 内容: " + content);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    // 有返回值异步方法
    @Async
    public Future<String> sendEmailWithResult(String to, String content) {
        try {
            Thread.sleep(3000);
            String result = "邮件已发送至: " + to;
            return new AsyncResult<>(result);
        } catch (InterruptedException e) {
            return new AsyncResult<>("发送失败");
        }
    }
}

2.2 调用异步方法

@RestController
@RequestMapping("/api/email")
public class EmailController {
    
    @Autowired
    private EmailService emailService;
    
    @GetMapping("/send")
    public String sendEmail() {
        long start = System.currentTimeMillis();
        
        // 调用异步方法
        emailService.sendEmail("user@example.com", "您的订单已创建");
        
        long elapsed = System.currentTimeMillis() - start;
        return "请求已处理,耗时: " + elapsed + "ms"; // 立即返回，不会等待邮件发送完成
    }
    
    @GetMapping("/send-with-result")
    public String sendEmailWithResult() throws Exception {
        Future<String> future = emailService.sendEmailWithResult("user@example.com", "订单详情");
        
        // 可以在这里做其他事情
        
        // 当需要结果时（阻塞等待）
        String result = future.get();
        return "处理结果: " + result;
    }
}

2.3 @Async方法限制

0. 必须public修饰：因为基于Spring AOP实现
1. 同类调用无效：this.asyncMethod()不会异步执行
2. 返回值限制：
3. void
4. Future及其子类（如AsyncResult）
5. CompletableFuture (Spring 4.2+)
6. ListenableFuture (Spring 4.2+)

三、线程池配置详解

3.1 默认线程池问题

Spring默认使用SimpleAsyncTaskExecutor，它不重用线程，每次调用都创建新线程，生产环境不推荐使用。

3.2 自定义线程池配置

方式1：配置类方式（推荐）

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
    
    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心线程数：线程池创建时初始化的线程数
        executor.setCorePoolSize(5);
        // 最大线程数：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程
        executor.setMaxPoolSize(10);
        // 缓冲队列：用来缓冲执行任务的队列
        executor.setQueueCapacity(50);
        // 线程名前缀
        executor.setThreadNamePrefix("Async-Executor-");
        // 线程池关闭时等待所有任务完成
        executor.setWaitForTasksToCompleteOnShutdown(true);
        // 等待时间
        executor.setAwaitTerminationSeconds(60);
        // 拒绝策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
    
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new CustomAsyncExceptionHandler();
    }
}

// 自定义异常处理器
public class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
    @Override
    public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        System.err.println("异步任务异常 - 方法: " + method.getName());
        System.err.println("异常信息: " + ex.getMessage());
        // 这里可以添加自定义处理逻辑，如记录日志、发送告警等
    }
}

方式2：使用@Bean定义多个线程池

@Configuration
public class TaskExecutorConfig {
    
    @Bean(name = "emailExecutor")
    public Executor emailTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(3);
        executor.setMaxPoolSize(5);
        executor.setQueueCapacity(30);
        executor.setThreadNamePrefix("Email-Executor-");
        executor.initialize();
        return executor;
    }
    
    @Bean(name = "reportExecutor")
    public Executor reportTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(4);
        executor.setQueueCapacity(20);
        executor.setThreadNamePrefix("Report-Executor-");
        executor.initialize();
        return executor;
    }
}

使用指定线程池：

@Async("emailExecutor")
public void sendEmail(String to) { /*...*/ }

@Async("reportExecutor")
public void generateReport() { /*...*/ }

3.3 线程池参数详解

拒绝策略选项：

• AbortPolicy：默认，直接抛出RejectedExecutionException
• CallerRunsPolicy：由调用者线程执行该任务
• DiscardPolicy：直接丢弃任务
• DiscardOldestPolicy：丢弃队列中最老的任务并重试

3.4 线程池监控

@Service
public class ThreadPoolMonitor {
    
    @Autowired
    private ThreadPoolTaskExecutor emailExecutor;
    
    @Scheduled(fixedRate = 5000)  // 每5秒监控一次
    public void monitor() {
        System.out.println("=== 线程池监控 ===");
        System.out.println("当前线程数: " + emailExecutor.getPoolSize());
        System.out.println("活跃线程数: " + emailExecutor.getActiveCount());
        System.out.println("完成任务数: " + emailExecutor.getCompletedTaskCount());
        System.out.println("队列剩余容量: " + emailExecutor.getThreadPoolExecutor().getQueue().remainingCapacity());
    }
}

四、@Async高级特性

4.1 返回值处理

1. Future模式

@Async
public Future<String> processData(String input) {
    // 模拟处理耗时
    try {
        Thread.sleep(2000);
        return new AsyncResult<>("处理完成: " + input.toUpperCase());
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        return new AsyncResult<>("处理中断");
    }
}

// 调用方
Future<String> future = service.processData("hello");
// 可以做其他事情...
String result = future.get(3, TimeUnit.SECONDS); // 带超时的等待

2. CompletableFuture (Java8+)

@Async
public CompletableFuture<String> fetchData(String param) {
    return CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(1000);
            return "Data for " + param;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    });
}

// 链式调用
service.fetchData("user123")
       .thenApply(String::toUpperCase)
       .thenAccept(System.out::println)
       .exceptionally(ex -> {
           System.err.println("Error: " + ex.getMessage());
           return null;
       });

4.2 基于条件的异步执行

1. 使用Spring Expression Language (SpEL)

@Async("#{systemProperties['async.enabled'] ? 'emailExecutor' : 'syncExecutor'}")
public void conditionalAsync() {
    // 根据系统属性决定使用哪个执行器
}

2. 基于配置的开关

@Async
@ConditionalOnProperty(name = "app.async.enabled", havingValue = "true")
public void configBasedAsync() {
    // 当app.async.enabled=true时才异步执行
}

4.3 事务处理

异步方法与事务的特殊关系：

0. 事务边界：@Async方法会在新线程中执行，与原方法不在同一事务中
1. 传播行为：需要在异步方法上单独声明@Transactional

@Async
@Transactional(propagation = Propagation.REQUIRES_NEW)
public void asyncWithTransaction() {
    // 这个方法会在新事务中执行
    userRepository.save(new User("AsyncUser"));
    // 如果发生异常，只会回滚当前方法内的操作
}

4.4 组合异步操作

场景：需要等待多个异步任务全部完成

@Async
public CompletableFuture<String> fetchUserInfo(String userId) {
    // 模拟获取用户信息
    return CompletableFuture.completedFuture("UserInfo-" + userId);
}

@Async
public CompletableFuture<String> fetchOrderInfo(String userId) {
    // 模拟获取订单信息
    return CompletableFuture.completedFuture("OrderInfo-" + userId);
}

// 组合多个异步任务
public CompletableFuture<Void> combineAsyncTasks(String userId) {
    return CompletableFuture.allOf(
            fetchUserInfo(userId),
            fetchOrderInfo(userId)
    ).thenRun(() -> {
        // 所有任务完成后的处理
        System.out.println("所有异步任务已完成");
    });
}

五、异常处理机制

5.1 异常处理方式对比

5.2 实践示例

1. 全局异常处理器

public class GlobalAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
    private static final Logger logger = LoggerFactory.getLogger(GlobalAsyncExceptionHandler.class);
    
    @Override
    public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        logger.error("异步任务异常 - 方法: {}, 参数: {}", method.getName(), Arrays.toString(params), ex);
        
        // 可以根据异常类型进行不同处理
        if (ex instanceof BusinessException) {
            // 业务异常处理
            sendAlert("业务异常警报: " + ex.getMessage());
        } else if (ex instanceof TimeoutException) {
            // 超时处理
            retryTask(method, params);
        }
    }
    
    private void sendAlert(String message) { /*...*/ }
    private void retryTask(Method method, Object... params) { /*...*/ }
}

2. Future方式的异常处理

@Async
public Future<String> asyncTaskWithException() {
    try {
        // 业务逻辑
        if (someCondition) {
            throw new BusinessException("业务异常");
        }
        return new AsyncResult<>("成功");
    } catch (BusinessException e) {
        return new AsyncResult<>("失败: " + e.getMessage());
    }
}

// 调用方处理
Future<String> future = service.asyncTaskWithException();
try {
    String result = future.get();
    if (result.startsWith("失败")) {
        // 处理失败情况
    }
} catch (ExecutionException e) {
    // 处理执行时异常
}

3. CompletableFuture的异常处理

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

// 调用方处理
service.asyncProcess(null)
       .exceptionally(ex -> {
           System.err.println("发生异常: " + ex.getMessage());
           return "默认返回值";
       })
       .thenAccept(result -> {
           System.out.println("最终结果: " + result);
       });

六、性能优化与最佳实践

6.1 性能优化建议

0. 线程池参数调优
1. 根据业务类型调整线程池大小
2. 监控线程池状态动态调整参数
3. 使用有界队列防止OOM
4. 避免长时间阻塞
5. 异步方法内避免同步阻塞操作
6. 使用带超时的阻塞调用
7. 资源清理
8. 确保异步方法正确释放资源
9. 使用try-with-resources管理资源
10. 上下文传递
11. 注意ThreadLocal变量在异步线程中的传递问题
12. 使用TaskDecorator传递上下文

executor.setTaskDecorator(new ContextCopyingDecorator());

public class ContextCopyingDecorator implements TaskDecorator {
    @Override
    public Runnable decorate(Runnable runnable) {
        // 获取当前线程的上下文
        RequestAttributes context = RequestContextHolder.currentRequestAttributes();
        return () -> {
            try {
                // 在新线程中设置上下文
                RequestContextHolder.setRequestAttributes(context);
                runnable.run();
            } finally {
                RequestContextHolder.resetRequestAttributes();
            }
        };
    }
}

6.2 最佳实践清单

0. 命名规范
1. 异步方法名以Async后缀标识，如sendEmailAsync
2. 线程池按业务命名，如orderTaskExecutor
3. 日志记录
4. 记录异步任务开始/结束时间
5. 为异步线程设置可追踪的上下文ID

@Async
public void asyncWithLogging() {
    String traceId = UUID.randomUUID().toString();
    MDC.put("traceId", traceId);
    try {
        log.info("异步任务开始");
        // 业务逻辑
        log.info("异步任务完成");
    } finally {
        MDC.clear();
    }
}

3. 防御性编程
4. 检查异步方法参数有效性
5. 添加合理的超时控制
6. 资源限制
7. 限制并发异步任务数量
8. 对重要任务设置优先级
9. 监控告警
10. 监控线程池关键指标
11. 设置异常告警阈值

七、与其他技术的整合

7.1 与Spring Retry整合

实现异步任务失败重试：

@Async
@Retryable(value = {RemoteAccessException.class}, 
           maxAttempts = 3, 
           backoff = @Backoff(delay = 1000, multiplier = 2))
public CompletableFuture<String> callExternalService() {
    // 调用可能失败的外部服务
    return CompletableFuture.completedFuture(externalService.call());
}

// 重试全部失败后的处理
@Recover
public CompletableFuture<String> recover(RemoteAccessException e) {
    return CompletableFuture.completedFuture("默认返回值");
}

7.2 与Spring Cache整合

异步缓存更新：

@Async
@CacheEvict(value = "users", key = "#userId")
public void evictUserCacheAsync(String userId) {
    // 异步清理缓存
}

@Async
@CachePut(value = "users", key = "#user.id")
public CompletableFuture<User> updateUserAsync(User user) {
    // 异步更新用户并更新缓存
    return CompletableFuture.completedFuture(userRepository.save(user));
}

7.3 与WebFlux响应式编程对比

八、常见问题与解决方案

8.1 问题排查表

8.2 实战问题案例

案例1：数据库连接泄漏

@Async
public void processData() {
    // 错误示范：未关闭Connection
    Connection conn = dataSource.getConnection();
    // 使用conn...
}

解决方案：

@Async
public void processData() {
    try (Connection conn = dataSource.getConnection()) {
        // 使用conn...
    } catch (SQLException e) {
        // 异常处理
    }
}

案例2：订单超时未支付取消

@Async
@Scheduled(fixedDelay = 60000)  // 每分钟检查一次
public void cancelUnpaidOrders() {
    List<Order> unpaidOrders = orderRepository.findByStatusAndCreateTimeBefore(
        OrderStatus.UNPAID, 
        LocalDateTime.now().minusMinutes(30));
    
    unpaidOrders.forEach(order -> {
        order.setStatus(OrderStatus.CANCELLED);
        orderRepository.save(order);
        notificationService.sendCancelNotice(order);
    });
}

九、总结

9.1 核心要点总结

0. 基础使用：
1. @EnableAsync启用支持
2. @Async标注异步方法
3. 避免同类调用
4. 线程池配置：
5. 生产环境必须自定义线程池
6. 合理设置核心参数
7. 监控线程池状态
8. 异常处理：
9. 区分返回值类型选择处理方式
10. 实现全局异常处理器
11. 日志记录完整上下文
12. 高级特性：
13. 组合多个异步操作
14. 事务边界处理
15. 条件异步执行
16. 最佳实践：
17. 命名规范
18. 防御性编程
19. 资源清理
20. 上下文传递

9.2 完整示例代码

订单处理服务示例：

@Service
public class OrderProcessingService {
    
    private static final Logger logger = LoggerFactory.getLogger(OrderProcessingService.class);
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private PaymentService paymentService;
    
    @Autowired
    private NotificationService notificationService;
    
    @Async("orderTaskExecutor")
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public CompletableFuture<OrderResult> processOrderAsync(Order order) {
        logger.info("开始异步处理订单: {}", order.getId());
        long startTime = System.currentTimeMillis();
        
        try {
            // 1. 保存订单
            Order savedOrder = orderRepository.save(order);
            
            // 2. 处理支付
            PaymentResult paymentResult = paymentService.processPayment(savedOrder);
            if (!paymentResult.isSuccess()) {
                throw new PaymentException("支付处理失败: " + paymentResult.getErrorMessage());
            }
            
            // 3. 更新订单状态
            savedOrder.setStatus(OrderStatus.PAID);
            orderRepository.save(savedOrder);
            
            // 4. 发送通知
            notificationService.sendOrderConfirmation(savedOrder);
            
            long elapsed = System.currentTimeMillis() - startTime;
            logger.info("订单处理完成: {}, 耗时: {}ms", savedOrder.getId(), elapsed);
            
            return CompletableFuture.completedFuture(
                new OrderResult(true, "订单处理成功", savedOrder));
                
        } catch (PaymentException e) {
            logger.error("订单支付异常: {}", order.getId(), e);
            return CompletableFuture.completedFuture(
                new OrderResult(false, e.getMessage(), order));
        } catch (Exception e) {
            logger.error("订单处理未知异常: {}", order.getId(), e);
            return CompletableFuture.failedFuture(e);
        }
    }
    
    // 批量异步处理订单
    @Async("batchOrderExecutor")
    public CompletableFuture<Void> processOrdersBatch(List<Order> orders) {
        List<CompletableFuture<OrderResult>> futures = orders.stream()
            .map(this::processOrderAsync)
            .collect(Collectors.toList());
            
        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .exceptionally(ex -> {
                logger.error("批量处理订单异常", ex);
                return null;
            });
    }
}

// 配置类
@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
    
    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("Order-Async-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize();
        return executor;
    }
    
    @Bean(name = "batchOrderExecutor")
    public Executor batchOrderExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("Batch-Order-");
        executor.initialize();
        return executor;
    }
    
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new OrderAsyncExceptionHandler();
    }
}

// 全局异常处理器
public class OrderAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
    private static final Logger logger = LoggerFactory.getLogger(OrderAsyncExceptionHandler.class);
    
    @Override
    public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        logger.error("异步订单处理异常 - 方法: {}, 参数: {}", method.getName(), Arrays.toString(params), ex);
        
        // 发送告警邮件
        if (ex instanceof CriticalOrderException) {
            sendCriticalAlert(method, ex, params);
        }
    }
    
    private void sendCriticalAlert(Method method, Throwable ex, Object... params) {
        // 实现告警逻辑
    }
}

头条对markdown的文章显示不太友好，想了解更多的可以关注微信公众号：“Eric的技术杂货库”，后期会有更多的干货以及资料下载。