响应式编程 Reactor 实战：异步数据流与背压

响应式编程（Reactive Programming）用 Reactor 的 Mono / Flux 把传统阻塞调用改造成异步数据流，靠背压（Backpressure）让慢消费者不被快生产者压垮。本文结合 Spring WebFlux 落地，讲清核心模型与实战避坑。

一、为什么需要响应式

传统 Spring MVC 是”一请求一线程”：每个 HTTP 请求占用一个 Servlet 线程，遇到数据库查询、远程调用等阻塞 IO 时，线程只能干等。当并发上来、下游变慢，线程池被占满，整个应用失去响应。响应式编程不消灭阻塞，而是把时间线从”线程等待”变成”数据流动”——上游产出数据就沿管道推送，谁慢谁先背压，线程始终在干活。

一个直观的对比：假设一条接口要串行调用三个下游服务，每个耗时 200ms。命令式模型下这条请求至少占用线程 600ms；而响应式模型在等待第一个响应时，事件循环线程已经去处理别的请求了，等数据回来再接着走后续操作符。高并发下，前者要靠堆线程数硬抗，后者用几百个事件循环线程就能撑起上万并发连接，资源利用率天差地别。

二、Reactor 核心：Mono 与 Flux

Reactor 提供两个发布者（Publisher）：Mono 发出 0~1 个元素，对应单结果查询；Flux 发出 0~N 个元素，对应列表、流、事件源。它们默认是”冷”的——只有被订阅（subscribe）才会真正执行，这点与 Java 8 Stream 不同：Stream 立即执行，Reactor 是声明式、惰性的。

Flux<String> flux = Flux.just("a", "b", "c")
        .map(s -> s.toUpperCase())
        .filter(s -> !s.equals("B"));

Mono<List<String>> mono = flux.collectList();

// 订阅才会触发整条链路执行
flux.subscribe(
    v -> System.out.println("onNext: " + v),
    err -> System.err.println("onError: " + err),
    () -> System.out.println("onComplete")
);

三、操作符链：map / flatMap / filter

响应式代码几乎全是操作符（operator）链式拼接。map 做同步 1:1 转换；flatMap 把每个元素映射成一个新的 Publisher 并合并，常用于”拿到 ID 再去查详情”这类异步扇出；filter 做过滤。注意 flatMap 默认不保序，需要保序用 concatMap 或 flatMapSequential。操作符本身不会触发执行，只有终端 subscribe 才会让整条声明好的管道真正流动起来，这也是排查”代码没跑”类问题时的第一着眼点。

Flux<Order> orders = orderRepo.findByUserId(userId)
        .flatMap(order -> itemRepo.findByOrderId(order.getId()))  // 异步扇出
        .filter(item -> item.getPrice() > 100)
        .map(this::enrich);                                       // 同步转换

四、背压：响应式真正的护城河

背压（Backpressure）是响应式区别于”异步回调地狱”的关键：下游能告诉上游”我一次只吃得下 N 条，多了先缓存或丢弃”。没有背压的异步只是把阻塞从线程转移到了内存队列，最终仍是 OOM。Reactor 提供 onBackpressureBuffer、onBackpressureDrop 与 limitRate 等策略。

Flux.range(1, 1_000_000)
        .onBackpressureBuffer(1024,
            dropped -> log.warn("dropped: {}", dropped))  // 缓冲 1024，溢出回调
        .limitRate(128)                                   // 每次向上游要 128
        .publishOn(Schedulers.boundedElastic())
        .subscribe(v -> slowConsumer.accept(v));

背压策略没有绝对好坏，取决于业务能否容忍丢数据。onBackpressureBuffer 适合”一条都不能少”的计费、订单场景，但要设上限否则仍是内存炸弹；onBackpressureDrop 适合监控指标、日志这类”丢一点无妨”的遥测流；limitRate 则最温和，通过”批量拉取”把上游生产节奏削平。实测中，对一条每秒 5 万条的事件流不加重压，下游消费者几秒内就会被积压拖垮，加上 limitRate(128) 后 CPU 曲线立刻平滑下来。

五、线程模型：Schedulers 与上下文切换

Reactor 默认在调用者线程执行，必须用 Schedulers 显式切换。subscribeOn 决定”订阅发生在哪个线程”（影响源头），publishOn 决定”其后操作符在哪个线程跑”（可多次切换）。IO 密集放 boundedElastic，计算密集放 parallel。这与 Java 21 虚拟线程 的”轻量载体线程”思路异曲同工，也常和 Spring Boot 异步线程池 配合做混合架构。

Mono.fromCallable(() -> blockingDbCall())      // 阻塞调用
    .subscribeOn(Schedulers.boundedElastic())  // 移到弹性线程池
    .publishOn(Schedulers.parallel())          // 后续计算在并行线程
    .map(this::transform)
    .subscribe(result -> log.info("done: {}", result));

六、Spring WebFlux 实战：一个非阻塞接口

落到 Web 层，Controller 方法直接返回 Mono / Flux，框架负责把响应式流适配成 HTTP 响应（Server-Sent Events 甚至能天然支持流式推送）。下面的仓储用响应式驱动（R2DBC），全程无阻塞线程等待。

@RestController
@RequestMapping("/api/users")
public class UserController {
    private final UserRepository repo;          // 响应式 R2DBC 仓储

    @GetMapping("/{id}/orders")
    public Flux<OrderDTO> orders(@PathVariable String id) {
        return repo.findById(id)
                   .flatMapMany(u -> orderRepo.findByUserId(u.getId()))
                   .map(OrderDTO::from);
    }
}

七、踩坑清单与可观测性

响应式最大的坑是”用了 WebFlux 却到处 .block()“——一旦在链路上调用阻塞方法，非阻塞优势瞬间归零，还容易死锁。其次是异常必须用 onErrorResume / doOnError 处理，否则错误会静默吞掉。生产环境务必接入链路追踪，推荐与 Spring Cloud 微服务治理 的熔断、限流组合，并在 GitHub Actions 流水线里把响应式单测纳入门禁；需要跨服务流式调用时，可配合 gRPC 微服务通信 做高性能 RPC。

• 不要混用阻塞：在 Flux 链里调用 JDBC、RestTemplate 等阻塞 API，务必 subscribeOn(boundedElastic) 隔离。
• 谨慎 flatMap 并发度：flatMap(fn, concurrency) 第二参数控制扇出并发，默认 256，容易打爆下游。
• 保留上下文：跨线程切换后 MDC 日志丢失，用 contextWrite 透传 traceId。

八、响应式 vs 虚拟线程：怎么选

Java 21 的虚拟线程让”一个请求一个线程”也能做到近乎无限的轻量并发，不少人因此质疑响应式是否还有必要。结论是两者解决不同层级的问题：虚拟线程降低”写阻塞代码”的成本，但并未改变”线程在 IO 等待期间仍被占用”的本质，遇到海量长连接（如 WebSocket 推送、SSE 流）依然吃紧；响应式则从数据模型层面消除等待，更适合网关、流式聚合、实时推送这类场景。实际架构里两者常共存——用虚拟线程承载传统阻塞业务，用响应式承载高扇出网关。

小结

响应式编程不是银弹，它用更高的认知成本换取 IO 密集场景下的高吞吐。掌握 Mono / Flux、背压策略与 Schedulers 线程模型三块，再配合 Spring WebFlux 落地，你的接口就能在有限线程下扛住更高的并发洪峰。落地时记住三条铁律：不在链上随意 block()、用 limitRate 兜底背压、异步调用务必隔离到 boundedElastic。先在小流量网关试水，再逐步向核心链路推广，才是最稳的演进路径。