跳到主要内容Spring WebFlux 深度实践指南 | 极客日志Javajava
Spring WebFlux 深度实践指南
本文介绍 Spring WebFlux 响应式 Web 框架的深度实践。内容包括构建响应式 REST API，支持注解式和函数式端点；使用 R2DBC 进行响应式数据库访问及事务管理；配置 WebSocket 实现实时通信及 RSocket 集成；最后涵盖性能监控与最佳实践，如线程模型理解、背压策略和错误处理。旨在帮助开发者构建高性能、可扩展的非阻塞应用。
Spring WebFlux 是 Spring Framework 5 引入的响应式 Web 框架，基于 Project Reactor 实现，支持非阻塞、函数式编程模型。本节将深入探讨 WebFlux 的核心功能，包括 REST API 构建、响应式数据库访问和实时通信。
4.3.1 构建 Reactive REST API
基础项目搭建
首先创建 Spring Boot WebFlux 项目（基于 Spring Initializr）：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
响应式控制器
注解式控制器（与传统 Spring MVC 类似但支持响应式类型）：
@RestController
@RequestMapping("/products")
public class ProductController {
private final ProductService productService;
public ProductController(ProductService productService) {
this.productService = productService;
}
@GetMapping
public Flux<Product> getAllProducts() {
productService.findAll();
}
Mono<Product> {
productService.findById(id);
}
Mono<Product> {
productService.save(productMono);
}
Mono<Product> {
productService.update(id, productMono);
}
Mono<Void> {
productService.delete(id);
}
}
微信扫一扫，关注极客日志
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
相关免费在线工具
- Keycode 信息
查找任何按下的键的javascript键代码、代码、位置和修饰符。 在线工具，Keycode 信息在线工具，online
- Escape 与 Native 编解码
JavaScript 字符串转义/反转义；Java 风格 \uXXXX（Native2Ascii）编码与解码。 在线工具，Escape 与 Native 编解码在线工具，online
- JavaScript / HTML 格式化
使用 Prettier 在浏览器内格式化 JavaScript 或 HTML 片段。 在线工具，JavaScript / HTML 格式化在线工具，online
- JavaScript 压缩与混淆
Terser 压缩、变量名混淆，或 javascript-obfuscator 高强度混淆（体积会增大）。 在线工具，JavaScript 压缩与混淆在线工具，online
- Base64 字符串编码/解码
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
- Base64 文件转换器
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
return
@GetMapping("/{id}")
public
getProductById
(@PathVariable String id)
return
@PostMapping
@ResponseStatus(HttpStatus.CREATED)
public
createProduct
(@RequestBody Mono<Product> productMono)
return
@PutMapping("/{id}")
public
updateProduct
(@PathVariable String id, @RequestBody Mono<Product> productMono)
return
@DeleteMapping("/{id}")
@ResponseStatus(HttpStatus.NO_CONTENT)
public
deleteProduct
(@PathVariable String id)
return
函数式端点（RouterFunction 方式）：
@Configuration
public class ProductRouter {
@Bean
public RouterFunction<ServerResponse> route(ProductHandler handler) {
return RouterFunctions.route()
.GET("/fn/products", handler::getAll)
.GET("/fn/products/{id}", handler::getById)
.POST("/fn/products", handler::create)
.PUT("/fn/products/{id}", handler::update)
.DELETE("/fn/products/{id}", handler::delete)
.build();
}
}
@Component
public class ProductHandler {
private final ProductService productService;
public ProductHandler(ProductService productService) {
this.productService = productService;
}
public Mono<ServerResponse> getAll(ServerRequest request) {
return ServerResponse.ok().contentType(MediaType.APPLICATION_NDJSON).body(productService.findAll(), Product.class);
}
public Mono<ServerResponse> getById(ServerRequest request) {
String id = request.pathVariable("id");
return productService.findById(id)
.flatMap(product -> ServerResponse.ok().bodyValue(product))
.switchIfEmpty(ServerResponse.notFound().build());
}
public Mono<ServerResponse> create(ServerRequest request) {
return request.bodyToMono(Product.class)
.flatMap(productService::save)
.flatMap(product -> ServerResponse.created(URI.create("/fn/products/" + product.getId())).bodyValue(product));
}
public Mono<ServerResponse> update(ServerRequest request) {
String id = request.pathVariable("id");
return request.bodyToMono(Product.class)
.flatMap(product -> productService.update(id, Mono.just(product)))
.flatMap(product -> ServerResponse.ok().bodyValue(product))
.switchIfEmpty(ServerResponse.notFound().build());
}
public Mono<ServerResponse> delete(ServerRequest request) {
String id = request.pathVariable("id");
return productService.delete(id).then(ServerResponse.noContent().build());
}
}
高级特性
流式响应（Server-Sent Events）：
@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ProductEvent> streamProducts() {
return Flux.interval(Duration.ofSeconds(1))
.map(sequence -> new ProductEvent("product-" + sequence, "Event at " + Instant.now()));
}
@ControllerAdvice
public class GlobalErrorHandler extends AbstractErrorWebExceptionHandler {
public GlobalErrorHandler(ErrorAttributes errorAttributes, WebProperties.Resources resources, ApplicationContext applicationContext, ServerCodecConfigurer serverCodecConfigurer) {
super(errorAttributes, resources, applicationContext);
this.setMessageWriters(serverCodecConfigurer.getWriters());
}
@Override
protected RouterFunction<ServerResponse> getRoutingFunction(ErrorAttributes errorAttributes) {
return RouterFunctions.route(RequestPredicates.all(), request -> {
Map<String, Object> errorProperties = getErrorAttributes(request, ErrorAttributeOptions.defaults());
HttpStatus status = getHttpStatus(errorProperties);
return ServerResponse.status(status).contentType(MediaType.APPLICATION_JSON).bodyValue(errorProperties);
});
}
private HttpStatus getHttpStatus(Map<String, Object> errorProperties) {
return HttpStatus.valueOf((Integer) errorProperties.get("status"));
}
}
public class ProductValidator {
public static Mono<Product> validate(Product product) {
return Mono.just(product).flatMap(p -> {
List<String> errors = new ArrayList<>();
if (p.getName() == null || p.getName().isEmpty()) {
errors.add("Product name is required");
}
if (p.getPrice() <= 0) {
errors.add("Price must be positive");
}
if (!errors.isEmpty()) {
return Mono.error(new ValidationException(errors));
}
return Mono.just(p);
});
}
}
4.3.2 响应式数据库访问（R2DBC）
R2DBC 配置
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
<groupId>io.r2dbc</groupId>
<artifactId>r2dbc-postgresql</artifactId>
<scope>runtime</scope>
</dependency>
spring:
r2dbc:
url: r2dbc:postgresql://localhost:5432/mydb
username: user
password: pass
pool:
enabled: true
max-size: 20
响应式 Repository
@Data
@Table("products")
public class Product {
@Id
private Long id;
private String name;
private String description;
private BigDecimal price;
private Instant createdAt;
}
public interface ProductRepository extends ReactiveCrudRepository<Product, Long> {
Flux<Product> findByNameContaining(String name);
@Query("SELECT * FROM products WHERE price > :minPrice")
Flux<Product> findByPriceGreaterThan(BigDecimal minPrice);
@Modifying
@Query("UPDATE products SET price = price * :factor")
Mono<Integer> updateAllPrices(BigDecimal factor);
}
复杂查询与事务
public class ProductRepositoryImpl implements CustomProductRepository {
private final DatabaseClient databaseClient;
public ProductRepositoryImpl(DatabaseClient databaseClient) {
this.databaseClient = databaseClient;
}
@Override
public Flux<Product> complexSearch(ProductCriteria criteria) {
return databaseClient.sql("""
SELECT * FROM products
WHERE name LIKE :name AND price BETWEEN :minPrice AND :maxPrice
ORDER BY :sortField :sortDirection
LIMIT :limit OFFSET :offset
""")
.bind("name", "%" + criteria.getName() + "%")
.bind("minPrice", criteria.getMinPrice())
.bind("maxPrice", criteria.getMaxPrice())
.bind("sortField", criteria.getSortField())
.bind("sortDirection", criteria.getSortDirection())
.bind("limit", criteria.getPageSize())
.bind("offset", (criteria.getPageNumber() - 1) * criteria.getPageSize())
.map((row, metadata) -> toProduct(row))
.all();
}
private Product toProduct(Row row) {
}
}
@Service
@RequiredArgsConstructor
public class ProductService {
private final ProductRepository productRepository;
private final TransactionalOperator transactionalOperator;
public Mono<Void> transferStock(String fromId, String toId, int quantity) {
return transactionalOperator.execute(status ->
productRepository.findById(fromId)
.flatMap(fromProduct -> {
if (fromProduct.getStock() < quantity) {
return Mono.error(new InsufficientStockException());
}
fromProduct.setStock(fromProduct.getStock() - quantity);
return productRepository.save(fromProduct)
.then(productRepository.findById(toId))
.flatMap(toProduct -> {
toProduct.setStock(toProduct.getStock() + quantity);
return productRepository.save(toProduct);
});
})
);
}
}
性能优化
spring:
r2dbc:
pool:
max-size: 20
initial-size: 5
max-idle-time: 30m
public Mono<Integer> batchInsert(List<Product> products) {
return databaseClient.inConnectionMany(connection -> {
Batch batch = connection.createBatch();
products.forEach(product ->
batch.add("INSERT INTO products(name, price) VALUES($1, $2)")
.bind(0, product.getName())
.bind(1, product.getPrice())
);
return Flux.from(batch.execute()).reduce(0, (count, result) -> count + result.getRowsUpdated());
});
}
4.3.3 WebSocket 实时通信
基础 WebSocket 配置
@Configuration
@EnableWebFlux
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/ws")
.setHandshakeHandler(new DefaultHandshakeHandler())
.setAllowedOrigins("*");
}
@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
registry.enableSimpleBroker("/topic");
registry.setApplicationDestinationPrefixes("/app");
}
}
响应式 WebSocket 处理
@Controller
public class StockTickerController {
private final Flux<StockQuote> stockQuoteFlux;
public StockTickerController(StockQuoteGenerator quoteGenerator) {
this.stockQuoteFlux = Flux.interval(Duration.ofMillis(500))
.map(sequence -> quoteGenerator.generate())
.share();
}
@MessageMapping("stocks.subscribe")
@SendTo("/topic/stocks")
public Flux<StockQuote> subscribe() {
return stockQuoteFlux;
}
@MessageMapping("stocks.filter")
public Flux<StockQuote> filter(@Payload String symbol) {
return stockQuoteFlux.filter(quote -> quote.getSymbol().equals(symbol));
}
}
const socket = new SockJS('/ws');
const stompClient = Stomp.over(socket);
stompClient.connect({}, () => {
stompClient.subscribe('/topic/stocks', (message) => {
const quote = JSON.parse(message.body);
updateStockTable(quote);
});
stompClient.send("/app/stocks.filter", {}, "AAPL");
});
高级特性
@Controller
@MessageMapping("stock.service")
public class RSocketStockController {
@MessageMapping("current")
public Mono<StockQuote> current(String symbol) {
return stockService.getCurrent(symbol);
}
@MessageMapping("stream")
public Flux<StockQuote> stream(String symbol) {
return stockService.getStream(symbol);
}
@MessageMapping("channel")
public Flux<StockQuote> channel(Flux<String> symbols) {
return symbols.flatMap(stockService::getStream);
}
}
@MessageMapping("large.data.stream")
public Flux<DataChunk> largeDataStream() {
return dataService.streamLargeData()
.onBackpressureBuffer(50, chunk -> log.warn("Dropping chunk due to backpressure"));
}
安全配置
@Configuration
@EnableWebFluxSecurity
public class SecurityConfig {
@Bean
public SecurityWebFilterChain securityFilterChain(ServerHttpSecurity http) {
return http
.authorizeExchange()
.pathMatchers("/ws/**").authenticated()
.anyExchange().permitAll()
.and()
.httpBasic()
.and()
.csrf().disable()
.build();
}
@Bean
public MapReactiveUserDetailsService userDetailsService() {
UserDetails user = User.withUsername("user")
.password("{noop}password")
.roles("USER")
.build();
return new MapReactiveUserDetailsService(user);
}
}
性能监控与最佳实践
监控端点配置
management:
endpoints:
web:
exposure:
include: health, metrics, prometheus
metrics:
tags:
application: ${spring.application.name}
响应式应用监控
@Bean
public MeterRegistryCustomizer<MeterRegistry> metricsCommonTags() {
return registry -> registry.config().commonTags("application", "reactive-demo");
}
@Bean
public WebFilter metricsWebFilter(MeterRegistry registry) {
return (exchange, chain) -> {
String path = exchange.getRequest().getPath().toString();
Timer.Sample sample = Timer.start(registry);
return chain.filter(exchange).doOnSuccessOrError((done, ex) -> {
sample.stop(registry.timer("http.requests", "uri", path, "status", exchange.getResponse().getStatusCode().toString(), "method", exchange.getRequest().getMethodValue()));
});
};
}
最佳实践总结
- 线程模型理解：
- WebFlux 默认使用 Netty 事件循环线程
- 阻塞操作必须使用
publishOn 切换到弹性线程池
- 背压策略选择：
- UI 客户端：使用
onBackpressureDrop 或 onBackpressureLatest
- 服务间通信：使用
onBackpressureBuffer 配合合理缓冲区大小
- 错误处理原则：
- 尽早处理错误
- 为每个 Flux/Mono 链添加错误处理
- 区分业务异常和系统异常
- 测试策略：
- 使用
StepVerifier 测试响应式流
- 使用
WebTestClient 测试控制器
- 虚拟时间测试长时间操作
- 性能调优：
- 合理配置连接池
- 监控关键指标（延迟、吞吐量、资源使用率）
- 使用响应式日志框架（如 Logback 异步 Appender）
通过以上全面实践，您将能够构建高性能、可扩展的响应式 Web 应用，充分利用 WebFlux 的非阻塞特性，处理高并发场景下的各种挑战。
