脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → Spring WebFlux实践

Spring WebFlux 核心作用

 更新时间：2025年08月11日 09:28:35   作者：百锦再@新空间  
Spring WebFlux是Spring Framework 5的响应式框架,结合性能优化与最佳实践,构建高效、可扩展的Web应用,本节将深入探讨 WebFlux 的核心功能,包括 REST API 构建、响应式数据库访问和实时通信,感兴趣的朋友一起看看吧

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() {
        return productService.findAll();
    }
    @GetMapping("/{id}")
    public Mono<Product> getProductById(@PathVariable String id) {
        return productService.findById(id);
    }
    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Mono<Product> createProduct(@RequestBody Mono<Product> productMono) {
        return productService.save(productMono);
    }
    @PutMapping("/{id}")
    public Mono<Product> updateProduct(
            @PathVariable String id, 
            @RequestBody Mono<Product> productMono) {
        return productService.update(id, productMono);
    }
    @DeleteMapping("/{id}")
    @ResponseStatus(HttpStatus.NO_CONTENT)
    public Mono<Void> deleteProduct(@PathVariable String id) {
        return productService.delete(id);
    }
}

函数式端点（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>

配置 application.yml：

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;
}

创建Repository接口：

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");
});

高级特性

RSocket集成（更强大的响应式协议）：

@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()));
            });
    };
}

最佳实践总结

  1. 线程模型理解：

    • WebFlux 默认使用 Netty 事件循环线程
    • 阻塞操作必须使用 publishOn 切换到弹性线程池

  2. 背压策略选择：

    • UI 客户端：使用 onBackpressureDrop 或 onBackpressureLatest
    • 服务间通信：使用 onBackpressureBuffer 配合合理缓冲区大小

  3. 错误处理原则：

    • 尽早处理错误
    • 为每个 Flux/Mono 链添加错误处理
    • 区分业务异常和系统异常

  4. 测试策略：

    • 使用 StepVerifier 测试响应式流
    • 使用 WebTestClient 测试控制器
    • 虚拟时间测试长时间操作

  5. 性能调优：

    • 合理配置连接池
    • 监控关键指标（延迟、吞吐量、资源使用率）
    • 使用响应式日志框架（如 Logback 异步Appender）

通过以上全面实践，您将能够构建高性能、可扩展的响应式 Web 应用，充分利用 WebFlux 的非阻塞特性，处理高并发场景下的各种挑战。

到此这篇关于Spring WebFlux 深度实践指南的文章就介绍到这了,更多相关Spring WebFlux内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • swing中Tree与滚动条用法实例分析

    swing中Tree与滚动条用法实例分析

    这篇文章主要介绍了swing中Tree与滚动条用法,以实例形式分析了java基于swing实现图形界面的使用技巧,需要的朋友可以参考下
    2015-09-09
  • RocketMQ消息发送与消息类别详解

    RocketMQ消息发送与消息类别详解

    这篇文章主要介绍了RocketMQ消息发送与消息类别详解,事务消息的生产者执行本地事务,并根据事务执行的结果选择是否提交或回滚事务,
    如果事务执行成功并选择提交事务,则产生注册成功消息,进入下一步,需要的朋友可以参考下
    2023-09-09
  • log4j使用详细解析

    log4j使用详细解析

    Log4j 除了可以记录程序运行日志信息外还有一重要的功能就是用来显示调试信息。下面通过本文给大家介绍log4j使用详细解析，感兴趣的朋友一起看看吧
    2017-10-10
  • MyBatis Plus实现一对多的查询场景的三种方法

    MyBatis Plus实现一对多的查询场景的三种方法

    MyBatis Plus提供了多种简便的方式来进行一对多子查询,本文主要介绍了MyBatis Plus实现一对多的查询场景的三种方法,具有一定的参考价值,感兴趣的可以了解一下
    2024-07-07
  • Java中结束循环的方法

    Java中结束循环的方法

    这篇文章主要介绍了Java中结束循环的方法,文中有段代码在return，结束了整个main方法，即使输出hello world的语句位于循环体外，也不会被执行，对java结束循环方法感兴趣的朋友跟随小编一起看看吧
    2023-06-06
  • Mybatis ResultMap和分页操作示例详解

    Mybatis ResultMap和分页操作示例详解

    这篇文章主要为大家介绍了Mybatis ResultMap和分页操作示例详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2023-10-10
  • 解决mybatis使用char类型字段查询oracle数据库时结果返回null问题

    解决mybatis使用char类型字段查询oracle数据库时结果返回null问题

    这篇文章主要介绍了mybatis使用char类型字段查询oracle数据库时结果返回null问题的解决方法，本文给大家介绍的非常详细，具有一定的参考借鉴价值,需要的朋友可以参考下
    2018-06-06
  • SpringBoot3.x打包Docker容器的实现

    SpringBoot3.x打包Docker容器的实现

    这篇文章主要介绍了SpringBoot3.x打包Docker容器的实现,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-11-11
  • 浅谈Java字符串比较的三种方法

    浅谈Java字符串比较的三种方法

    这篇文章主要介绍了浅谈Java字符串比较的三种方法,字符串比较是常见的操作，包括比较相等、比较大小、比较前缀和后缀串等,需要的朋友可以参考下
    2023-04-04
  • Java利用POI实现导入导出Excel表格示例代码

    Java利用POI实现导入导出Excel表格示例代码

    最近工作中遇到一个需求，是需要导出数据到Excel表格里，所以写个Demo测试一下，还是比较简单的，现在分享给大家，有需要的朋友们可以参考借鉴，下面来一起看看吧。
    2016-10-10

最新评论