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

news2024/11/27 8:16:45

更多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 {
    @Async
    public 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 {
    
    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(25);
        executor.setThreadNamePrefix("AsyncThread-");
        executor.initialize();
        return executor;
    }
    
    @Override
    public 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 {
    
    @Autowired
    private 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
    @EventListener
    public 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: localhost
    port: 5672
    username: guest
    password: guest

6.2 消息队列配置类

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

6.3 消息生产者

@Service
@Slf4j
public class OrderProducer {
    
    @Autowired
    private 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 {
    
    @Bean
    public 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;
    }
}