第 15 篇 · 响应式编程与系统设计

这道题考察对 "响应式编程如何用异步数据流抽象统一同步与异步、并解决生产消费速率不匹配" 的底层认知。核心不在罗列操作符，而在回答：Mono/Flux 如何把异步事件抽象为可组合的数据流？当生产者速度远超消费者时，背压（Backpressure）如何用"拉模式"避免 OOM？Reactive Streams 规范（JDK 9 Flow / JVM 1.0.4）的四接口契约为何能成为跨厂商互操作标准？Reactor 3.x 在此规范之上提供了哪些工程增强？

Java 异步编程的五代演进，每一代都在解决上一代的"组合困难"与"回调污染"：

回调地狱（早期）：异步操作用嵌套 Callback 实现，多层嵌套导致代码可读性极差、错误处理散落、难以组合。这是响应式编程要消灭的头号敌人。

Future（JDK 5）：提供异步结果占位符，但只能 get() 阻塞获取，无法链式编排多个异步任务，无法注册回调。

CompletableFuture（JDK 8）：支持 thenApply/thenCompose/exceptionally 链式组合，解决回调地狱。但仍缺背压机制——无法表达"数据流"的速率控制，且仍是 Push 模型（生产者主动推）。

Reactive Streams 规范（2015，JVM 1.0.4，JDK 9 通过 java.util.concurrent.Flow 采纳）：定义 Publisher/Subscriber/Subscription/Processor 四接口，确立异步流处理 + 非阻塞背压的标准契约。Netflix/RxJava/Pivotal/TypeSafe/Oracle 联合制定，成为跨厂商互操作基础。

Reactor 3.x（Pivotal，Spring 默认）：Reactive Streams 规范的工程级实现，提供 Mono（0/1 元素）/ Flux（N 元素）两种发布者、数百个操作符、丰富的 Scheduler 线程模型。Spring WebFlux 从 5.0 起将其作为响应式基石。

演进驱动力：① 组合性——异步操作需要像同步流水线一样可组合、可变换；② 资源效率——Thread-per-request 模型下线程昂贵（MB 级栈），响应式用少量线程高并发；③ 背压——生产快于消费时，Push 模型必然 OOM，需 Pull 模式速率控制。

Reactive Streams 四接口契约：这是理解一切响应式框架的根基。Publisher.subscribe(Subscriber) 建立订阅 → Subscriber.onSubscribe(Subscription) 获得订阅凭证 → Subscriber 通过 Subscription.request(n) 主动拉取 n 个元素（背压核心）→ Publisher 按 onNext(item) 推送，不超过请求量 → 完成时 onComplete() 或出错 onError(t)。关键约束：request 调用必须是非阻塞的，且 Publisher 发送的元素数不得超过累计 request 量。

Mono vs Flux：Mono<T> 发射 0 或 1 个元素后完成（对应 Optional/Future 语义），Flux<T> 发射 0 到 N 个元素（对应 Stream/Collection 语义）。二者不可混用边界——HTTP 请求返回单个结果用 Mono，SSE 推送用 Flux。Mono 有额外优化：不维护队列、无并发计数开销。

Cold vs Hot Publisher：这是 Reactor 最核心的语义区分。

Cold Publisher（冷发布者）：每个订阅者独立触发数据流，数据"按需生产"。如 Flux.fromIterable()、HTTP 请求 WebClient.get()——每个 Subscriber 触发一次新请求，订阅 N 次执行 N 次。

Hot Publisher（热发布者）：数据流与订阅无关，先生产后订阅，迟到的订阅者只收到之后的数据。如 Sinks.many().multicast()、鼠标事件流。可用 cache()/replay() 转为带缓存的热流。

背压策略：当 Subscriber 处理慢于 Publisher 生产时，Reactor 提供四种溢出策略：

Scheduler 线程模型：Reactor 用 publishOn(scheduler)（影响下游）和 subscribeOn(scheduler)（影响上游）切换线程。内置调度器：parallel()（CPU 密集，线程数=CPU 核数）、boundedElastic()（IO 密集，有界弹性池，默认 10×CPU 核数上限）、single()（单线程，适合串行）。禁止用 elastic()（已弃用并在 Reactor 3.5+ 移除，无界线程池）。

主流响应式框架对比：

Reactor 链式操作与背压处理：以某租赁平台系统查询车辆列表为例，演示完整的响应式链——WebClient 并发调用库存服务、背压控制、线程切换、错误降级。

// Reactor 链式操作: 并发查询 + 背压处理 + 线程切换
@Service
public class VehicleReactiveService {

    private final WebClient webClient;

    public VehicleReactiveService(WebClient.Builder builder) {
        this.webClient = builder.baseUrl("http://inventory-service").build();
    }

    // Flux: N 元素流, 并发调用, 背压 DROP 策略
    public Flux<VehicleDTO> searchVehicles(List<String> carCodes) {
        return Flux.fromIterable(carCodes)
            // flatMap: 并发订阅内部流, concurrency 控制并发度
            .flatMap(code -> webClient.get()
                    .uri("/cars/{code}", code)
                    .retrieve()
                    .bodyToMono(VehicleDTO.class),
                16)  // 最大并发 16
            // 背压: 下游消费慢时丢弃最旧, 保留最新
            .onBackpressureLatest()
            // IO 操作切到 boundedElastic 线程池
            .publishOn(Schedulers.boundedElastic())
            // 过滤可用车辆
            .filter(VehicleDTO::isAvailable)
            // 超时降级: 2 秒未返回则用缓存
            .timeout(Duration.ofSeconds(2),
                Flux.fromIterable(getCachedVehicles()))
            // 错误降级: 任意异常返回空, 不让链中断
            .onErrorResume(e -> {
                log.error("查询车辆失败", e);
                return Flux.empty();
            });
    }

    // Mono: 单结果, 链式变换 + 阻塞边界演示
    public Mono<RentResult> rentCar(String code, String userId) {
        return webClient.post()
                .uri("/rent")
                .bodyValue(new RentRequest(code, userId))
                .retrieve()
                .bodyToMono(RentResult.class)
                // 订阅时切到 IO 线程池
                .subscribeOn(Schedulers.boundedElastic());
    }
}

// Controller 层: 直接返回 Mono/Flux, 框架自动订阅, 禁止 block()
@RestController
@RequestMapping("/api/vehicles")
public class VehicleController {
    private final VehicleReactiveService service;

    @GetMapping("/search")
    public Flux<VehicleDTO> search(@RequestParam List<String> codes) {
        return service.searchVehicles(codes);  // 非阻塞返回
    }
}

追问 1：Cold Publisher 与 Hot Publisher 的本质区别是什么？如何转换？

本质区别在于数据生产与订阅的关系：Cold 是"每个订阅者独立触发一次数据生产"（类似 HTTP 请求——调用一次执行一次，各订阅者拿到独立结果），Hot 是"数据生产与订阅无关，先有流后有订阅者"（类似广播电台——迟到的听众听不到之前的内容）。Cold 转 Hot 的方式：connect()（手动控制开始时机）、publish().autoConnect(n)（n 个订阅者到达后自动开始）、cache()（缓存生产结果供后续订阅者，相当于 replay 到无限）、replay(n)（缓存最近 n 个元素供新订阅者）、Sinks.many()（手动 push 数据，最灵活的热源）。实战中 SSE 推送、WebSocket 消息用 Hot 流，HTTP 请求/数据库查询用 Cold 流。

追问 2：flatMap、concatMap、merge 的并发差异是什么？

三者都把上游元素映射为内部 Publisher 再合并，但并发与顺序不同：flatMap 并发订阅所有内部流，结果按到达顺序到达（不保证顺序），可通过第二参数限制并发度（如 flatMap(fn, 16)）；concatMap 串行订阅——等前一个内部流完成再订阅下一个，严格保证顺序但吞吐低；merge 是 flatMap 的无并发限制版（并发度无限）。选型：无顺序要求且需高吞吐用 flatMap；要求顺序（如分页有序遍历）用 concatMap；zip 则是等所有源都发射后再组合，适合并行调用多个服务后合并结果。

核心原理之外，响应式落地最常见的工程痛点是"上下文丢失"——ThreadLocal（MDC、安全上下文、事务上下文）在线程切换后失效，需要用 Reactor Context 解决，这也是高频追问点。

• ContextView 与 ThreadLocal 的本质区别：ThreadLocal 绑定线程，Reactor 线程切换（publishOn/subscribeOn、EventLoop）后即丢失；Context 是响应式流的一部分，随数据流传递，不绑定线程。
• 写入与读取：写入用 Mono.deferContextual(ctx -> ...).contextWrite(Context.of("traceId", traceId))，读取用 Mono.deferContextual(ctx -> ctx.get("traceId"))。注意 contextWrite 影响上游操作符、deferContextual 读取下游传入的 Context，位置放错会读到空值。
• 不可变语义：Context 不可变，contextWrite 返回新 Context；多次写入时最靠近读取点的写入生效，遵循"最近写入者生效"。

MDC 适配：

• 问题：MDC 基于 ThreadLocal，Reactor 线程切换后 MDC 内容丢失，日志中 traceId 为空，无法串联调用链。
• 方案一（Hook 自动恢复）：用 Hooks.onEachOperator 在每个操作符执行前从 Context 取 traceId 放入 MDC：

Hooks.onEachOperator(fn -> fn.contextView().getOrEmpty("traceId")
    .ifPresent(traceId -> MDC.put("traceId", traceId)));

• 方案二（手动管理）：用 Mono.deferContextual 读取后 MDC.put，配合 doFinally 执行 MDC.remove，精确控制但侵入代码。

全链路 traceId 传递：

• WebFlux 层：WebFilter 从 HTTP Header 提取 traceId → contextWrite 写入 Reactor Context。
• Service 层：deferContextual 读取 traceId → 注入下游 RPC/WebClient Header。
• DB 层：MyBatis/R2DBC 拦截器从 MDC 读取 traceId → 写入 SQL 日志（前提是链内已恢复 MDC）。
• 异步线程：使用 Schedulers.boundedElastic() 等切换调度器时，配合 contextCapture()（Reactor 3.5+）确保 Context 跨线程传播。

上线门禁：背压与超时、Scheduler 隔离、block 禁区审查。表达提纲：信号语义→操作符链证据→线程模型。

这道题考察对 "非阻塞 Web 框架为何能提升吞吐、以及它的代价与适用边界" 的架构判断力。核心不在背诵差异表，而在回答：WebFlux 用少量 EventLoop 线程为何能比 Thread-per-request 的 WebMVC 承载更多并发？什么场景下该选 WebFlux 而非 WebMVC？WebFlux 的代价（调试难、生态约束、编程心智）是什么？当 Virtual Threads 出现后，WebFlux 的价值是否被削弱？

Java Web 并发模型的三代演进，核心是"如何用更少的线程承载更多并发"：

Servlet/WebMVC（2003+，Thread-per-Request）：每个 HTTP 请求分配一个线程（Tomcat 默认 200 线程），线程在等待 IO（DB/下游服务）时阻塞挂起。优点是编程模型简单（同步代码、栈可调试、ThreadLocal 可用），缺点是线程昂贵（每线程 MB 级栈），高并发下线程数是瓶颈，IO 等待期间线程资源被浪费。

WebFlux（Spring 5.0，2017）：基于 Reactor + Netty（或 Servlet 3.1+ 异步），用少量 EventLoop 线程（默认 CPU 核数×2）处理所有请求，IO 等待时不阻塞线程而是释放回 EventLoop，回调就绪后继续处理。在IO 密集 + 高并发场景下吞吐显著提升，但前提仍是全链路非阻塞——一处阻塞就污染整个 EventLoop，吞吐优势随之消失。

Virtual Threads（JDK 21，JEP 444，2023 GA）：JVM 管理的用户态轻量线程，阻塞时只挂起虚拟线程不占用 OS 载体线程，用同步代码风格获得响应式级别的吞吐。Spring Boot 3.2+ 支持 spring.threads.virtual.enabled=true 让 WebMVC + JDBC 直接跑在虚拟线程上，削弱了 WebFlux 的核心价值。

演进驱动力：① 资源效率——线程是昂贵资源，IO 密集场景下大量线程在阻塞，浪费内存与上下文切换开销；② 编程简洁性——响应式虽高效但调试难、心智负担重，业界一直在追求"高效且简单"的方案，Virtual Threads 是这一方向的突破；③ 微服务高并发——服务间调用链长，每跳 IO 等待累积，非阻塞能显著降低资源占用。

WebFlux 的 EventLoop 模型：以 Netty 为底座，启动时创建 BossEventLoopGroup（接收连接）和 WorkerEventLoopGroup（处理 IO，默认 CPU 核数×2 线程）。每个连接绑定到一个 Worker 线程，该线程用 Selector 多路复用，同时管理数千个连接。当某连接 IO 就绪时线程处理，等待时立即切换处理其他连接——线程永不空闲等待。这与 WebMVC 的"一个请求独占一个线程直到响应完成"形成根本对比。

Handler 处理链：WebFlux 请求处理经过 WebFilter（前置过滤）→ WebHandler（DispatcherHandler）→ HandlerMapping（路由匹配）→ HandlerAdapter（调用 Controller）→ HandlerResultHandler（序列化响应）。全程返回 Mono<Void>，每一步都是非阻塞的——没有一步可以阻塞线程。

背压全链路传导：WebFlux 天然支持背压——当客户端消费慢时，Netty 的写缓冲填满，背压向上游传导到 Controller 返回的 Flux，再传导到 R2DBC/WebClient，最终自动降低数据生产速率。这是 WebMVC 无法做到的（WebMVC 是 Push 模型，生产者不管消费者速率）。

WebMVC vs WebFlux 全维度对比：

迁移成本矩阵（WebMVC → WebFlux，需先做依赖画像）：响应式迁移不是"加个 starter"就完事，必须按依赖画像逐层评估——只有全栈非阻塞才有吞吐收益，遗留阻塞依赖必须隔离或替换，否则"假响应式"反而更糟：

WebFlux RouterFunction + WebClient + R2DBC 全栈示例：以某租赁平台的响应式 API 为例，展示路由定义、非阻塞下游调用、响应式数据访问。

// 1. RouterFunction 路由定义 (替代 @RestController 注解式)
@Configuration
public class VehicleRouter {

    @Bean
    public RouterFunction<ServerResponse> vehicleRoutes(VehicleHandler handler) {
        return RouterFunctions.route()
            .GET("/api/vehicles", handler::listVehicles)
            .GET("/api/vehicles/{code}", handler::getVehicle)
            .POST("/api/vehicles/{code}/rent", handler::rentVehicle)
            .build();
    }
}

// 2. Handler 返回 Mono, 全程非阻塞
@Component
public class VehicleHandler {

    private final VehicleRepository repository;  // ReactiveCrudRepository
    private final WebClient pricingClient;

    public Mono<ServerResponse> getVehicle(ServerRequest req) {
        String code = req.pathVariable("code");
        return repository.findByCode(code)
            // flatMap: 车辆存在则并发查价格
            .flatMap(vehicle -> pricingClient.get()
                    .uri("/price/{code}", code)
                    .retrieve()
                    .bodyToMono(PriceDTO.class)
                    .map(price -> enrichVehicle(vehicle, price)))
            .flatMap(v -> ServerResponse.ok()
                    .contentType(MediaType.APPLICATION_JSON)
                    .bodyValue(v))
            .switchIfEmpty(ServerResponse.notFound().build());  // 404
    }

    // SSE 流式推送车辆状态变更
    public Mono<ServerResponse> streamStatus(ServerRequest req) {
        return ServerResponse.ok()
            .contentType(MediaType.TEXT_EVENT_STREAM)
            .body(repository.streamStatusChanges(), VehicleStatus.class);
    }
}

// 3. R2DBC 响应式 Repository
public interface VehicleRepository extends ReactiveCrudRepository<Vehicle, Long> {
    Mono<Vehicle> findByCode(String code);
    Flux<VehicleStatus> streamStatusChanges();
}

追问 1：WebFlux 吞吐高，但为何单请求延迟不一定比 WebMVC 低？

吞吐与延迟是两个维度。吞吐高是因为少量 EventLoop 线程不阻塞，能同时处理更多并发请求（资源利用率高）；单请求延迟不一定低甚至略高，原因：① Reactor 链有操作符调度、回调编排开销，比同步直接调用多一层间接性；② 线程切换（publishOn/subscribeOn）有上下文切换成本；③ 非阻塞 IO 在低并发时无法摊销多路复用优势。WebFlux 的优势在高并发场景下的尾部延迟稳定性和吞吐上限——WebMVC 线程耗尽时请求排队延迟飙升，WebFlux 少量线程持续处理，尾部延迟更平稳。低并发场景 WebMVC 延迟反而更低。

追问 2：WebFlux 与 Spring WebMVC 能否在同一个应用混用？

技术上不建议混用。Spring Boot 的 spring-boot-starter-web（WebMVC）和 spring-boot-starter-webflux（WebFlux）同时引入时，Spring Boot 会优先按 WebMVC 启动（WebMVC 的自动配置优先级更高），WebFlux 的异步特性被降级为 Servlet 3.1 异步处理，失去了 Netty EventLoop 的性能优势。若确实需要混合（如渐进式迁移），正确做法是：用 WebFlux 但在 Handler 中通过 Schedulers.boundedElastic() 调用阻塞的 WebMVC 代码（如遗留 JDBC），而非两个 Starter 共存。Spring 官方建议新项目二选一，不要混用。

这道题考察对 "为什么 JDBC 的阻塞模型无法适配响应式架构、R2DBC 如何用非阻塞 SPI 重新定义关系数据库访问" 的理解。核心不在 API 差异，而在回答：JDBC 的同步阻塞为何会破坏 WebFlux 全链路非阻塞？R2DBC 如何基于 Reactive Streams 实现非阻塞数据库访问？Spring Data R2DBC 在 R2DBC SPI 之上提供了哪些 Repository 抽象？响应式事务与 JDBC 事务有何不同？

Java 关系数据库访问的响应式演进：

JDBC（1997，JDK 1.1）：Java 数据库访问的标准 SPI，Connection/Statement/ResultSet 全部同步阻塞——执行 SQL 时线程挂起等待数据库返回。在 Thread-per-Request 的 WebMVC 下没问题（线程反正要等），但在 WebFlux 的 EventLoop 下会阻塞少量共享线程，是响应式架构的头号敌人。

Reactive JDBC 封装（rxjava2-jdbc 等）：早期过渡方案——用线程池包装 JDBC 阻塞调用，把同步结果转为 Flowable/Mono。本质是"假响应式"：底层仍阻塞，只是换了线程执行，没有真正解决资源浪费，线程池配置不当仍是瓶颈。

R2DBC（Reactive Relational Database Connectivity，2019 起）：从零设计的原生非阻塞数据库访问 SPI，基于 Reactive Streams 规范。Connection/Statement/Result 全部返回 Publisher，数据库 IO 等待时不占线程。PostgreSQL/MySQL/MariaDB/H2/MSSQL/Oracle 均有驱动实现。

R2DBC 技术分层（明确各层职责，避免与 JDBC 混淆）：① API 规范——R2DBC SPI 定义 ConnectionFactory/Connection/Statement/Result 接口，基于 Reactive Streams 规范（Publisher/Subscriber/Subscription）；② 驱动——各数据库厂商实现 R2DBC SPI（如 r2dbc-postgresql、r2dbc-mysql），负责协议握手、SQL 执行、结果流式返回，是非阻塞网络驱动，与 JDBC 驱动（阻塞 socket）完全不同；③ 框架——Spring Data R2DBC 在 SPI 之上提供 Repository 抽象、实体映射、响应式事务；④ 执行模型——全链路非阻塞，连接等待时不占线程，与 JDBC 的"线程阻塞等待 DB 响应"模型本质不同。注意：HikariCP 是 JDBC 连接池，不适用于 R2DBC；R2DBC 使用独立的 r2dbc-pool 连接池。

Spring Data R2DBC：在 R2DBC SPI 之上提供 ReactiveCrudRepository、DatabaseClient、实体映射、响应式事务管理（TransactionalOperator），类比 Spring Data JPA 但全异步。

演进驱动力：① 全链路非阻塞——WebFlux 架构要求从 Web 到数据库全部非阻塞，JDBC 是最后一块阻塞短板；② 资源效率——R2DBC 用少量连接处理高并发查询，连接等待时不占线程；③ 背压传导——大结果集可背压分批拉取，避免一次性加载 OOM。

R2DBC SPI 核心接口：ConnectionFactory（创建连接，类比 JDBC DriverManager）→ Connection（数据库连接，createStatement() 返回 Statement）→ Statement（执行 SQL，execute() 返回 Publisher<Result>）→ Result（结果集，map((row, meta) -> ...) 逐行非阻塞消费）。所有方法返回 Publisher，无任何阻塞方法。

背压传导机制：当查询返回大结果集时，R2DBC 不会一次性把所有行加载到内存。Subscriber 通过 request(n) 拉取 n 行，驱动从数据库 socket 读取 n 行后回调 onNext，Subscriber 处理完再 request 下一批——背压从应用层一路传导到数据库网络层，避免大结果集 OOM。这是 JDBC ResultSet.next() 同步迭代无法做到的。

连接池（R2DBC Pool）：R2DBC 有独立的连接池实现 r2dbc-pool，提供 ConnectionPool 包装 ConnectionFactory。配置 maxSize（最大连接数）、maxIdleTime（空闲超时）、acquireTimeout（获取超时）。注意：R2DBC 连接池是异步获取的，create() 返回 Mono<Connection>，连接耗尽时不会阻塞而是排队等待（或超时失败）。

响应式事务（TransactionalOperator）：传统 @Transactional 基于 ThreadLocal 绑定连接，WebFlux 线程切换下失效。R2DBC 用 TransactionalOperator（操作符形式）或 ReactiveTransactionManager，通过 Reactor Context 而非 ThreadLocal 传递事务上下文。

JDBC vs R2DBC 本质对比：

Spring Data R2DBC 完整示例：实体定义、Repository、响应式事务、DatabaseClient 自定义查询。

// 1. R2DBC 实体 (类似 JPA @Entity, 但无懒加载/级联)
@Table("vehicle")
@Data  // Lombok
public class Vehicle {
    @Id
    private Long id;
    private String code;
    private String model;
    private BigDecimal dailyRate;
    private Boolean available;
}

// 2. ReactiveCrudRepository: 开箱即用的响应式 CRUD
public interface VehicleRepository extends ReactiveCrudRepository<Vehicle, Long> {

    Mono<Vehicle> findByCode(String code);

    Flux<Vehicle> findByAvailableTrue();

    // 自定义查询
    @Query("SELECT * FROM vehicle WHERE daily_rate < :maxRate ORDER BY daily_rate")
    Flux<Vehicle> findCheapVehicles(@Param("maxRate") BigDecimal maxRate);
}

// 3. 响应式事务: TransactionalOperator (替代 @Transactional)
@Service
public class RentService {

    private final VehicleRepository vehicleRepo;
    private final OrderRepository orderRepo;
    private final TransactionalOperator txOperator;  // 响应式事务

    public Mono<RentOrder> rent(String code, String userId) {
        Mono<RentOrder> pipeline = vehicleRepo.findByCode(code)
            .flatMap(vehicle -> {
                vehicle.setAvailable(false);
                return vehicleRepo.save(vehicle)  // 更新车辆状态
                    .then(orderRepo.save(new RentOrder(vehicle.getId(), userId)));  // 创建订单
            });

        // 用 as(txOperator::transactional) 包裹, 自动提交/回滚
        return pipeline.as(txOperator::transactional);
    }
}

// 4. DatabaseClient: 灵活的响应式 SQL 执行
@Service
public class VehicleQueryService {

    private final DatabaseClient dbClient;

    public Flux<Vehicle> searchByModel(String model) {
        return dbClient.sql("SELECT * FROM vehicle WHERE model LIKE :model")
                .bind("model", "%" + model + "%")
                .map((row, meta) -> rowToVehicle(row))
                .all();  // 返回 Flux, 背压分批
    }
}

追问 1：R2DBC 的成熟度与 JDBC 差距多大？生产可用吗？

R2DBC 1.0 于 2022 年 GA，主流数据库（PostgreSQL/MySQL/MariaDB/H2/MSSQL/Oracle）均有驱动，核心 CRUD 和事务生产可用。但与 JDBC 25 年生态相比仍有差距：① 高级特性缺失——部分数据库特有功能（如 MySQL 的批量写入优化、Oracle 的 PL/SQL 存储过程）支持不全；② ORM 能力弱——Spring Data R2DBC 无 JPA 的级联/懒加载/二级缓存，复杂对象图映射需手写；③ 运维工具少——连接池监控、慢查询分析工具不如 HikariCP 成熟；④ 驱动 Bug——部分边缘场景（如大事务、批量操作）仍有稳定性问题。选型建议：新项目 IO 密集 + WebFlux 可用 R2DBC；复杂 ORM 需求或遗留系统优先 JDBC + WebMVC/Virtual Threads。

追问 2：响应式事务与传统 JDBC 事务有何本质差异？

核心差异在事务上下文传递方式：JDBC 事务用 @Transactional + TransactionSynchronizationManager，通过 ThreadLocal 把 Connection 绑定到当前线程，同一线程内所有 DAO 自动复用该连接——但这要求事务内不切换线程。响应式事务（ReactiveTransactionManager + TransactionalOperator）通过 Reactor Context（随数据流传播的上下文）传递事务，线程切换不影响事务——无论链中 publishOn/subscribeOn 切到哪个线程，Context 跟着流走，事务始终一致。代价是：不能再用 @Transactional 注解（除非配合 ReactiveTransactionManager 的 AOP 适配），且事务边界要显式用 as(txOperator::transactional) 包裹。

选用 R2DBC 的核心量化结论：少量连接（数十级）即可支撑高并发落库，连接数上限与资源占用远低于 JDBC 每请求一连接的线程模型，容量取决于事件循环与背压而非线程数。

这道题考察对 "响应式链中错误如何传播、如何设计健壮的重试与降级机制" 的工程能力。核心不在罗列操作符，而在回答：响应式链中 onError 信号如何传播（与 try-catch 的区别）？重试策略有哪些（固定/退避/抖动/断路器），各自适用什么场景？如何区分可重试错误（网络抖动）与不可重试错误（参数校验）？错误边界如何设计避免局部故障雪崩？

错误处理策略的演进，从简单到健壮：

try-catch（同步时代）：命令式代码用 try-catch-finally 捕获异常，栈展开机制保证异常沿调用链向上传播。但在响应式链中，异常发生在异步回调里，try-catch 包不住——异步错误无法用 try-catch 捕获。

onError 信号（Reactive Streams）：响应式链用 onError 信号传播错误，它是终止信号——一旦发出，流终止，不再发射数据。错误作为终止信号沿操作符链向下游（朝 Subscriber 方向）传播（注意：这与 try-catch 的栈展开"向上调用栈"方向相反，响应式是沿数据流"向下游"），直到被下游的 onErrorResume/onErrorReturn/doOnError 拦截，否则最终到达 Subscriber 的 onError 回调；未处理则抛到全局（Hooks.onErrorDropped）。

retry(n) 固定重试：出错后立即重试 n 次，简单但粗暴——不退避、不区分错误类型，对瞬时故障有效但对持续故障会快速耗尽重试次数，且无退避可能压垮下游。

retryWhen + 指数退避（Exponential Backoff）：出错后按指数递增间隔重试（1s→2s→4s→8s），加抖动（Jitter）避免重试风暴。Reactor 3.x 用 Retry.backoff(maxAttempts, minDuration) 流式配置，支持过滤可重试异常。

断路器 + 降级（Resilience4j）：当错误率超阈值时断路器熔断，直接走降级逻辑不再调用下游，保护系统不被故障服务拖垮。Spring Cloud Circuit Breaker + Resilience4j 提供响应式断路器（CircuitBreakerOperator）。

演进驱动力：① 分布式故障常态化——微服务间网络抖动、超时、限流频繁，需健壮重试；② 防止雪崩——无退避的重试会放大流量压垮下游；③ 快速失败——不可重试错误（如 4xx）不应浪费重试，需分类处理；④ 降级保可用——断路器熔断后走降级，比让用户等超时体验更好。

onError 信号传播规则：在 Reactor 链中，onError 是终止信号——发出后流立即终止，后续 onNext 不再发射。错误沿链向下游传播，直到被错误处理操作符捕获：onErrorResume(fn)（切换到降级流）、onErrorReturn(val)（返回默认值）、doOnError(fn)（副作用，如日志，不消费错误）、onErrorMap(fn)（异常转换）。若链中无任何错误处理，错误传到 subscribe() 的 onError 回调，未处理的错误会抛到全局异常处理器（Hooks.onErrorDropped）。

重试策略对比：

retryWhen 指数退避原理：Reactor 的 retryWhen(Retry) 基于 RetrySignal 流实现——每次出错发射一个信号到内部重试流，重试流决定是否重订阅上游。Retry.backoff(3, Duration.ofSeconds(1)) 表示最多重试 3 次，初始间隔 1 秒，按指数递增（1s→2s→4s），可加 .maxBackoff(Duration.ofSeconds(10)) 封顶、.jitter(0.5) 加 50% 抖动。关键：.filter(ex -> ex instanceof IOException) 只对可重试异常重试。

错误分类原则：这是健壮重试的核心——不是所有错误都该重试。可重试错误（瞬时）：网络超时、连接拒绝、5xx 服务端错误、限流 429；不可重试错误（永久）：4xx 客户端错误（参数错误 400/404/401）、业务校验失败、NPE 等编程错误。重试不可重试错误不仅无效还会浪费资源。

完整的重试 + 退避 + 降级 + 错误分类链：以租赁系统调用定价服务为例，展示生产级错误处理。

@Service
public class PricingService {

    private final WebClient pricingClient;
    private final ReactivePriceCache cache;

    public Mono<BigDecimal> getPrice(String vehicleCode) {
        return pricingClient.get()
                .uri("/price/{code}", vehicleCode)
                .retrieve()
                // HTTP 4xx/5xx 转为异常
                .onStatus(s -> s.is4xxClientError(),
                    resp -> Mono.error(new ClientException(resp.statusCode())))
                .onStatus(s -> s.is5xxServerError(),
                    resp -> Mono.error(new ServerException(resp.statusCode())))
                .bodyToMono(PriceDTO.class)
                .map(PriceDTO::getAmount)
                // 1. 指数退避重试: 仅对可重试异常, 3 次, 1s 起, 50% 抖动
                .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
                    .maxBackoff(Duration.ofSeconds(10))
                    .jitter(0.5)
                    // 仅重试服务端异常和超时, 不重试客户端异常
                    .filter(ex -> ex instanceof ServerException
                                || ex instanceof TimeoutException
                                || ex instanceof WebClientRequestException)
                    .doBeforeRetry(rs -> log.warn("第{}次重试定价服务: {}",
                        rs.totalRetries(), rs.failure())))
                // 2. 超时控制: 单次请求 2 秒
                .timeout(Duration.ofSeconds(2))
                // 3. 降级: 重试耗尽后查缓存
                .onErrorResume(ex -> {
                    log.error("定价服务不可用, 降级查缓存", ex);
                    return cache.getPrice(vehicleCode)
                        // 缓存也没有, 返回默认价
                        .switchIfEmpty(Mono.just(DEFAULT_PRICE));
                })
                // 4. 副作用: 无论成功失败都记录调用
                .doOnSuccess(price -> metrics.recordSuccess(vehicleCode))
                .doOnError(ex -> metrics.recordFailure(vehicleCode, ex));
    }
}

追问 1：重试与幂等的关系是什么？非幂等接口能重试吗？

重试的前提是幂等——即多次执行与一次执行效果相同。幂等接口（GET、PUT、DELETE、带幂等键的 POST）重试安全，因为重复调用不会产生副作用。非幂等接口（如普通 POST 创建订单）重试有风险——第一次请求可能已成功但响应丢失，重试会创建重复订单。非幂等接口要重试必须保证幂等性：① 用幂等键（Idempotency-Key）——客户端生成唯一 ID，服务端去重，重复请求返回首次结果；② 用唯一约束——数据库唯一索引兜底，重复插入抛异常被 onErrorResume 处理；③ 状态检查——重试前先查询是否已创建。实战：支付/下单等关键写操作必须实现幂等键再允许重试，否则宁可快速失败不重试。

追问 2：如何区分网络错误（可重试）与业务错误（不可重试）？

三层分类策略：① 异常类型——WebClientRequestException（连接失败/DNS 解析失败）、TimeoutException、IOException 是网络错误可重试；WebClientResponseException 按 HTTP 状态码细分；② HTTP 状态码——5xx（500/502/503/504）是服务端临时故障可重试，4xx（400/401/403/404/409）是客户端错误不可重试，429（限流）可重试但需更长退避；③ 自定义业务码——响应体 200 但含业务错误码（如 code=INSUFFICIENT_BALANCE），这类业务逻辑错误不可重试。实现：用 onStatus 把 HTTP 状态码转为不同异常类型，再用 retryWhen.filter() 按异常类型过滤。也可封装 RetryableException 标记接口，自定义异常实现该接口表示可重试。

错误处理与重试之外，响应式排错是高频考点——Reactor 的异步链让异常栈"没头没尾"，需要专门的工具链定位问题。

• Hooks.onOperatorDebug()：全局开启后所有操作符记录装配栈（assembly trace），异常时打印完整操作符链调用栈而非只有异常发生处。性能开销大，仅排错时临时开启，生产禁止常开。
• checkpoint("name")：在关键操作符后加标记，异常时输出该标记点。只记录标记不记录完整栈，性能影响小，生产可用——建议在关键业务路径上添加。
• BlockHound：运行时检测响应式流中的阻塞调用（Thread.sleep、同步 IO），检测到即抛 BlockingOperationException 并附带调用栈。
• BlockHound 集成（开发/测试期检测，非生产熔断）：标准做法在应用启动早期调用 BlockHound.install()（常配合 @Profile("!prod") 仅开发/测试启用，避免生产性能开销）；误报用 BlockHound.builder().allowBlockingCallsInside(MyClass.class, "myMethod") 白名单豁免。注意区分：BlockHound 是开发期阻塞调用检测器，生产环境的过载保护应交给 Resilience4j 断路器（CircuitBreakerOperator），二者职责不同、不可混用。

常见排错场景：

• "Assembly trace is not enabled"：异常信息提示未开启装配栈 → 开启 Hooks.onOperatorDebug() 或添加 checkpoint()。
• "Operator called without subscriber"：操作符链构建时机错误 → 用 defer 延迟到订阅时构建。
• "BlockHound detected blocking call"：响应式流中调用了阻塞 IO → 改为异步 API 或切 boundedElastic() 隔离。

这道题考察对 "Virtual Threads 是否终结响应式编程、两种并发模型的本质差异与选型决策" 的前瞻判断力。核心不在捧一贬一，而在回答：Virtual Threads（JEP 444，JDK 21）用同步代码风格获得响应式级别吞吐，是否意味着 Reactive 被淘汰？两种模型在调度机制、资源占用、调试体验、生态兼容上的本质差异是什么？IO 密集 + CPU 密集混合场景如何取舍？未来架构如何组合二者？这是架构师必须能讲清的"并发范式抉择"问题。

Java 并发模型的三代演进，核心是"如何用更轻量的并发单元承载更多并发"：

线程池（JDK 5，ThreadPoolExecutor）：复用 OS 线程避免频繁创建销毁，但每线程仍占 1-2MB 栈空间，且 OS 线程数受内核限制（通常几千）。IO 阻塞时线程挂起浪费资源，高并发下线程数是瓶颈。Thread-per-Request 的 WebMVC 即此模型。

Reactive（Reactor 3.x，2017）：用少量 EventLoop 线程 + 异步回调实现高并发，IO 等待时不占线程。吞吐极高但代价是编程模型复杂——异步链式操作、调试困难、ThreadLocal 失效、生态需响应式驱动。它是"为了效率牺牲简洁性"的方案。

Virtual Threads（JEP 444，JDK 21 GA，2023）：JVM 管理的用户态轻量线程，阻塞时只挂起虚拟线程（unmount），不占用 OS 载体线程（carrier thread），载体线程立即服务其他虚拟线程。用同步代码风格获得 Reactive 级别吞吐——阻塞调用不再昂贵。每虚拟线程仅几 KB 栈，可创建数百万个。这是 Loom 项目十年磨一剑的成果。

演进驱动力：① 简洁性回归——业界厌倦了 Reactive 的调试困难与心智负担，渴望"高效且简单"；② 生态兼容——Virtual Threads 兼容现有同步代码（JDBC/阻塞 HTTP），无需重写；③ JVM 原生——虚拟线程是 JDK 内置能力，无需第三方框架，调度由 JVM 优化；④ 调试友好——虚拟线程有完整的线程栈，可断点调试，不像 Reactor 操作符栈难追踪。

Virtual Threads 调度机制：虚拟线程由 JVM 调度器（ForkJoinPool 共享池，载体线程数=CPU 核数）管理。当虚拟线程执行阻塞操作（IO/BlockingQueue.take()）时，JVM 把它的栈帧从载体线程 unmount（卸载），载体线程立即运行下一个虚拟线程；阻塞就绪后 JVM 把栈帧 mount（装载）回某载体线程恢复执行。关键：阻塞调用不占 OS 线程，这就是"同步代码、异步效率"的原理。

Pinning（载体线程固定）问题：这是 Virtual Threads 的核心陷阱。当虚拟线程在 synchronized 块内或 native 方法中阻塞时，无法 unmount，载体线程被固定（pinned），退化为传统阻塞——此时虚拟线程与 OS 线程 1:1 绑定，失去并发优势。JDK 21 中 synchronized 导致的 pinning 仍未完全解决（JDK 24 由 JEP 491 修复，并继承到 JDK 25 LTS；完整对照见 C04.09）。解决方案：在 JDK 21–23 用 ReentrantLock 替代 synchronized，ReentrantLock.lock() 阻塞时虚拟线程可正常 unmount；JDK 24+ 可直接用 synchronized。检测方法：JDK 21–23 可用 -Djdk.tracePinnedThreads=full；JDK 24+ 该属性已移除，改用 JFR jdk.VirtualThreadPinned（见 C04.09）。

结构化并发（Structured Concurrency，JEP 453 起多次预览，最新 JEP 499/505）：StructuredTaskScope 让并发任务的父子关系显式化——父任务等待所有子任务完成，子任务异常可取消兄弟任务，避免任务泄漏。这是配合 Virtual Threads 的并发编排 API，让并发代码更安全可维护。

三种并发模型全维度对比：

虚拟线程兼容性分场景评估（迁移结论不绝对化）：Virtual Threads 与 Reactive 的选型必须按具体容器、JDBC 驱动、Redis 客户端与线程模型分别评估，不能一刀切"上 VT 就完事"：

Virtual Threads 启用与 pinning 修复：Spring Boot 3.2+ 一行配置启用虚拟线程，同步 JDBC 代码直接获得响应式吞吐。

# application.yml: 一行启用 Virtual Threads (Spring Boot 3.2+)
spring:
  threads:
    virtual:
      enabled: true   # Tomcat 每个请求跑在虚拟线程上

// 1. 启用后, 同步 Controller + JDBC 直接获得高吞吐, 无需改代码
@RestController
public class VehicleController {

    private final JdbcTemplate jdbc;  // 阻塞 JDBC, 虚拟线程下安全
    private final RestTemplate http;  // 阻塞 HTTP, 虚拟线程下安全

    @GetMapping("/vehicles/{code}")
    public Vehicle get(@PathVariable String code) {
        // 阻塞调用, 但跑在虚拟线程上, 阻塞时释放载体线程
        Vehicle v = jdbc.queryForObject("SELECT *...", mapper, code);
        Price p = http.getForObject("http://price/" + code, Price.class);
        return enrich(v, p);  // 同步代码, 简单直观, 可断点调试
    }
}

// 2. 陷阱: JDK 21–23 下 synchronized 块内阻塞会 pinning（JDK 24+ 已由 JEP 491 修复）
public class VehicleCache {

    // BAD: synchronized 块内阻塞会 pin 载体线程, 退化为阻塞
    public synchronized Vehicle getBad(String key) throws Exception {
        return loadFromDb(key);  // 阻塞 DB 调用在 synchronized 内, pinning!
    }

    private final ReentrantLock lock = new ReentrantLock();

    // GOOD: ReentrantLock 阻塞时虚拟线程可正常 unmount
    public Vehicle getGood(String key) throws Exception {
        lock.lock();
        try {
            return loadFromDb(key);  // 阻塞时虚拟线程 unmount, 不 pin
        } finally {
            lock.unlock();
        }
    }
}

// 3. 手动创建虚拟线程 (不依赖 Spring)
public Thread startVirtual(Runnable task) {
    return Thread.ofVirtual().name("vt-", 0).start(task);
}

// 4. 结构化并发 (JDK 21 预览): 并行调用车况和价格, 任一失败则取消全部
public VehicleDetail loadDetail(String code) throws Exception {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        Subtask<Vehicle> vTask = scope.fork(() -> jdbc.queryForObject(...));
        Subtask<Price> pTask = scope.fork(() -> http.getForObject(...));
        scope.join();            // 等待全部完成
        scope.throwIfFailed();   // 任一失败抛异常, 取消其他
        return new VehicleDetail(vTask.get(), pTask.get());
    }
}

追问 1：Virtual Threads 能否完全替代 Reactive？

不能完全替代，二者有不可重叠的优势区。Virtual Threads 替代了 Reactive 的大部分 IO 密集场景——同步代码 + JDBC + 阻塞 HTTP 在虚拟线程下吞吐接近 Reactive，且调试简单、生态兼容，传统 CRUD/微服务调用用 Virtual Threads 更优。但 Reactive 在以下场景仍不可替代：① 流式背压——SSE 推送、大文件流式传输、数据库大结果集分批拉取，Reactive 的 request(n) 背压机制能精确控速，Virtual Threads 是 Push 模型无背压；② 极高并发连接——百万级 WebSocket 长连接，虚拟线程虽轻但每个仍占几 KB 栈 + 调度开销，Reactive 回调无栈更省内存；③ 操作符组合——Reactor 的 flatMap/zip/merge/combineLatest 复杂流编排，用同步代码 + Virtual Threads 实现会很啰嗦；④ 已有 Reactive 代码资产——迁移成本高，不必为迁移而迁移。结论：新项目 IO 密集 + 无流式背压需求优先 Virtual Threads，流式/背压/超高并发长连接保留 Reactive。

追问 2：IO 密集 + CPU 密集混合场景如何选型？

混合场景应分层调度：IO 密集部分用 Virtual Threads（阻塞 IO 不占载体线程），CPU 密集部分用固定线程池或 Reactive parallel() 调度器（CPU 核数个线程）。原因：Virtual Threads 是协作式调度不抢占，CPU 密集任务跑在虚拟线程上会长时间占住载体线程，饿死同一载体线程上的其他虚拟线程，导致 IO 任务延迟飙升。实践模式：① WebMVC + Virtual Threads 处理请求和 IO 调用，CPU 密集计算（如加密/压缩/大数据处理）提交到 ForkJoinPool 或专用 ExecutorService（CompletableFuture.supplyAsync(task, cpuPool)）；② WebFlux + Reactor 用 subscribeOn(Schedulers.boundedElastic()) 跑 IO，publishOn(Schedulers.parallel()) 跑 CPU 密集；③ 混合架构——Web 层用 Virtual Threads（同步简洁），数据流处理层用 Reactive（背压控速），通过 Mono.fromFuture/block() 边界桥接。核心原则：按任务类型分调度器，不要一种并发模型打天下。

选型之外，两种并发模型在真实系统中常需共存——Reactor 链要调阻塞 JDBC、虚拟线程侧要调响应式服务，桥接边界设计是落地必修课。

• 互补而非替代：VT 是"同步代码 + 虚拟线程调度"模型，适合 IO 密集同步代码；Reactor 是"异步回调 + 事件循环"模型，适合背压 + 流式处理。
• VT 内的阻塞调用：Thread.sleep / 阻塞 IO 在 VT 中不会 pin（VT 可 unmount）；synchronized 块内的阻塞 IO 仍会 pin（JDK 21 已修复 synchronized + Object.wait 的 pinning，但 synchronized + IO 仍 pin）。
• JDBC 在 VT 中安全：主流 JDBC 驱动不使用 synchronized + IO，而是 ReentrantLock，阻塞可正常 unmount，是虚拟线程最大的受益场景。

桥接方案：

• Reactor 调 VT：Mono.fromCallable(() -> blockingCall()).subscribeOn(Schedulers.fromExecutor(Executors.newVirtualThreadPerTaskExecutor()))——把阻塞调用放到虚拟线程调度器上，避免污染 EventLoop。
• VT 调 Reactor：mono.block() 在虚拟线程中安全——虚拟线程可以阻塞等待而不占用载体线程，是"合法阻塞点"。

混合编程风险：

• 在 Reactor 非阻塞链中调用 block() 会抛 blockAllowed() == false 异常（IllegalStateException），这是框架对非阻塞线程调 block 的保护。
• VT 中 block() 不影响载体线程，但每次阻塞都经历挂起/恢复调度，存在开销，不应滥用。
• 最佳实践：IO 密集用 VT、背压/流式用 Reactor，两者通过 fromCallable / block 在边界收敛，禁止在 Reactor 链内裸调阻塞方法。

这道题考察对 "Reactive Manifesto 四大原则如何从理论落地为全栈响应式架构" 的系统设计能力。核心不在背诵原则，而在回答：响应式系统的四大原则（Responsive 响应/Resilient 弹性/Elastic 弹性伸缩/Message Driven 消息驱动）如何在技术栈中具体落地？什么是全栈响应式——从 Web 到数据访问到缓存到服务间调用全部非阻塞？全链路背压如何从客户端一路传导到数据库？在 Virtual Threads 时代，响应式系统架构该如何演进？这是架构师整合前 5 题知识点的综合设计题。

系统架构的响应式演进：

单体架构（同步阻塞）：所有模块在一个进程内同步调用，Thread-per-Request 模型。简单但扩展性差，一处阻塞影响全局，资源利用率低。

微服务（同步 RPC）：服务拆分后通过 HTTP/RPC 同步调用，线程池等待下游响应。解决了扩展性但引入了分布式阻塞链——A 调 B 调 C，每跳线程都在等，线程资源沿调用链累积浪费，且一处慢拖垮全链（雪崩）。

响应式微服务（全栈非阻塞）：基于 Reactive Manifesto（2014，Jonas Bonér 等提出）四大原则——Responsive（快速响应）、Resilient（弹性/容错）、Elastic（弹性伸缩）、Message Driven（消息驱动）。全栈用 WebFlux + R2DBC + Reactive Redis + WebClient，背压从客户端经 Web 层、服务层、数据层一路传导，慢消费者自动降速不堆积。少量线程承载高并发，资源利用率极高。

云原生响应式：响应式微服务 + K8s 弹性伸缩（按指标自动扩缩 Pod）+ Service Mesh（Istio/Linkerd 非阻塞 sidecar）+ 事件驱动（Kafka/Pulsar 异步解耦）。消息驱动从"请求-响应"升级为"事件流"，进一步解耦与削峰。

演进驱动力：① 资源效率——微服务同步调用链的线程浪费在云原生规模下放大，响应式用少量线程降本；② 弹性容错——Reactive Manifesto 的 Resilient 原则要求隔离舱/熔断/降级，避免局部故障雪崩；③ 背压全链路——慢消费者能反压上游，避免大流量压垮系统；④ 云原生弹性——响应式应用启动快、内存占用低，适合容器化弹性伸缩。

Reactive Manifesto 四大原则落地：

1. Responsive（响应性）：系统在任何情况下都快速响应。落地：非阻塞全栈（WebFlux/R2DBC/WebClient）保证 IO 等待不占线程；超时控制（timeout()）快速失败；降级（onErrorResume）保可用。

2. Resilient（弹性/容错）：部分故障不蔓延。落地：隔离舱（Bulkhead）——不同业务用独立线程池/连接池隔离，避免互相拖垮；熔断器（Circuit Breaker）——Resilience4j CircuitBreakerOperator，错误率超阈值熔断走降级；重试退避——retryWhen(backoff()) 瞬时故障自愈。

3. Elastic（弹性伸缩）：随负载自动扩缩。落地：响应式应用内存占用低、启动快，适合 K8s HPA（Horizontal Pod Autoscaler）按 CPU/内存/自定义指标扩缩；无状态设计 + 响应式 Redis 共享会话。

4. Message Driven（消息驱动）：组件间异步消息通信。落地：服务内用 Reactor 的 Mono/Flux 消息流；服务间用 Kafka/Pulsar 事件流异步解耦；全链路背压——慢消费者 request(n) 反压上游，消息队列作为缓冲削峰。

全链路背压传导：这是响应式系统的核心价值。客户端消费慢 → Netty 写缓冲满 → WebFlux 不读取新数据 → R2DBC 不从数据库拉取 → 数据库 socket 缓冲满暂停发送。背压从客户端经 Web 层、服务层、数据层一路反向传导，整个链路自动降速，无任何环节堆积 OOM。这是命令式架构无法实现的——命令式是 Push 模型，生产者不管消费者速率。

全栈响应式技术选型：Web 层 WebFlux（Netty EventLoop）→ 服务层 Reactor（Mono/Flux 编排）→ 数据层 R2DBC（非阻塞数据库）→ 缓存层 Reactive Redis（Lettuce）→ 服务间调用 WebClient（非阻塞 HTTP）→ 消息层 Reactive Kafka（Reactor Kafka）。任一环节阻塞都会破坏全链路。

命令式架构 vs 响应式架构对比：

全栈响应式架构示例：某租赁平台系统——WebFlux + R2DBC + Reactive Redis + WebClient + 限流背压协同 + Micrometer 可观测。

// 全栈响应式: WebFlux + R2DBC + Reactive Redis + WebClient
@Service
public class RentReactiveService {

    private final VehicleRepository vehicleRepo;      // R2DBC
    private final ReactiveRedisTemplate<String, Vehicle> redis;  // Reactive Redis
    private final WebClient pricingClient;             // 非阻塞 HTTP
    private final OrderRepository orderRepo;             // R2DBC 订单
    private final TransactionalOperator txOp;         // 响应式事务

    public Mono<RentResult> rent(String code, String userId) {
        // 1. 先查 Reactive Redis 缓存 (非阻塞)
        return redis.opsForValue().get("vehicle:" + code)
            // 缓存未命中查 R2DBC
            .switchIfEmpty(vehicleRepo.findByCode(code)
                // 缓存回写必须绑定到响应式链, 禁止手动 subscribe() 破坏 WebFlux 请求契约
                .flatMap(v -> redis.opsForValue()
                    .set("vehicle:" + code, v, Duration.ofMinutes(10))
                    .thenReturn(v)))
            // 2. 并发调用定价服务 (WebClient 非阻塞)
            .flatMap(vehicle -> pricingClient.get()
                    .uri("/price/{code}", code)
                    .retrieve()
                    .bodyToMono(BigDecimal.class)
                    .timeout(Duration.ofSeconds(2))
                    .onErrorReturn(vehicle.getDailyRate())  // 降级用本地价
                    .map(price -> new RentResult(vehicle, price, userId)))
            // 3. 响应式事务: 更新车辆状态 + 创建订单
            .flatMap(result -> doRentInTransaction(result).as(txOp::transactional))
            // 4. 指标埋点 (Micrometer Reactive)
            .doOnSuccess(r -> Metrics.counter("rent.success").increment())
            .doOnError(e -> Metrics.counter("rent.failure").increment());
    }

    private Mono<RentResult> doRentInTransaction(RentResult r) {
        // 事务边界内: 更新车辆状态 + 创建订单, 二者同生同灭 (由上层 .as(txOp::transactional) 包裹)
        return vehicleRepo.findByCode(r.getCode())
            .flatMap(v -> {
                v.setAvailable(false);
                return vehicleRepo.save(v)
                    .then(orderRepo.save(new RentOrder(v.getId(), r.getUserId())));
            })
            .thenReturn(r);  // 事务成功返回结果
    }
}

// 限流与背压协同: RateLimiter + 背压
@Configuration
public class RateLimitConfig {
    @Bean
    public RateLimiter rentRateLimiter() {
        // Resilience4j 响应式限流: 每秒 100 请求
        return RateLimiter.of("rent", RateLimiterConfig.custom()
            .limitForPeriod(100)
            .limitRefreshPeriod(Duration.ofSeconds(1))
            .build());
    }
}

// WebFilter: 限流 + 背压, 超限返回 429
@Component
public class RateLimitFilter implements WebFilter {
    private final RateLimiter limiter;
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        if (limiter.acquirePermission()) {
            return chain.filter(exchange);
        }
        exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
        return exchange.getResponse().setComplete();
    }
}

追问 1：全栈响应式的落地难点是什么？

五大难点：① 生态约束——全栈要求每个组件都有响应式驱动，但部分中间件（如旧版 Elasticsearch 客户端、部分云服务 SDK）只有阻塞客户端，强行用 boundedElastic 包装是"假响应式"；② 调试与排障——Reactor 操作符栈不反映业务代码行，异常难定位，需大量 checkpoint() + Hooks.onOperatorDebug()，团队需重新学习排障方法论；③ 事务与上下文——@Transactional 和 ThreadLocal 失效，需用 TransactionalOperator 和 Reactor Context 重写所有横切逻辑；④ 团队心智成本——响应式编程范式与命令式差异大，团队需系统培训，误用（阻塞调用、多次订阅、无背压）导致难以排查的生产事故；⑤ 混合架构边界——与遗留同步系统对接时，block()/fromFuture 边界设计不当会引入阻塞或性能损耗。建议：非必要不全栈响应式，评估 Virtual Threads + WebMVC 能否满足吞吐需求，流式背压场景才上全栈响应式。

追问 2：响应式与 Virtual Threads 混合架构如何设计？

混合架构是当前的务实选择——取二者所长。设计原则：① Web 层用 Virtual Threads——WebMVC + spring.threads.virtual.enabled=true，请求处理和同步 IO（JDBC/阻塞 HTTP）跑虚拟线程，简单高效；② 流式/背压层用 Reactive——SSE 推送、大文件流式传输、实时数据流用 WebFlux + Flux，利用背压控速；③ 边界桥接——Virtual Threads 侧调 Reactive 侧用 mono.block()（虚拟线程阻塞安全），Reactive 侧调 Virtual Threads 侧用 Mono.fromCallable(() -> syncCall()).subscribeOn(boundedElastic()) 或 Schedulers.fromExecutor(virtualThreadExecutor)；④ 消息层统一——Kafka/Redis Stream 作为两侧的异步通信枢纽，生产侧和消费侧各自用同步或响应式，解耦编程模型；⑤ 可观测性统一——Micrometer 同时支持同步和响应式埋点，用 OpenTelemetry 统一 Trace 串联两种模型的调用链。核心：不要强迫统一编程模型，按场景选最优——IO 密集 CRUD 用 Virtual Threads，流式背压用 Reactive，二者通过消息总线和边界桥接协作。

架构落地离不开质量保障——响应式链无法用传统 JUnit 直接验证，StepVerifier 是 Reactor 官方测试框架，也是必须掌握的工具。

• 基本用法：StepVerifier.create(Flux.just(1, 2, 3)).expectNext(1, 2, 3).verifyComplete()——创建 → 断言元素 → 断言完成信号，测试才会真正触发订阅。
• 虚拟时间（Virtual Time）：StepVerifier.withVirtualTime(() -> Flux.interval(Duration.ofSeconds(1)).take(3))——虚拟时钟立即推进，无需真实等待；thenAwait(Duration.ofSeconds(10)) 虚拟推进 10 秒，适合测试延迟、定时器。
• 测试背压：create(source, 0) 初始 request 0，thenRequest(n) 分步拉取，thenCancel() 模拟下游取消订阅——用于验证背压传导是否正确。
• 测试异常：expectError(RuntimeException.class) 验证异常类型，expectErrorMessage("msg") 验证异常消息。
• 时序断言：expectNoEvent(Duration.ofMillis(100)) 验证指定窗口内无事件，用于超时、限流等场景。

// 虚拟时间 + 时序断言 + 取消 综合示例
StepVerifier.withVirtualTime(() -> Flux.interval(Duration.ofSeconds(1)).take(3))
    .expectSubscription()
    .expectNoEvent(Duration.ofSeconds(1))  // 1 秒内无事件
    .thenAwait(Duration.ofSeconds(1))
    .expectNext(0L)
    .thenCancel()                       // 模拟下游取消
    .verify();

生产实践建议：

• 覆盖率：每个操作符链至少覆盖一个成功场景（verifyComplete()）和一个异常场景（expectError()）。
• 边界测试：空 Flux、单元素 Flux、无限 Flux（用 thenCancel 截断）。
• 时序测试：配合 withVirtualTime + thenAwait，避免测试依赖真实时钟导致慢且不稳定。

上线门禁：背压与超时、Scheduler 隔离、block 禁区审查。表达提纲：信号语义→操作符链证据→线程模型。

按 Reactor Core Reference：Flux<T> 表示异步序列 0..N 元素；Mono<T> 表示异步结果 0..1（单值或空）。类型携带基数语义——HTTP 单次响应应是 Mono，SSE/列表流应是 Flux；改变最大基数的操作符会切换类型（如 Flux#count → Mono<Long>）。

本题要答清：① 为何不能「一律 Flux」；② 创建序列时 just/defer/from*/create/generate 的官方选择表；③ Cold 订阅语义与「已有值却用 just 捕获过早」的坑。

Reactive Streams 只规定 Publisher；Reactor 用 Mono/Flux 双类型把「基数」编码进 API，减少误用（对 Mono 做无意义的 buffer 等）。官方附录 Which operator do I need? 把创建、变换、过滤、错误、时间、拆分、回同步世界、多播分成九类决策树——工程表达应以该附录为索引，而不是背零散 API 名。

Mono 信号约束（官方）：成功路径是至多一次 onNext 后 onComplete，或仅 onError；禁止 onNext+onError 组合。Mono.never() 是例外（无信号，多用于测试）。

基数切换示例：Mono 上 flatMapMany → Flux；Flux 上 next/single/collectList → Mono。

// 懒捕获：每次订阅读最新配置（官方：just 会立即捕获）
Mono<Cfg> cfg = Mono.defer(() -> Mono.just(loadCfg()));

// Optional / null 安全
Mono<User> u = Mono.justOrEmpty(repo.find(id));

// 资源生命周期
Flux<String> lines = Flux.using(
    () -> new BufferedReader(...),
    br -> Flux.fromStream(br.lines()),
    BufferedReader::close);

// Controller：单资源 Mono，事件流 Flux
@GetMapping("/orders/{id}")
Mono<Order> one(@PathVariable String id) { return service.find(id); }

@GetMapping(value="/events", produces=MediaType.TEXT_EVENT_STREAM_VALUE)
Flux<Event> stream() { return hub.stream(); }

评审清单：返回类型基数是否与契约一致；有副作用的源是否 defer；文件/连接是否 using。

追问 1：为何 HTTP 客户端结果用 Mono 而不是 Flux.just？

官方说明：单次请求只有一个响应，Mono 只暴露 0..1 相关操作符，避免误用 count/buffer 等；语义上与「可空异步结果」一致。

追问 2：just 与 defer 差在哪？

just(x) 在组装链时就捕获 x；若 x 来自可变状态或应每订阅计算一次，须 defer/fromSupplier。这是配置热更新、Request 作用域 Bean 的高频坑。

追问 3：create 与 generate 怎么选？

generate：同步、每次回调至多一个 onNext，适合状态机。create：可多线程多发，适合把监听器/回调桥进 Reactor，须处理背压（OVERFLOW 策略）。

对外 API 契约写明 Mono/Flux；OpenAPI/文档与实际返回类型不一致是联调事故源。可观测：对 Cold 源多次订阅会放大下游 QPS——入口做 cache/share 要有意识（见 C15.11）。

参考：Reactor Core Features；Which operator do I need?§1 Creating。

官方附录「Transforming」：1-to-1 用 map；1-to-n 或异步 Publisher 返回用 flatMap；要保留顺序用 flatMapSequential 或 concatMap；「只关心最新」用 switchMap。混淆它们是响应式链顺序错乱、下游放大、取消泄漏的根因。

早期资料常只写 flatMap≈mergeMap。Reactor 把并发、顺序、取消语义拆到不同操作符：flatMap（并发、不保序）、concatMap（串行、保序）、flatMapSequential（立刻订阅但重排输出）、switchMap（切换时取消前一个内层）。Mono 还有 flatMapMany 升到 Flux。

空内层：flatMap 中返回 Mono.empty() 可「忽略某些数据」（官方原文）。聚合到单值用 collectList/reduce/count（结果为 Mono）。

// 并发拉取详情，限制并发度 16（官方 flatMap 第二参数）
Flux<Detail> details = ids.flatMap(id -> client.get(id), 16);

// 必须按 ID 顺序写库
Flux<Void> writes = ids.concatMap(id -> repo.save(load(id)));

// 搜索框：每次击键取消上一次请求
Flux<Result> search = keystrokes.switchMap(q -> api.search(q));

// 并发执行但按原序合并结果
Flux<T> ordered = src.flatMapSequential(this::asyncCall, 8);

// Mono → 多行
Flux<Item> items = orderMono.flatMapMany(o -> Flux.fromIterable(o.getItems()));

选型口诀：同步纯函数 → map；IO 且可乱序 → flatMap+并发度；IO 且保序 → concatMap 或 flatMapSequential；UI/搜索最新值 → switchMap。

追问 1：flatMap 与 merge 关系？

官方把「发射顺序合并」归在 combine 段的 merge；flatMap 是「每元素映射为 Publisher 再 merge」。无并发上限的 flatMap 近似无限并发 merge，生产必须显式 concurrency。

追问 2：为何分页遍历常用 concatMap？

页与页有先后依赖或必须保序落库时，concatMap 保证前页完成再订后页；flatMap 会打乱页序并可能压垮下游。

追问 3：switchMap 取消会导致什么？

上一次 HTTP/DB 订阅被 cancel；若内层有不可取消副作用，需幂等或改用 concatMap/flatMapLatest 语义评估。搜索联想是经典正向场景。

代码评审检查表：每个 flatMap* 写明「是否保序 / 并发度 / 失败是否吞掉 empty」。压测观察：内层超时与取消是否与 switchMap 预期一致。参考官方附录 §2 Transforming。

官方把「组合多个 Publisher」与「拆分 Flux」分列：组合看齐步（zip）、交错（merge）、最新组合（combineLatest）、串接（concat）；拆分看边界（buffer/window）与键（groupBy）。选错会导致「短板卡死」「内存窗口爆炸」「分组无界」。

组合语义对齐 Rx 家族，但 Reactor 强调与 Mono 的 zip/when/and 终点语义（等完成而非元素流）。拆分上 window 产出 Flux<Flux<T>>，buffer 产出 Flux<List<T>>——背压与内存画像不同。

空序列回退：defaultIfEmpty（值）与 switchIfEmpty（另一 Publisher）——官方明确区分。

追问 1：zip 一方永不发射会怎样？

下游永久等待（除非 timeout）。生产 zip 必须给各方 timeout/默认值。

追问 2：buffer 与 window 怎么选？

要 List 批处理用 buffer；每窗还要再 flatMap 流式处理用 window，避免大 List 堆峰值。

追问 3：groupBy 的隐患？

官方 TIP：返回 GroupedFlux；key 高基数会创建大量内层流，必须限制并行 group 数并处理慢组反压。

组合点统一加 timeout 与指标（各方 RT）。微批参数与 DB 批量上限、锁等待一起评审。参考官方附录 §2 Combining / §7 Splitting。

官方 §4–§6：过滤（filter/take/skip/sample/single）、错误（onErrorReturn/Resume/Complete/Map、retry/retryWhen、timeout）、时间（delay*/interval/timed）、以及背压溢出策略（onBackpressure*）。与 C15.04 错误策略卡互补——本卡按操作符决策表落地。

重试从简单 retry(n) 演进到 retryWhen(Retry.backoff)（抖动退避）。背压从「无界缓冲默认风险」到显式 Drop/Latest/Buffer/Error。超时是错误的一种触发源，常与 onErrorResume 组合。

client.get(id)
  .timeout(Duration.ofSeconds(2))
  .retryWhen(Retry.backoff(3, Duration.ofMillis(100))
      .filter(ex -> ex instanceof TimeoutException))
  .onErrorResume(ex -> cache.get(id));

Flux.interval(Duration.ofMillis(10))
  .onBackpressureDrop(i -> metrics.dropped())
  .concatMap(this::handle);

与 G06.01 超时预算对齐：链式 timeout 可加总；重试次数计入总预算。

追问 1：onErrorReturn 与 onErrorResume 怎么选？

固定值 → Return；要查缓存/调降级服务 → Resume（可按异常类型分支）。

追问 2：Drop 与 Latest？

Drop 丢弃溢出；Latest 只保留最新一个。行情/指标常用 Latest；审计日志不能丢则应 Buffer 有界+限速，而非 Drop。

追问 3：take(n) 的 request 行为？

官方区分 take 是否向上游请求无界；测试背压时要用 StepVerifier 显式 request（见 C15.06 扩展）。

统一错误分类：可重试（超时/5xx）vs 不可重试（4xx 业务）。背压策略写进接口 SLA。参考官方附录 §4–§6 与 C15.04。

线程模型：subscribeOn 影响源订阅线程，publishOn 影响下游执行线程。多播：share/publish/cache/replay（官方 §9）。回同步世界：block/blockFirst/toFuture（§8）——在标为 non-blocking 的 Scheduler 上会抛 UnsupportedOperatorException。

elastic() 已弃用移除，IO 用 boundedElastic()。Context 替代 ThreadLocal 做跨 publishOn 传播。调试用 checkpoint/Hooks.onOperatorDebug（与 C15.06 呼应）。

Context：contextWrite 影响上游读取；用 deferContextual 读取。放错位置读空是高频坑（C15.01 已述，本卡强调与调度联用）。

追问 1：两个 publishOn 谁说了算？

下游最近的 publishOn 决定后续段线程；subscribeOn 通常放靠近源，且多次 subscribeOn 时最接近源的生效（以实际链为准，表达时画线程切换图）。

追问 2：cache 与 share 区别？

cache 重放历史（可带 TTL/大小）；share 偏「现在开始共享」，迟到者不看历史。

追问 3：为何 parallel()/single() 上 block 会炸？

官方 §8：这些 Scheduler 标为 non-blocking only，block 会 UnsupportedOperatorException——防止在 EventLoop 类线程上阻塞。

边界清单：WebFlux 入口禁 block；与 VT/MVC 桥接时 block 必须有超时；boundedElastic 活跃线程进看板。参考官方 Core Features（Schedulers）与附录 §8–§9。

背压断了会在某段无界缓冲里堆内存或丢请求。要会认症状并找到可关的缓冲点。

Reactive 链要求需求信号向上游传递；无界 buffer、阻塞调用藏在 flatMap 都会切断背压。

症状：下游慢但上游仍高速进；某 operator 后队列长度狂涨；延迟升高但 CPU 不高。动作：降上游速率、缩小缓冲、启用 drop/latest（需业务允许）、扩下游。

旋钮：prefetch、缓冲容量、onBackpressure* 策略、连接池 max。先临时限流，再修传递链。

失败模式：无界 buffer 当「解法」；drop 未告知业务；阻塞调用藏在 flatMap。回滚点：恢复旧缓冲策略并限流上游。

背压未传递常见于：边界用了无界队列、block 在弹性线程、下游超时被吞。症状是内存涨而错误率不高。

临时止血：降产/限流/切有界队列；长期把背压写成契约并做拒绝可观测。

量化信号：某 operator 后队列长度 5 分钟内单调上涨，且内存曲线与错误率脱钩（内存涨而错误率不升），即判定背压断裂；prefetch 默认 256（flatMap 时再乘并发度）常是隐藏缓冲点。演进：把背压改成契约——每个跨边界点声明有界容量与 onBackpressure 策略并暴露拒绝计数；drop / latest 须业务书面允许后启用，且拒绝必须可观测。