Spring WebFlux 学习
Spring WebFlux 是 Spring Framework 5 引入的响应式（Reactive）Web 框架，用于构建非阻塞、异步、事件驱动的 Web 应用程序。它与传统的基于 Servlet 的 Spring MVC 并行存在，但底层架构完全不同。
一、WebFlux 核心概念
1.1 什么是响应式编程？
数据流：将数据视为随时间变化的流
非阻塞：不等待操作完成，继续执行其他任务
背压：消费者控制生产者速度，防止数据积压
1.2 与传统 Spring MVC 对比
| 特性 | Spring MVC | Spring WebFlux |
|---|---|---|
| 编程模型 | 命令式、同步 | 响应式、异步 |
| 并发模型 | 线程池（每个请求一个线程） | 事件循环（少量线程处理大量请求） |
| 阻塞性 | 可能阻塞（如DB查询） | 完全非阻塞 |
| 适用场景 | 传统应用、简单CRUD | 高并发、实时应用、流处理 |
二、WebFlux 核心组件
2.1 Reactive Streams API
Publisher<T> // 发布者 Subscriber<T> // 订阅者 Subscription // 订阅关系 Processor<T,R> // 处理器2.2 Reactor 核心类
// Mono: 0-1个元素的异步序列 Mono<String> mono = Mono.just("Hello"); Mono<Void> monoEmpty = Mono.empty(); Mono<String> monoError = Mono.error(new RuntimeException()); // Flux: 0-N个元素的异步序列 Flux<String> flux = Flux.just("A", "B", "C"); Flux<Integer> fluxRange = Flux.range(1, 10); Flux<Long> fluxInterval = Flux.interval(Duration.ofSeconds(1));三、WebFlux 开发方式
3.1 注解式控制器（类似 MVC）
@RestController @RequestMapping("/api") public class UserController { @GetMapping("/users/{id}") public Mono<User> getUser(@PathVariable String id) { return userRepository.findById(id); } @GetMapping("/users") public Flux<User> getAllUsers() { return userRepository.findAll(); } @PostMapping("/users") public Mono<User> createUser(@RequestBody User user) { return userRepository.save(user); } // SSE（服务器发送事件） @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux<User> streamUsers() { return userRepository.findAll().delayElements(Duration.ofSeconds(1)); } }3.2 函数式端点（Router Functions）
@Configuration public class RouterConfig { @Bean public RouterFunction<ServerResponse> routes(UserHandler userHandler) { return RouterFunctions.route() .GET("/api/users", userHandler::getAllUsers) .GET("/api/users/{id}", userHandler::getUser) .POST("/api/users", userHandler::createUser) .build(); } } @Component public class UserHandler { public Mono<ServerResponse> getAllUsers(ServerRequest request) { Flux<User> users = userRepository.findAll(); return ServerResponse.ok() .contentType(MediaType.APPLICATION_JSON) .body(users, User.class); } public Mono<ServerResponse> getUser(ServerRequest request) { String id = request.pathVariable("id"); Mono<User> user = userRepository.findById(id); return user.flatMap(u -> ServerResponse.ok().bodyValue(u)) .switchIfEmpty(ServerResponse.notFound().build()); } }四、WebFlux 与数据层集成
4.1 Reactive MongoDB
@Document public class Product { @Id private String id; private String name; private Double price; } public interface ProductRepository extends ReactiveMongoRepository<Product, String> { Flux<Product> findByPriceGreaterThan(Double price); @Query("{ 'name': { $regex: ?0 } }") Flux<Product> findByNameRegex(String regex); }4.2 Reactive Redis
@Configuration public class RedisConfig { @Bean public ReactiveRedisTemplate<String, Object> reactiveRedisTemplate( ReactiveRedisConnectionFactory factory) { RedisSerializationContext<String, Object> context = RedisSerializationContext .newSerializationContext(new StringRedisSerializer()) .hashKey(new StringRedisSerializer()) .hashValue(new GenericJackson2JsonRedisSerializer()) .build(); return new ReactiveRedisTemplate<>(factory, context); } } @Service public class CacheService { @Autowired private ReactiveRedisTemplate<String, Object> redisTemplate; public Mono<Object> get(String key) { return redisTemplate.opsForValue().get(key); } public Mono<Boolean> set(String key, Object value, Duration timeout) { return redisTemplate.opsForValue() .set(key, value, timeout); } }五、WebFlux 配置
5.1 应用配置
# application.yml spring: webflux: base-path: /api hidden-method: filter: enabled: true codec: max-in-memory-size: 10MB thymeleaf: reactive: max-chunk-size: 8192Maven 依赖（Spring Boot）
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency>5.2 服务器配置
@Configuration public class ServerConfig { @Bean public WebFluxConfigurer webFluxConfigurer() { return new WebFluxConfigurer() { @Override public void configureHttpMessageCodecs( ServerCodecConfigurer configurer) { configurer.defaultCodecs() .maxInMemorySize(10 * 1024 * 1024); } @Override public void addCorsMappings(CorsRegistry registry) { registry.addMapping("/**") .allowedOrigins("*") .allowedMethods("*"); } }; } }六、响应式 WebClient
@Service public class ApiClientService { private final WebClient webClient; public ApiClientService(WebClient.Builder builder) { this.webClient = builder .baseUrl("https://api.chengxuyuanshitang.com") .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) .build(); } public Mono<User> fetchUser(String userId) { return webClient.get() .uri("/users/{id}", userId) .retrieve() .bodyToMono(User.class) .timeout(Duration.ofSeconds(5)) .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(1))); } public Flux<Product> streamProducts() { return webClient.get() .uri("/products/stream") .accept(MediaType.TEXT_EVENT_STREAM) .retrieve() .bodyToFlux(Product.class); } }七、错误处理
@RestControllerAdvice public class GlobalErrorHandler { @ExceptionHandler(ResourceNotFoundException.class) public Mono<ResponseEntity<ErrorResponse>> handleNotFound( ResourceNotFoundException ex) { ErrorResponse error = new ErrorResponse( "NOT_FOUND", ex.getMessage() ); return Mono.just( ResponseEntity.status(HttpStatus.NOT_FOUND) .body(error) ); } // 响应式异常处理 @ExceptionHandler(WebExchangeBindException.class) public Mono<ResponseEntity<ErrorResponse>> handleValidationException( WebExchangeBindException ex) { List<String> errors = ex.getFieldErrors() .stream() .map(fe -> fe.getField() + ": " + fe.getDefaultMessage()) .collect(Collectors.toList()); ErrorResponse error = new ErrorResponse( "VALIDATION_ERROR", errors.toString() ); return Mono.just( ResponseEntity.badRequest().body(error) ); } }八、WebFlux 测试
@SpringBootTest @AutoConfigureWebTestClient class UserControllerTest { @Autowired private WebTestClient webTestClient; @Test void testGetUser() { webTestClient.get() .uri("/api/users/1") .exchange() .expectStatus().isOk() .expectBody() .jsonPath("$.name").isEqualTo("zhangsanfeng"); } @Test void testStreamUsers() { webTestClient.get() .uri("/api/users/stream") .accept(MediaType.TEXT_EVENT_STREAM) .exchange() .expectStatus().isOk() .expectHeader().contentType(MediaType.TEXT_EVENT_STREAM) .returnResult(User.class) .getResponseBody() .take(3) .as(StepVerifier::create) .expectNextCount(3) .verifyComplete(); } }九、性能优化建议
避免阻塞操作：不要在响应式链中调用阻塞方法
合理使用线程池：对阻塞操作使用
publishOn切换到弹性线程池背压处理：合理使用
onBackpressureBuffer、onBackpressureDrop等策略连接池配置：配置数据库、Redis等连接池大小
响应式驱动：使用响应式数据库驱动（如R2DBC、Reactive MongoDB）
十、实战项目结构
text
src/main/java/ ├── config/ # 配置类 ├── controller/ # 注解式控制器 ├── router/ # 函数式路由 ├── handler/ # 处理器函数 ├── service/ # 业务逻辑层 ├── repository/ # 响应式仓库 ├── model/ # 数据模型 └── exception/ # 异常处理
学习路径建议
先学习 Reactor：掌握 Mono/Flux 的常用操作符
理解背压机制：学习如何控制数据流
从注解式开始：如果你熟悉 Spring MVC，更容易上手
尝试函数式编程：理解 Router Functions 的优势
集成响应式数据库：实践完整的数据流处理
学习测试：掌握 WebTestClient 和 StepVerifier
- 官方文档：Spring WebFlux
- Project Reactor 文档：Reactor Reference Guide