重学SpringBoot3-异步编程完全指南

更多SpringBoot3内容请关注我的专栏：《SpringBoot3》
期待您的点赞👍收藏⭐评论✍

重学SpringBoot3-异步编程完全指南

1. 简介
2. @Async注解
- 2.1 基础配置
- 2.2 基本使用
- 2.3 自定义线程池
3. WebFlux响应式编程
- 3.1 依赖配置
- 3.2 响应式Controller示例
- 3.3 响应式Service实现
4. CompletableFuture的使用
- 4.1 基本操作
- 4.2 组合操作
5. 事件驱动模型
- 5.1 自定义事件
- 5.2 事件监听器
6. 消息队列(MQ)异步处理
- 6.1 RabbitMQ配置
- 6.2 消息队列配置类
- 6.3 消息生产者
- 6.4 消息消费者
- 6.5 消息确认机制
7. 最佳实践
- 7.1 异常处理
- 7.2 线程池配置
- 7.3 性能优化
- 7.4 消息队列使用建议
8. 总结
参考资料

1. 简介

在现代应用程序开发中，异步编程已经成为提升应用性能和用户体验的重要手段。SpringBoot 3提供了多种异步编程的方式，本文将详细介绍这些实现方式及其最佳实践。

2. @Async注解

2.1 基础配置

首先需要在启动类或配置类上启用异步支持：

@EnableAsync
@SpringBootApplication
public class Application {public static void main(String[] args) {SpringApplication.run(Application.class, args);}
}

2.2 基本使用

@Service
public class EmailService {@Asyncpublic CompletableFuture<String> sendEmail(String to) {// 模拟发送邮件耗时try {Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}return CompletableFuture.completedFuture("邮件发送成功：" + to);}
}

2.3 自定义线程池

@Configuration
public class AsyncConfig implements AsyncConfigurer {@Overridepublic Executor getAsyncExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(5);executor.setMaxPoolSize(10);executor.setQueueCapacity(25);executor.setThreadNamePrefix("AsyncThread-");executor.initialize();return executor;}@Overridepublic AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {return new SimpleAsyncUncaughtExceptionHandler();}
}

3. WebFlux响应式编程

3.1 依赖配置

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

3.2 响应式Controller示例

@RestController
@RequestMapping("/api")
public class UserController {@Autowiredprivate UserService userService;@GetMapping("/users")public Flux<User> getAllUsers() {return userService.findAllUsers();}@GetMapping("/users/{id}")public Mono<User> getUser(@PathVariable String id) {return userService.findById(id);}
}

3.3 响应式Service实现

@Service
public class UserService {public Flux<User> findAllUsers() {return Flux.fromIterable(users).delayElements(Duration.ofMillis(100));}public Mono<User> findById(String id) {return Mono.justOrEmpty(findUserById(id)).delayElement(Duration.ofMillis(100));}
}

4. CompletableFuture的使用

4.1 基本操作

@Service
public class OrderService {public CompletableFuture<Order> processOrder(Order order) {return CompletableFuture.supplyAsync(() -> {// 处理订单逻辑return order;});}public CompletableFuture<Order> validateOrder(Order order) {return CompletableFuture.supplyAsync(() -> {// 验证订单逻辑return order;});}
}

4.2 组合操作

public CompletableFuture<Order> createOrder(Order order) {return validateOrder(order).thenCompose(this::processOrder).thenApply(processedOrder -> {// 更新订单状态return processedOrder;}).exceptionally(ex -> {// 异常处理log.error("订单处理失败", ex);return null;});
}

5. 事件驱动模型

5.1 自定义事件

public class OrderCreatedEvent extends ApplicationEvent {private final Order order;public OrderCreatedEvent(Object source, Order order) {super(source);this.order = order;}public Order getOrder() {return order;}
}

5.2 事件监听器

@Component
public class OrderEventListener {@Async@EventListenerpublic void handleOrderCreatedEvent(OrderCreatedEvent event) {Order order = event.getOrder();// 异步处理订单逻辑}
}

6. 消息队列(MQ)异步处理

6.1 RabbitMQ配置

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring:rabbitmq:host: localhostport: 5672username: guestpassword: guest

6.2 消息队列配置类

@Configuration
public class RabbitMQConfig {@Beanpublic Queue orderQueue() {return new Queue("order.queue", true);}@Beanpublic DirectExchange orderExchange() {return new DirectExchange("order.exchange");}@Beanpublic Binding orderBinding(Queue orderQueue, DirectExchange orderExchange) {return BindingBuilder.bind(orderQueue).to(orderExchange).with("order.routing.key");}
}

6.3 消息生产者

@Service
@Slf4j
public class OrderProducer {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendOrder(Order order) {try {rabbitTemplate.convertAndSend("order.exchange", "order.routing.key", order);log.info("订单消息发送成功: {}", order.getId());} catch (Exception e) {log.error("订单消息发送失败", e);}}
}

6.4 消息消费者

@Component
@Slf4j
public class OrderConsumer {@RabbitListener(queues = "order.queue")public void processOrder(Order order) {try {log.info("收到订单消息: {}", order.getId());// 异步处理订单逻辑processOrderAsync(order);} catch (Exception e) {log.error("订单处理失败", e);}}private void processOrderAsync(Order order) {// 具体的订单处理逻辑}
}

6.5 消息确认机制

@Configuration
public class RabbitMQConfirmConfig {@Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);// 消息发送到交换机确认rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {if (!ack) {log.error("消息发送到交换机失败: {}", cause);}});// 消息从交换机路由到队列确认rabbitTemplate.setReturnsCallback(returned -> {log.error("消息从交换机路由到队列失败: {}", returned);});return rabbitTemplate;}
}