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: 8192

Maven 依赖（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(); } }

九、性能优化建议

1. 避免阻塞操作：不要在响应式链中调用阻塞方法

2. 合理使用线程池：对阻塞操作使用publishOn切换到弹性线程池

3. 背压处理：合理使用onBackpressureBuffer、onBackpressureDrop等策略

4. 连接池配置：配置数据库、Redis等连接池大小

5. 响应式驱动：使用响应式数据库驱动（如R2DBC、Reactive MongoDB）

十、实战项目结构

text

src/main/java/ ├── config/ # 配置类 ├── controller/ # 注解式控制器 ├── router/ # 函数式路由 ├── handler/ # 处理器函数 ├── service/ # 业务逻辑层 ├── repository/ # 响应式仓库 ├── model/ # 数据模型 └── exception/ # 异常处理

学习路径建议

1. 先学习 Reactor：掌握 Mono/Flux 的常用操作符

2. 理解背压机制：学习如何控制数据流

3. 从注解式开始：如果你熟悉 Spring MVC，更容易上手

4. 尝试函数式编程：理解 Router Functions 的优势

5. 集成响应式数据库：实践完整的数据流处理

6. 学习测试：掌握 WebTestClient 和 StepVerifier

7. 官方文档：Spring WebFlux
8. Project Reactor 文档：Reactor Reference Guide

---

---

---