Spring Boot @Async 实现异步的最佳实践

1
@Configuration
2
@EnableAsync
3
public class AsyncConfig implements AsyncConfigurer {
4

5
    public static final String NOTICE_EXECUTOR = "noticeExecutor";
6
    public static final String REPORT_EXECUTOR = "reportExecutor";
7

8
    @Bean(NOTICE_EXECUTOR)
9
    public ThreadPoolTaskExecutor noticeExecutor() {
10
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
11
        // IO 密集型：起步按 线程数 = 核数 * (1 + 等待时间/计算时间) 估算，以压测为准
12
        executor.setCorePoolSize(4);
13
        executor.setMaxPoolSize(8);
14
        // 有界队列是底线，容量 = 预期可容忍的积压量
15
        executor.setQueueCapacity(200);
16
        executor.setKeepAliveSeconds(60);
17
        // 线程名前缀用于日志与 jstack 定位，按业务命名
18
        executor.setThreadNamePrefix("notice-");
19
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
20
        // 优雅停机：等待队列中任务执行完再关池
21
        executor.setWaitForTasksToCompleteOnShutdown(true);
22
        executor.setAwaitTerminationSeconds(30);
23
        executor.setTaskDecorator(new ContextCopyTaskDecorator());
24
        executor.initialize();
25
        return executor;
26
    }
27

28
    @Bean(REPORT_EXECUTOR)
29
    public ThreadPoolTaskExecutor reportExecutor() {
30
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
31
        // CPU 密集型：核数 + 1 起步
32
        executor.setCorePoolSize(Runtime.getRuntime().availableProcessors() + 1);
33
        executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() + 1);
34
        executor.setQueueCapacity(100);
35
        executor.setThreadNamePrefix("report-");
36
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
37
        executor.setWaitForTasksToCompleteOnShutdown(true);
38
        executor.setAwaitTerminationSeconds(30);
39
        executor.setTaskDecorator(new ContextCopyTaskDecorator());
40
        executor.initialize();
41
        return executor;
42
    }
43

44
    /** 未指定执行器的 @Async 方法的默认池 */
45
    @Override
46
    public Executor getAsyncExecutor() {
47
        return noticeExecutor();
48
    }
49

50
    /** void 方法未捕获异常的统一兜底 */
51
    @Override
52
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
53
        return (ex, method, params) ->
54
                LoggerFactory.getLogger("ASYNC_UNCAUGHT")
55
                        .error("async method failed, method={}, params={}",
56
                                method.getDeclaringClass().getSimpleName() + "#" + method.getName(),
57
                                Arrays.toString(params), ex);
58
    }
59
}

1
// 形态一：void。异常不会传给调用方，默认行为是打一行日志后丢弃。
2
// 必须通过 AsyncConfig 里的 AsyncUncaughtExceptionHandler 兜底。
3
@Async("noticeExecutor")
4
public void sendPaidNotice(Long orderId) {
5
    smsClient.send(orderService.loadPhone(orderId), "支付成功");
6
}
7

8
// 形态二：CompletableFuture。异常被捕获进返回值，调用方不消费就等于丢失。
9
@Async("reportExecutor")
10
public CompletableFuture<OrderStats> loadStats(Long orderId) {
11
    return CompletableFuture.completedFuture(reportService.stats(orderId));
12
}
13

14
// 形态三：非 Future 返回类型。代理立即返回 null，方法体在后台照常执行。
15
// 编译器不报错，运行期不报错，是最隐蔽的一类错误。
16
@Async("reportExecutor")
17
public OrderStats loadStatsWrong(Long orderId) {   // 调用方拿到 null
18
    return reportService.stats(orderId);
19
}

1
public class ContextCopyTaskDecorator implements TaskDecorator {
2

3
    @Override
4
    public Runnable decorate(Runnable runnable) {
5
        // 在调用线程捕获快照
6
        Map<String, String> mdc = MDC.getCopyOfContextMap();
7
        SecurityContext securityContext = SecurityContextHolder.getContext();
8
        return () -> {
9
            try {
10
                if (mdc != null) {
11
                    MDC.setContextMap(mdc);
12
                }
13
                if (securityContext != null) {
14
                    SecurityContextHolder.setContext(securityContext);
15
                }
16
                runnable.run();
17
            } finally {
18
                // 池线程复用，必须清理，否则上下文残留污染下一个任务
19
                MDC.clear();
20
                SecurityContextHolder.clearContext();
21
            }
22
        };
23
    }
24
}

1
// 事务内：只发事件，不做任何异步动作
2
@Service
3
public class OrderService {
4

5
    private final ApplicationEventPublisher publisher;
6

7
    @Transactional
8
    public void paySuccess(Long orderId) {
9
        orderMapper.updateStatusPaid(orderId);
10
        publisher.publishEvent(new OrderPaidEvent(orderId));
11
    }
12
}
13

14
// 事务提交后触发；监听器方法内的 @Async 调用把执行挪到工作线程
15
@Component
16
public class OrderPaidListener {
17

18
    private final NoticeService noticeService;
19

20
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
21
    public void onPaid(OrderPaidEvent event) {
22
        noticeService.sendPaidNotice(event.orderId());
23
    }
24
}

1
@Configuration
2
@EnableAsync
3
public class AsyncConfig implements AsyncConfigurer {
4

5
    public static final String NOTICE_EXECUTOR = "noticeExecutor";
6

7
    @Bean(NOTICE_EXECUTOR)
8
    public ThreadPoolTaskExecutor noticeExecutor() {
9
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
10
        executor.setCorePoolSize(4);
11
        executor.setMaxPoolSize(8);
12
        executor.setQueueCapacity(200);
13
        executor.setKeepAliveSeconds(60);
14
        executor.setThreadNamePrefix("notice-");
15
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
16
        executor.setWaitForTasksToCompleteOnShutdown(true);
17
        executor.setAwaitTerminationSeconds(30);
18
        executor.setTaskDecorator(new ContextCopyTaskDecorator());
19
        executor.initialize();
20
        return executor;
21
    }
22

23
    @Override
24
    public Executor getAsyncExecutor() {
25
        return noticeExecutor();
26
    }
27

28
    @Override
29
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
30
        return (ex, method, params) ->
31
                LoggerFactory.getLogger("ASYNC_UNCAUGHT")
32
                        .error("async failed, method={}, params={}", method.getName(),
33
                                Arrays.toString(params), ex);
34
    }
35
}

1
@Service
2
public class NoticeService {
3

4
    private static final Logger log = LoggerFactory.getLogger(NoticeService.class);
5

6
    private final OrderMapper orderMapper;
7
    private final SmsClient smsClient;
8

9
    public NoticeService(OrderMapper orderMapper, SmsClient smsClient) {
10
        this.orderMapper = orderMapper;
11
        this.smsClient = smsClient;
12
    }
13

14
    /** 新线程上执行，@Transactional 开新事务，读写都基于已提交的数据 */
15
    @Async(AsyncConfig.NOTICE_EXECUTOR)
16
    @Transactional
17
    public void sendPaidNotice(Long orderId) {
18
        Order order = orderMapper.selectById(orderId);
19
        if (order == null) {
20
            log.warn("order not found, skip notice, orderId={}", orderId);
21
            return;
22
        }
23
        smsClient.send(order.getPhone(), "订单 " + order.getOrderNo() + " 支付成功");
24
    }
25
}

1
@Component
2
public class OrderPaidListener {
3

4
    private static final Logger log = LoggerFactory.getLogger(OrderPaidListener.class);
5

6
    private final NoticeService noticeService;
7

8
    public OrderPaidListener(NoticeService noticeService) {
9
        this.noticeService = noticeService;
10
    }
11

12
    /** 事务提交后才触发；本方法只转发，不写库 */
13
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
14
    public void onPaid(OrderPaidEvent event) {
15
        try {
16
            noticeService.sendPaidNotice(event.orderId());
17
        } catch (Exception ex) {
18
            // AFTER_COMMIT 阶段原事务已提交，这里的失败只能靠日志与补偿
19
            log.error("dispatch paid notice failed, orderId={}", event.orderId(), ex);
20
        }
21
    }
22
}