前面十章讲的都是 Servlet 那一套（阻塞式）：一个请求占一个线程，线程等 IO 的时候就在那儿干等。这套模型简单直观，绝大多数业务够用。但当并发从几百涨到几万，问题就来了——线程是稀缺资源，一个线程占几 MB 栈内存，等 IO 的线程全是浪费。WebFlux 是 Spring 给出的另一条路：用少量线程处理海量请求，用"回调链"代替"阻塞等待"。这一章把它的执行模型、Reactor 核心 API、以及"什么时候千万别用"讲清楚。

两条路线的执行模型：线程到底在干什么

理解 WebFlux 的关键只有一句话：Servlet 是"一个请求一个线程，线程阻塞等 IO"，WebFlux 是"少量线程轮转处理，IO 完成后再回来接着处理"。下面用一张表把这个差别量化。

Reactor 核心：Mono、Flux 与那几个最容易搞混的操作符

WebFlux 默认使用 Reactor 作为响应式库。它的两个核心类型很朴素：Mono<T> 表示"0 或 1 个结果"，Flux<T> 表示"0 到 N 个结果"。它们都是"发布者（Publisher）"，在你订阅之前什么都不做。

先跑一遍最基础的：创建、变换、消费

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class ReactorBasics {

    public static void main(String[] args) {
        // ---------- Mono：0 或 1 个元素 ----------
        Mono<String> mono = Mono.just("hello");
        Mono<String> empty = Mono.empty();
        Mono<String> error = Mono.error(new RuntimeException("炸了"));

        // 关键：不订阅，什么都不会发生！下面这行不会打印任何东西
        mono.map(String::toUpperCase);

        // 订阅之后才真正执行
        mono.map(String::toUpperCase)
            .subscribe(
                value -> System.out.println("收到: " + value),
                err   -> System.out.println("出错: " + err.getMessage()),
                ()    -> System.out.println("完成")
            );

        // ---------- Flux：0 到 N 个元素 ----------
        Flux<Integer> nums = Flux.just(1, 2, 3, 4, 5);

        nums.filter(n -> n % 2 == 0)      // 只要偶数
            .map(n -> n * 10)             // 每个都乘 10
            .subscribe(System.out::println);  // 20 40

        // 定时产生数据：每秒一个，共 5 个（常用于测试或心跳场景）
        Flux.interval(java.time.Duration.ofSeconds(1))
            .take(5)
            .subscribe(i -> System.out.println("tick " + i));
    }
}

flatMap、concatMap、zip 的差别，用一段代码看清

// 假设有个远程接口：根据 id 查用户名，耗时 100ms
private Mono<String> queryNameAsync(Long id) {
    return Mono.fromCallable(() -> {
        Thread.sleep(100);              // 模拟远程调用（真实场景用 WebClient）
        return "user-" + id;
    });
}

// ---------- flatMap：并发执行，总耗时约 100ms，但顺序不保证 ----------
Flux.just(1L, 2L, 3L)
    .flatMap(id -> queryNameAsync(id))
    .subscribe(System.out::println);     // 可能是 user-2 user-1 user-3

// ---------- concatMap：串行执行，总耗时约 300ms，顺序严格保证 ----------
Flux.just(1L, 2L, 3L)
    .concatMap(id -> queryNameAsync(id))
    .subscribe(System.out::println);     // 一定按 user-1 user-2 user-3 顺序，但更慢

// ---------- zip：两个流配对，等到两边都有第一个元素才开始输出 ----------
Mono<String> user  = queryNameAsync(1L);
Mono<Integer> score = Mono.just(90);

Mono<String> combined = Mono.zip(user, score)
    .map(t -> t.getT1() + " 得分 " + t.getT2());
combined.subscribe(System.out::println);   // user-1 得分 90

// ---------- 注意：zip 有"短板效应" ----------
// 如果一边只发 3 个元素、另一边只发 2 个，zip 结果只有 2 个
// 如果一边永远不发（比如 Mono.never()），zip 会一直等着，表现为"请求卡住"

subscribeOn 与 publishOn：调度到底在哪一步发生

这是响应式编程最容易搞混的一对。用一句话记住：subscribeOn 决定"从哪开始执行"（影响订阅信号向上传播），publishOn 决定"从哪之后在哪个线程执行"（影响数据向下游传播）。

用打印出来的线程名把语义看清楚

import reactor.core.scheduler.Schedulers;

// ---------- 场景一：subscribeOn 决定"源头在哪个线程执行" ----------
Mono.fromCallable(() -> {
        System.out.println("执行线程: " + Thread.currentThread().getName());
        return "data";
    })
    .subscribeOn(Schedulers.boundedElastic())     // 让订阅动作在弹性线程池执行
    .subscribe();
// 输出：执行线程: boundedElastic-1
// subscribeOn 的位置不重要，它影响的是"整个链从哪开始跑"

// ---------- 场景二：publishOn 切换"下游在哪执行" ----------
Mono.just("a")
    .map(v -> { System.out.println("map1: " + Thread.currentThread().getName()); return v; })
    .publishOn(Schedulers.parallel())              // 从这行之后换线程
    .map(v -> { System.out.println("map2: " + Thread.currentThread().getName()); return v; })
    .subscribe();
// 输出：
//   map1: main                  ← publishOn 之前，还是 main 线程
//   map2: parallel-1            ← publishOn 之后，切到了 parallel 线程

// ---------- 调度器的选择 ----------
// immediate()      ：当前线程，不切换。适合"已经是异步的了，不用再切"
// single()         ：单个可复用线程。适合"必须串行、但有顺序要求"
// parallel()       ：固定大小线程池（约等于 CPU 核数）。适合纯 CPU 计算
// boundedElastic() ：弹性线程池，有上限。适合阻塞式 IO（老 JDBC、文件 IO）
// ==========> 记住：CPU 密集用 parallel，阻塞调用用 boundedElastic

// ---------- 最标准的写法：把阻塞调用隔离到 boundedElastic ----------
public Mono<String> callLegacyBlockingApi(String param) {
    return Mono.fromCallable(() -> legacyService.blockingCall(param))
               .subscribeOn(Schedulers.boundedElastic());
    // 这样阻塞调用被挪到了弹性线程池，不会卡住 Netty 的事件循环
    // 这是"必须调用阻塞 API 时"唯一的正确姿势
}

WebClient 与函数式端点

进入实践部分。WebFlux 生态里，所有"调用别人的 HTTP 接口"都应该用 WebClient 而不是 RestTemplate，因为 RestTemplate 是阻塞的——在事件循环线程里调用它，等于把整条事件循环卡住。

WebClient 替代 RestTemplate：一次配置，处处异步调用

import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import java.time.Duration;

@Configuration
public class WebClientConfig {

    @Bean
    public WebClient userWebClient() {
        return WebClient.builder()
            .baseUrl("http://user-service")
            // 关键：设置连接池上限与超时，避免下游慢时把连接耗尽
            .clientConnector(new ReactorClientHttpConnector(
                HttpClient.create()
                    .responseTimeout(Duration.ofSeconds(3))
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000)
                    .doOnConnected(conn -> conn
                        .addHandlerLast(new ReadTimeoutHandler(3))
                        .addHandlerLast(new WriteTimeoutHandler(3)))))
            .build();
    }
}

@Service
public class UserReactiveService {

    private final WebClient userWebClient;
    public UserReactiveService(WebClient userWebClient) { this.userWebClient = userWebClient; }

    // 返回 Mono，不阻塞。调用方继续往下编排，等结果回来再处理
    public Mono<UserDTO> getUser(Long id) {
        return userWebClient.get()
            .uri("/user/{id}", id)
            .retrieve()
            // 4xx/5xx 时转成异常，否则 retrieve() 会把错误当正常响应返回
            .onStatus(status -> status.isError(),
                      resp -> resp.bodyToMono(String.class)
                                  .flatMap(body -> Mono.error(
                                      new RuntimeException("下游返回 " + resp.statusCode() + ": " + body))))
            .bodyToMono(UserDTO.class)
            .timeout(Duration.ofSeconds(3))            // 兜底超时，防止无限等待
            .onErrorResume(e -> Mono.just(UserDTO.guest()));  // 降级
    }

    // 并发调用两个下游，再合并结果：总耗时 = max(两个耗时)，而不是相加
    public Mono<OrderDetailVO> getOrderDetail(Long userId) {
        Mono<UserDTO>  user  = getUser(userId);
        Mono<List<OrderDTO>> orders = getOrders(userId);
        return Mono.zip(user, orders)
                   .map(t -> new OrderDetailVO(t.getT1(), t.getT2()));
    }

    private Mono<List<OrderDTO>> getOrders(Long userId) {
        return userWebClient.get().uri("/order/{id}", userId)
            .retrieve().bodyToFlux(OrderDTO.class).collectList();
    }
}

函数式端点（RouterFunction）与注解式端点的差别

// ================= 方式一：注解式（和 Spring MVC 写法几乎一样）=================
@RestController
@RequestMapping("/api/user")
public class UserController {

    private final UserReactiveService service;
    public UserController(UserReactiveService service) { this.service = service; }

    // 注意区别：返回类型是 Mono/Flux，而不是普通对象
    // 千万不要在里面 .block()！那等于把响应式又改回了阻塞
    @GetMapping("/{id}")
    public Mono<UserDTO> getUser(@PathVariable Long id) {
        return service.getUser(id);        // 直接返回，框架负责订阅并写出响应
    }

    // Flux 用于流式返回：客户端会分批收到数据（如 SSE）
    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> stream() {
        return Flux.interval(Duration.ofSeconds(1)).map(i -> "tick " + i);
    }
}

// ================= 方式二：函数式端点（WebFlux 特有）=================
// 路由和处理器分开写：路由负责"路径 → 处理器"，处理器负责业务逻辑
@Configuration
public class UserRouter {

    @Bean
    public RouterFunction<ServerResponse> userRoutes(UserHandler handler) {
        return RouterFunctions.route()
            .GET("/fn/user/{id}", accept(MediaType.APPLICATION_JSON), handler::getUser)
            .POST("/fn/user", handler::createUser)
            .GET("/fn/users", handler::listUsers)
            .build();     // 链式声明，路由表一眼可见
    }
}

@Component
public class UserHandler {

    private final UserReactiveService service;
    public UserHandler(UserReactiveService service) { this.service = service; }

    public Mono<ServerResponse> getUser(ServerRequest request) {
        Long id = Long.valueOf(request.pathVariable("id"));
        return service.getUser(id)
            .flatMap(user -> ServerResponse.ok().bodyValue(user))
            .switchIfEmpty(ServerResponse.notFound().build());
    }

    public Mono<ServerResponse> createUser(ServerRequest request) {
        return request.bodyToMono(UserDTO.class)
            .flatMap(service::save)
            .flatMap(saved -> ServerResponse.created(
                    URI.create("/fn/user/" + saved.getId())).bodyValue(saved));
    }
}

// 对比结论：注解式上手快、和 MVC 一致；函数式路由集中、易做动态组合与测试
// 团队熟悉度优先时选注解式；需要"路由即数据"（比如按配置动态拼装）时选函数式

R2DBC 与响应式数据访问

到这里会遇到一个很尴尬的现实：业务逻辑可以写得很响应式，但 JDBC 是阻塞的。JDBC 规范里没有异步接口，一旦调用 Statement.executeQuery()，线程就必须等。在 WebFlux 里用 JDBC，等于在高速公路上开拖拉机。

Spring Data R2DBC：接口写法与 JPA 类似，但返回类型是响应式的

// ---------- 1) 实体：不用 JPA 注解，用 Spring Data 的 @Table / @Id ----------
@Table("users")
public class UserEntity {
    @Id
    private Long id;
    private String name;
    private Integer level;
    // getter / setter 省略
}

// ---------- 2) Repository：继承 ReactiveCrudRepository ----------
public interface UserRepository extends ReactiveCrudRepository<UserEntity, Long> {

    // 方法名派生查询，和 JPA 一样；但返回类型是 Mono / Flux
    Flux<UserEntity> findByLevel(Integer level);

    // 自定义 SQL 用 @Query，注意语法是数据库方言
    @Query("SELECT * FROM users WHERE name LIKE :prefix || '%'")
    Flux<UserEntity> searchByNamePrefix(String prefix);
}

// ---------- 3) 配置：连接工厂 + 事务管理 ----------
@Configuration
public class R2dbcConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        return ConnectionFactories.get(
            "r2dbc:pool:postgresql://localhost:5432/mydb?user=app&password=secret");
        // r2dbc:pool: 前缀会启用连接池，生产必须用池，别直连
    }

    @Bean
    public ReactiveTransactionManager transactionManager(ConnectionFactory cf) {
        return new R2dbcTransactionManager(cf);
    }
}

// ---------- 4) 事务：注意是 @Transactional 的响应式版本 ----------
@Service
public class UserService {

    @Transactional     // 在 WebFlux 下，这个注解会走 ReactiveTransactionManager
    public Mono<UserEntity> createUser(UserEntity user) {
        // 关键：整条链必须在同一个"响应式事务上下文"里
        // 所以用 flatMap 串联，不要在里面 .subscribe()，也不要返回普通对象
        return userRepository.save(user)
            .flatMap(saved -> recordAuditLog(saved))   // 同一事务内，失败一起回滚
            .thenReturn(user);
    }

    private Mono<Void> recordAuditLog(UserEntity user) {
        return auditRepository.save(new AuditLog(user.getId(), "CREATE")).then();
    }
}

什么场景用 WebFlux，什么场景别碰

这一节可能是整章最有价值的部分。WebFlux 不是"更先进的 MVC"，它是为特定问题设计的工具。选错了，你会用十倍的复杂度换来更差的性能。

一个务实的技术选型判断流程

第一步：这个服务的瓶颈是 IO 还是 CPU？
    如果是 CPU 密集（加解密、图像处理、复杂计算）→ 用 MVC + 线程池，别用 WebFlux
    如果是 IO 密集 → 继续第二步

第二步：峰值并发大概多少？单机需要多少线程才能扛住？
    QPS × 平均响应时间 = 需要的并发线程数
    例：2000 QPS × 300ms = 600 个线程 → MVC 默认 200 线程池已经不够了
    如果算下来需要几百上千线程 → WebFlux 有明确收益
    如果只需要几十个线程 → MVC 完全够，没必要

第三步：整条链路能做到全非阻塞吗？
    数据访问：能不能上 R2DBC？不能的话，阻塞部分占多大比例？
    下游调用：能不能全换成 WebClient？
    如果有超过 30% 的调用是阻塞的 → WebFlux 收益被吃掉大半，慎重

第四步：团队能承担这个学习成本吗？
    响应式的调试、排查、新人上手都是实打实的成本
    答案是否 → 老实选 MVC，把线程池调大 + 加超时熔断，性价比更高

最终建议：微服务架构里最常见的落地方式是
    「网关层 + BFF 聚合层用 WebFlux，业务服务保持 MVC」
    这样收益最大、风险最小

章末面试题