English 中文 العربية 日本語 Português

Spring Boot: Spring WebFlux 响应式编程

最后更新：2026-08-26

WebFlux 是 Spring 的异步引擎——Mono 一个、Flux 一流，非阻塞 I/O 应对万级并发。

1. 你将学到

2. 一个高并发架构师的真实故事

(1) 痛点：线程池耗尽

Alice 的 OrderFlow 在大促时遇到瓶颈——500 并发用户查询订单，Tomcat 线程池（默认 200 线程）耗尽，请求排队等待，P99 延迟飙到 3 秒。线程大部分时间在等 I/O（数据库、网络），CPU 利用率只有 15%。Bob 建议加线程数，但每增加 1000 线程就多消耗 1 GB 内存。

(2) WebFlux 的解法

WebFlux 用少量线程处理大量并发——I/O 等待时线程不阻塞，去服务其他请求：

JAVA 
@GetMapping("/{id}")
public Mono<OrderResponse> getOrder(@PathVariable Long id) {
    return orderRepository.findById(id)
        .map(OrderResponse::from)
        .switchIfEmpty(Mono.error(new ResourceNotFoundException("Order", id)));
}

(3) 收益

Alice 用 WebFlux 重构查询接口后，4 个线程支撑 2000 并发，内存占用降低 70%，P99 延迟从 3 秒降到 100ms，CPU 利用率提升到 60%。

3. 响应式编程核心概念

(1) Mono 和 Flux

类型 含义 类比 示例
Mono<T> 0 或 1 个元素 Optional 的异步版 查询单个订单
Flux<T> 0 到 N 个元素 Stream 的异步版 查询订单列表
Mono<Void> 无返回值 完成信号 删除操作
100% 
graph LR
    A["Publisher"] --> B["Mono<br/>0 or 1 element"]
    A --> C["Flux<br/>0 to N elements"]
    B --> D["onNext → onComplete"]
    C --> E["onNext × N → onComplete"]
    B --> F["onError"]
    C --> F

▶ 示例： Mono 和 Flux 基本操作

JAVA 
// Mono: single value
Mono<String> mono = Mono.just("Hello OrderFlow");
Mono<String> empty = Mono.empty();
Mono<String> fromCallable = Mono.fromCallable(() -> fetchOrder(1L));

// Flux: multiple values
Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);
Flux<Integer> range = Flux.range(1, 100);
Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));

输出：

TEXT 📖 仅展示 
// 执行成功

(2) 常用操作符

操作符 作用 示例
map 转换元素 .map(OrderResponse::from)
flatMap 异步转换 .flatMap(this::enrichOrder)
filter 过滤 .filter(order -> "PENDING".equals(order.getStatus()))
switchIfEmpty 空值替代 .switchIfEmpty(Mono.error(...))
onErrorResume 错误降级 .onErrorResume(e -> fallbackOrder())
timeout 超时控制 .timeout(Duration.ofSeconds(5))
retry 重试 .retry(3)
subscribeOn 指定订阅线程 .subscribeOn(Schedulers.boundedElastic())

▶ 示例： 操作符链式调用

JAVA 
public Mono<OrderResponse> getOrderWithDetails(Long id) {
    return orderRepository.findById(id)
        .flatMap(order ->
            productRepository.findById(order.getProductId())
                .map(product -> OrderResponse.from(order, product)))
        .switchIfEmpty(Mono.error(
            new ResourceNotFoundException("Order", id)))
        .timeout(Duration.ofSeconds(3))
        .onErrorResume(TimeoutException.class,
            e -> Mono.error(new RuntimeException("Order query timed out")))
        .retry(2);
}

输出：

TEXT 📖 仅展示 
// 执行成功

4. WebFlux REST API

▶ 示例： 响应式 Controller

JAVA 
@RestController
@RequestMapping("/api/v1/reactive/orders")
public class ReactiveOrderController {

    private final ReactiveOrderService orderService;

    public ReactiveOrderController(ReactiveOrderService orderService) {
        this.orderService = orderService;
    }

    @GetMapping("/{id}")
    public Mono<OrderResponse> getOrder(@PathVariable Long id) {
        return orderService.findById(id);
    }

    @GetMapping
    public Flux<OrderResponse> listOrders(
            @RequestParam(defaultValue = "0") int page,
            @RequestParam(defaultValue = "20") int size) {
        return orderService.findAll(page, size);
    }

    @PostMapping
    public Mono<ResponseEntity<OrderResponse>> createOrder(
            @Valid @RequestBody CreateOrderRequest request) {
        return orderService.create(request)
            .map(order -> ResponseEntity
                .status(HttpStatus.CREATED).body(order));
    }

    @DeleteMapping("/{id}")
    public Mono<ResponseEntity<Void>> cancelOrder(@PathVariable Long id) {
        return orderService.cancel(id)
            .then(Mono.just(ResponseEntity.noContent().<Void>build()));
    }
}

输出：

TEXT 📖 仅展示 
// 执行成功
维度 WebMVC WebFlux
返回类型 同步对象 Mono<T> / Flux<T>
线程模型 一请求一线程 事件循环（少量线程）
I/O 模型 阻塞 非阻塞
容器 Tomcat / Jetty Netty
并发上限 线程池大小 几乎无上限

5. WebFlux vs WebMVC 选型

(1) 选型决策

场景 推荐 原因
CRUD API，并发 < 500 WebMVC 简单直观，生态成熟
I/O 密集型，高并发 WebFlux 非阻塞，少量线程支撑大量并发
实时流（SSE/WebSocket） WebFlux 原生支持
大量阻塞操作（JDBC） WebMVC WebFlux 中阻塞操作反而更差
混合场景 WebMVC + 部分异步 渐进式优化
⚠️ 注意： WebFlux 不是万能药。如果你的代码大量使用阻塞 I/O（如 JDBC），WebFlux 反而比 WebMVC 更差。选择 WebFlux 意味着整个调用链都必须是非阻塞的。

▶ 示例： Server-Sent Events (SSE)

JAVA 
@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<OrderEvent> streamOrderEvents() {
    return orderEventPublisher.eventStream()
        .log("order-events");
}

输出：

TEXT 📖 仅展示 
// 执行成功

6. Spring Data R2DBC

(1) 响应式数据库访问

▶ 示例： R2DBC Entity 和 Repository

XML 
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
    <groupId>io.r2dbc</groupId>
    <artifactId>r2dbc-mysql</artifactId>
</dependency>

输出：

TEXT 📖 仅展示 
// 执行成功
JAVA 
@Table("products")
public class Product {

    @Id
    private Long id;
    private String name;
    private BigDecimal price;
    private Integer stock;

    public Product() {}

    public Product(String name, BigDecimal price, Integer stock) {
        this.name = name;
        this.price = price;
        this.stock = stock;
    }
    // getters
}

public interface ReactiveProductRepository
        extends ReactiveCrudRepository<Product, Long> {

    Flux<Product> findByNameContaining(String keyword);

    Flux<Product> findByStockLessThan(Integer threshold);
}
维度 JPA + JDBC R2DBC
驱动模型 阻塞 非阻塞
API 风格 同步 响应式（Mono/Flux）
关系映射 支持（@OneToMany） 不支持
查询语言 JPQL 原生 SQL / 方法名派生
事务 @Transactional @Transactional（Reactive）

▶ 示例： 响应式 Service

JAVA 
@Service
public class ReactiveOrderService {

    private final ReactiveOrderRepository orderRepository;
    private final ReactiveProductRepository productRepository;

    public ReactiveOrderService(ReactiveOrderRepository orderRepository,
                                ReactiveProductRepository productRepository) {
        this.orderRepository = orderRepository;
        this.productRepository = productRepository;
    }

    public Mono<OrderResponse> create(CreateOrderRequest request) {
        return productRepository.findById(request.productId())
            .switchIfEmpty(Mono.error(
                new ResourceNotFoundException("Product", request.productId())))
            .flatMap(product -> {
                if (product.getStock() < request.quantity()) {
                    return Mono.error(new InsufficientStockException(
                        product.getId(), product.getStock(), request.quantity()));
                }
                product.deductStock(request.quantity());
                return productRepository.save(product)
                    .then(orderRepository.save(new Order(product.getId(), request.quantity())));
            })
            .map(OrderResponse::from);
    }

    public Flux<OrderResponse> findAll(int page, int size) {
        return orderRepository.findAll()
            .skip(page * size)
            .take(size)
            .map(OrderResponse::from);
    }

    public Mono<Void> cancel(Long orderId) {
        return orderRepository.findById(orderId)
            .flatMap(order -> {
                order.setStatus("CANCELLED");
                return orderRepository.save(order);
            })
            .then();
    }
}

输出：

TEXT 📖 仅展示 
// 执行成功

7. 综合示例：OrderFlow 响应式查询 API

JAVA 
// ReactiveProductRepository.java
public interface ReactiveProductRepository
        extends ReactiveCrudRepository<Product, Long> {
    Flux<Product> findByNameContaining(String keyword);
}

// ReactiveOrderRepository.java
public interface ReactiveOrderRepository
        extends ReactiveCrudRepository<Order, Long> {
    Flux<Order> findByStatus(String status);
}

// ReactiveOrderController.java
@RestController
@RequestMapping("/api/v1/reactive/orders")
public class ReactiveOrderController {

    private final ReactiveOrderService orderService;

    public ReactiveOrderController(ReactiveOrderService orderService) {
        this.orderService = orderService;
    }

    @GetMapping("/{id}")
    public Mono<OrderResponse> get(@PathVariable Long id) {
        return orderService.findById(id);
    }

    @GetMapping
    public Flux<OrderResponse> list(
            @RequestParam(defaultValue = "0") int page,
            @RequestParam(defaultValue = "20") int size) {
        return orderService.findAll(page, size);
    }

    @GetMapping("/status/{status}")
    public Flux<OrderResponse> byStatus(@PathVariable String status) {
        return orderService.findByStatus(status);
    }

    @PostMapping
    public Mono<ResponseEntity<OrderResponse>> create(
            @RequestBody CreateOrderRequest req) {
        return orderService.create(req)
            .map(o -> ResponseEntity.status(HttpStatus.CREATED).body(o));
    }

    @DeleteMapping("/{id}")
    public Mono<ResponseEntity<Void>> cancel(@PathVariable Long id) {
        return orderService.cancel(id)
            .thenReturn(ResponseEntity.noContent().<Void>build());
    }
}

// ReactiveOrderService.java
@Service
public class ReactiveOrderService {

    private final ReactiveOrderRepository orderRepo;
    private final ReactiveProductRepository productRepo;

    public ReactiveOrderService(ReactiveOrderRepository orderRepo,
                                ReactiveProductRepository productRepo) {
        this.orderRepo = orderRepo;
        this.productRepo = productRepo;
    }

    public Mono<OrderResponse> findById(Long id) {
        return orderRepo.findById(id)
            .map(OrderResponse::from)
            .switchIfEmpty(Mono.error(new ResourceNotFoundException("Order", id)));
    }

    public Flux<OrderResponse> findAll(int page, int size) {
        return orderRepo.findAll().skip((long) page * size).take(size)
            .map(OrderResponse::from);
    }

    public Flux<OrderResponse> findByStatus(String status) {
        return orderRepo.findByStatus(status).map(OrderResponse::from);
    }

    public Mono<OrderResponse> create(CreateOrderRequest req) {
        return productRepo.findById(req.productId())
            .switchIfEmpty(Mono.error(new ResourceNotFoundException("Product", req.productId())))
            .flatMap(p -> {
                if (p.getStock() < req.quantity()) {
                    return Mono.error(new InsufficientStockException(p.getId(), p.getStock(), req.quantity()));
                }
                p.deductStock(req.quantity());
                return productRepo.save(p)
                    .then(orderRepo.save(new Order(p.getId(), req.quantity())));
            })
            .map(OrderResponse::from);
    }

    public Mono<Void> cancel(Long id) {
        return orderRepo.findById(id)
            .flatMap(o -> { o.setStatus("CANCELLED"); return orderRepo.save(o); })
            .then();
    }
}

❓ 常见问题

Q WebFlux 能完全替代 WebMVC 吗？
A 不建议。WebMVC 生态更成熟，绝大多数 CRUD 项目用 WebMVC 更简单。WebFlux 适合 I/O 密集、高并发、实时流的场景。两者可以共存，但不要混用（不要在 WebFlux 中用阻塞代码）。
Q Mono 和 CompletableFuture 有什么区别？
A Mono 是冷源（lazy，订阅时才执行），CompletableFuture 是热源（创建时就开始执行）。Mono 支持背压（backpressure）和取消，CompletableFuture 不支持。
Q R2DBC 支持 @OneToMany 关系映射吗？
A 不支持。R2DBC 是轻量级的响应式数据库驱动，不做关系映射。需要关联查询时，手动使用 flatMap 组合多个查询。
Q 在 WebFlux 中调用阻塞代码怎么办？
A 用 Mono.fromCallable() + subscribeOn(Schedulers.boundedElastic()) 将阻塞操作调度到弹性线程池。但频繁使用说明不适合 WebFlux。
Q WebFlux 的错误处理怎么做？
A 1）操作符 onErrorResume / onErrorReturn；2）全局 @ExceptionHandler（WebFlux 也支持）；3）onErrorMap 转换异常类型。
Q WebFlux 如何测试？
A 使用 WebTestClient 替代 MockMvc。@WebFluxTest 加载 WebFlux 层。StepVerifier 用于测试 Mono/Flux 的元素序列。

📖 小节

📝 作业

  1. 基础题（难度⭐）：使用 WebFlux 实现商品查询响应式 API（GET /api/v1/reactive/products/{id}），用 R2DBC 连接 H2 内存数据库。

  2. 进阶题（难度⭐⭐）：实现响应式下单接口，包含库存检查、扣减、创建订单的完整流程，使用 flatMap 组合多个响应式操作。

  3. 挑战题（难度⭐⭐⭐）：使用 WebTestClient 和 StepVerifier 编写响应式 API 测试，实现 SSE 实时订单事件推送接口，对比 WebMVC 和 WebFlux 在 1000 并发下的性能差异。

Web-Tutorial.com

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

100%

🙏 帮我们做得更好

我们是刚上线的编程教程站，几个人的小团队，精力有限。页面虽经检查，难免还有疏漏——链接失效、排版错乱、内容有误、语言生硬……

如果您发现了，麻烦告诉我们，我们会在收到反馈后第一时间进行修复，再次感谢您的光临 🙏