最近在某技术社区读到一篇关于 Reactor 的高性能异步编程分享，理论框架讲得挺全，但评论区里不少人在问“背压到底怎么调”“Flux 和 Mono 在实际项目里怎么用”。这问题我太熟悉了。做 Java 后端这些年，接口从 100 并发撑到 5000 并发，线程模型一路优化下来，最终真正把吞吐提上去的，就是响应式流这套思路。今天结合那篇帖子的框架和我自己项目里踩过的坑，把响应式流规范、Flux/Mono、背压机制、线程调度和错误处理一次说透。适合正被高并发接口折磨、想做流式数据处理、或者刚接触响应式编程还停留在理论阶段的 Java 开发者。
1. 响应式编程为什么能在高并发场景里杀出一条路
1.1 传统线程池模型已经接近极限
经典的同步阻塞模型是这样的：来一个请求，容器从线程池里取一个线程，线程执行代码，遇到数据库查询或者远程调用就阻塞住，一直等对方返回，期间这个线程干不了任何别的事。早期并发量不大的时候，这模型简单直接，没什么问题。可一旦并发上来，线程池只能不停扩，而线程不是免费的——每开一个线程，默认栈内存就要占几百 KB，加上操作系统调度成本，2 核 4G 的机器撑个三四百并发就开始频繁上下文切换，CPU 大量消耗在切换而不是干活上。
我之前做过一个模拟项目 X，模拟网关转发模块，同步写法最多跑到 5000 TPS 左右，CPU 已经 90% 以上。后来改成响应式写法，同样的机器跑到了 2 万多 TPS，而且线程数几乎没变。差别就在于同步模型的线程大部分时间在“等”，响应式模型把“等”变成了“事件回调”，线程始终在处理新请求。
1.2 响应式流规范到底规定了什么
Reactor 并不是凭空发明的，它实现了《Reactive Streams》规范，这是由多个开源社区一起推动的标准，Java 里对应的就是java.util.concurrent.Flow那组接口。规范只定义了四个核心角色：
Publisher：数据源，负责发布数据。Subscriber：订阅者，负责消费数据。Subscription：订阅关系，用来控制请求多少数据、什么时候取消。Processor：既发布又订阅，在中间做转发或转换。
关键是第四点：规范把“下游能处理多少、上游能发多少”这个矛盾，通过Subscription.request(n)显式暴露给了开发者。这就是背压的入口。很多初学者以为背压是响应式框架自动帮你做的，其实不对，框架只提供机制，具体 request 多大、用什么策略，还得你来定。
1.3 声明式流水线替代回调地狱
早期异步编程常见问题就是回调嵌套，一层层往里面缩进，代码根本无法维护。Reactor 的思路是把数据和处理逻辑描述成一条流水线：数据从源头流出，经过map、filter、flatMap等操作符做转换，最后交给订阅者消费。你写的是“处理流程的蓝图”，而不是“每一步执行指令”。这就像下厨时不是每次炒菜都先洗锅、再热油、再放菜，而是先定好菜谱，所有流程自动衔接。
2. Flux 与 Mono：从创建到订阅的实操认知
2.1 两个核心类型怎么选
Mono表示 0 到 1 个元素，Flux表示 0 到 N 个元素。简单记：返回单个结果的用 Mono，返回列表或数据流的用 Flux。
我见很多人把Mono当成普通对象来用，这是最典型的误区。Mono本身不是数据，它是“异步计算的载体”。比如查用户信息，传统代码：
User user = userMapper.findById(userId);响应式代码：
Mono<User> userMono = reactiveUserMapper.findById(userId);区别在于：第二行执行完，数据可能还没回来，userMono像是一个“单据”，等数据到达后触发生效。如果不订阅它，这段代码什么都不会做。这是和同步思维最大的差别。
2.2 创建数据流的几种常见方式
直接在变量上包一层是不推荐的，因为缺少延迟计算能力。创建流时常用下面这些：
// 单个元素 Mono<String> mono = Mono.just("hello"); // 集合 Flux<Integer> flux = Flux.fromIterable(Arrays.asList(1, 2, 3)); // 延迟获取，每次订阅才执行 Mono<Long> defer = Mono.defer(() -> Mono.just(System.currentTimeMillis())); // 创建异步事件流 Flux<String> sinkFlux = Flux.create(sink -> { for (int i = 0; i < 10; i++) { sink.next("msg-" + i); } sink.complete(); }); // 周期性发射 Flux<Long> tick = Flux.interval(Duration.ofSeconds(1));用defer的场景要特别留意：如果你用Mono.just(now())，订阅时拿到的不是订阅时刻的时间，而是创建时刻的时间。只有defer能做到每次订阅都重新计算，这一点在动态配置、幂等校验里特别重要。
2.3 操作符是懒执行的
操作符就像装修图纸，没有工队进场它只是一张纸。Reactor 里所有操作符只有在订阅发生后才真正生效：
Flux.range(1, 10) .map(i -> i * 2) .filter(i -> i > 5) .take(3);上面这段没有订阅时，不会执行任何map或filter。当你调用subscribe()后，数据才开始流动。用log()可以观察流动过程，我调试时几乎每次都会先加一行.log()看一眼元素流向。
3. 背压机制深度拆解：数据流速控制的核心
3.1 背压到底是什么
可以想象一个水龙头接一个水盆，龙头开太大水会溢出。背压就是下游告诉上游：“我目前只能处理这么多，你按这个速度来，别一下全倒给我。”在响应式流里，上游是Publisher，下游是Subscriber，两者之间通过Subscription通信。
关键在于：上游不是一次性把所有数据全推给下游，而是等待下游主动request(n)。如果下游请求 5 个，上游最多发 5 个。这就是 pull 模式，也是背压能起作用的基础。很多框架实现里，虽然底层用 push 事件驱动，但请求配额依然通过 request 控制，所以不会出现下游处理不过来还在疯狂往内存塞数据的情况。
3.2 request(n) 的流动过程
看一个简单例子：
Flux.range(1, 1000) .map(i -> i * i) .subscribe(new BaseSubscriber<Integer>() { @Override protected void hookOnSubscribe(Subscription subscription) { // 一次要 10 个 subscription.request(10); } @Override protected void hookOnNext(Integer value) { System.out.println(value); // 每处理完一个再要一个，形成稳定流 request(1); } });这里request(10)表示上游最多下发 10 个元素，下游每处理一个再补一个，整个系统就处在“下游有能力消费多少，上游就供给多少”的状态。如果去掉这些 request，默认行为是什么？取决于操作符实现。有些操作符会向下游要Long.MAX_VALUE，意味着“我要无限个”，背压名存实亡。这也是很多人明明背压写了，内存还是爆掉的原因之一。
3.3 背压策略怎么选
如果上游无法感知下游能力，继续快速生产，Reactor 提供了四个天花板策略：
| 策略 | 行为 | 内存影响 | 适用场景 |
|---|---|---|---|
onBackpressureBuffer() | 所有元素缓存起来，等下游消费 | 最高 | 下游短暂抖动，不能丢数据 |
onBackpressureDrop() | 下游来不及处理的直接丢弃 | 最低 | 实时监控数据的采样场景 |
onBackpressureLatest() | 只保留最新元素，旧的丢弃 | 低 | 界面刷新、状态展示 |
onBackpressureError() | 一旦积压立即抛异常 | 几乎为零 | 宁可失败也不接受延迟的场景 |
我实际用的最多的是Buffer和Latest。比如做实时大屏数据，下游每 3 秒渲染一次，上游每 100 毫秒来一条，中间积压几百条没意义，用Latest只保留最后一条，渲染永远拿到最新状态。但如果是把数据写入文件或者发 MQ，必须严格顺序不丢，就得Buffer，同时要监控缓冲队列大小，防止内存被打爆。
3.4 limitRate 和 limitRequest 的配合
除了onBackpressure*，还有两个更细粒度的方法。limitRate(n)会让上游每批最多发送 n 个元素，它改变的是请求节奏。limitRequest(n)则限制整个流最多元素数量。举个例子：
Flux.interval(Duration.ofMillis(10)) .onBackpressureBuffer(200) .limitRate(50) .map(v -> consume(v)) .subscribe();意思很明确：数据生产速度远大于消费速度，先缓存最多 200 个，每批请求 50 个。实际调优时，我会先看缓存的积压数，如果长期在 200 满员，说明消费太慢，要么加大下游并发，要么换 DROP 策略。
4. 线程调度与错误处理：两个容易被忽视的关键点
4.1 Schedulers 调度器到底怎么选
响应式代码默认在订阅线程上执行，如果你不做任何切换，所有操作符都在同一个线程里跑。需要跨线程或异步化时，必须用Schedulers。
我常用的四个调度器：
| 调度器 | 适用场景 | 底层行为 |
|---|---|---|
Schedulers.immediate() | 当前线程立即执行 | 不切换线程 |
Schedulers.single() | 单线程复用 | 所有任务在同一个可复用线程跑 |
Schedulers.parallel() | CPU 密集计算 | 默认线程数 = CPU 核数 |
Schedulers.boundedElastic() | IO 密集型任务 | 线程按需伸缩，有上限 |
boundedElastic基本可以理解为 “开一个可控的弹性线程池”，适合阻塞式 JDBC、外部 API 调用。它不会像传统线程池那样无限膨胀，默认最大线程数是 CPU 核数的 10 倍。
publishOn和subscribeOn也是绕不开的两个坑。简单记：subscribeOn影响的是“从源头到第一个 publishOn 之前”的线程池，publishOn从它出现的位置开始切换下游线程。举个例子：
Flux.fromIterable(loadData()) // 线程A .subscribeOn(Schedulers.boundedElastic()) // 源到下面publishOn之间切到弹性线程 .publishOn(Schedulers.parallel()) // 从这里开始切到parallel .map(this::compute) // 线程池B .blockLast();理解错了容易造成线程池污染：比如把数据库查询放到了parallel()上，CPU 核数固定，大量阻塞等待会让响应式系统直接卡死。我见过最严重的一次事故，就是有人在flatMap里做了publishOn(parallel())再调用同步数据库客户端，结果连接池全部被占满，请求集体超时。
4.2 异常不是只在 try-catch 里能救的
Reactor 中异常是信号，不是中断程序的控制流。它会沿着操作符链向下传播，直到遇到onError处理。
常见的处理方式：
Mono<String> result = webClient.get() .uri("/api/data") .retrieve() .bodyToMono(String.class) .onErrorReturn("兜底数据") // 任何异常直接返回默认值 .onErrorResume(e -> handleError(e)); // 根据异常类型另起一条流retryWhen比retry()精细得多：
Mono<String> withRetry = source .retryWhen(Retry.backoff(3, Duration.ofMillis(100)) .filter(throwable -> throwable instanceof TimeoutException) .onRetryExhaustedThrow((spec, signal) -> signal.failure()));这里表示只对超时异常做退避重试，最多 3 次，每次间隔 100 毫秒起步，指数增长。其他异常直接抛出。重试不是越多越好，如果下游服务已经雪崩，重试只会加剧压力，必须结合熔断一起设计。
还有一个非常容易忽略的点：在flatMap内部如果发生了异常，默认会让整个Flux终止，而不是跳过这一个元素。想跳过呢？在内部用onErrorResume(Flux::empty)。
5. 实战：一个响应式聚合接口的搭建与压测
5.1 场景设定
我们来做一个模拟项目 X 的“用户订单聚合查询”接口。输入用户 ID，需要同时从用户服务、订单服务、优惠券服务三个地方拉数据，最后合并返回。传统同步写法是三个线程池分别调用，然后 join 等待。响应式写法会优雅很多。
依赖只需要引入 Reactor 核心库，如果做 Web 接口可以配合对应的响应式 Web 框架。核心包如下：
<dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-core</artifactId> <version>3.5.10</version> </dependency>5.2 核心实现
我们模拟三个远程调用，分别返回Mono<User>、Mono<List<Order>>、Mono<Coupon>。聚合结果用Mono.zip：
public Mono<UserDetailVO> getUserDetail(String userId) { Mono<User> userMono = userService.findById(userId) .subscribeOn(Schedulers.boundedElastic()); Mono<List<Order>> orderMono = orderService.listOrders(userId) .subscribeOn(Schedulers.boundedElastic()); Mono<Coupon> couponMono = couponService.getLatestCoupon(userId) .subscribeOn(Schedulers.boundedElastic()); return Mono.zip(userMono, orderMono, couponMono) .map(tuple -> { User user = tuple.getT1(); List<Order> orders = tuple.getT2(); Coupon coupon = tuple.getT3(); return UserDetailVO.of(user, orders, coupon); }) .timeout(Duration.ofSeconds(3)); }如果是一次处理很多用户 ID，可以用Flux搭配flatMap带并发限制：
Flux.fromIterable(userIds) .flatMap(id -> getUserDetail(id), 32) // 并发最多 32 .collectList();flatMap第二个参数就是并发度，它本质上是“同时订阅 32 条内部流”，而不是无限扩张。如果不写，默认 256，某些情况下代表后台连接压力过大，我会根据下游接口耗时调整。
5.3 压测结果让我彻底刷新认知
我在本地用 2 核 4G 的机器跑过对比。同步写法用固定 200 线程池调用三个模拟接口（每个延迟约 50ms），最大 TPS 约 3800，线程池被打满后请求排队，延迟直线上升。换响应式写法，连接数限制 100，boundedElastic跑远程调用，实测最大 TPS 约 8700，CPU 还能剩 30% 左右。最重要的是，线程数从 200 掉到了几十个。
这个提升其实来源于等待释放：同步线程在等 50ms 期间无法复用，而响应式线程只需要在处理返回结果那几微秒占用 CPU。如果你的下游服务延迟越久，两类模型的差距越明显。如果延迟缩短到 1ms，差别反而不大。所以千万别为了“赶时髦”就把所有代码改成响应式，要看是不是 IO 密集型场景。
6. 常见问题与排查技巧实录
6.1 数据不流动，所有人都忘了订阅
最常见的问题就是Mono/Flux建好了，但页面不显示，日志什么都没有。原因十有八九是没有调用subscribe()。所有操作符都懒，你不订阅它就不跑。调试时先检查：
flux.doOnNext(item -> System.out.println(item)) .subscribe();如果确认订阅了还不跑，再看是不是上游被filter全过滤了，或者take(0)。
6.2 背压过猛导致内存上涨
我们项目有一次用onBackpressureBuffer()处理高峰流量，队列无上限，结果上游每秒发 5 万条，下游处理速度只有 1 万条，几分钟后内存直接 OOM。后来改成：
.onBackpressureBuffer(5000)队列满了以后怎么办？默认会抛异常。我们改成了 wait 策略：
.onBackpressureBuffer(5000, BufferOverflowStrategy.DROP_OLDEST)每个背压策略都必须设计“溢出行为”，不要用默认的无限缓冲。
6.3 flatMap 里 block 导致线程卡死
响应式代码最忌讳阻塞调用。有人图省事，在flatMap里写：
User user = userRepository.findById(id).block();这在响应式线程里会非常危险：如果线程池只有 10 个线程，10 个请求全部阻塞，这个线程池就报废了。正确做法是把阻塞调用包到boundedElastic调度器里，但更好的方案是用响应式客户端替代阻塞客户端。
6.4 热流与冷流混用导致数据丢失
默认创建的 Reactor 流大都是冷流，即每次订阅都会从头开始执行，比如Flux.interval每订阅一次都会生成新的数据序列。而热流使用Sinks.many()创建多个订阅者共享同一数据源。如果你希望“广播给所有订阅者”，却错误用了冷流，每次订阅都会重新执行，数据可能重复计算或丢失。判断方法是加.log()，看订阅时是否会重新触发上游生产。
6.5 排查工具不能只看日志堆栈
响应式流的异常堆栈往往不是原始发送点，而是订阅点。所以排查要结合doOnNext、doOnError、log()几个触点，分阶段打点。我习惯在入口、第一个操作符、最终订阅处各打一次，定位慢在哪个区间。
最后说一点个人体会：Reactor 真正难的不是 API，而是思维方式。你不再能假设“上一行代码执行完，下一行就跟着执行”，而是要考虑数据在管道里如何流动、如何暂停、如何恢复。背压调优也不是一步到位的，我的做法是先加limitRate限制批量大小，再观察消费耗时和缓冲水位，最后再选择Buffer、Drop还是Latest。这套流程跑几轮，就能摸到系统的极限而不是凭感觉拍脑袋。
