Reactor 响应式编程

一句话：Reactor 用 Publisher/Subscriber/Subscription 把「生产多少、消费多少」变成显式的 request(n) 协议，用 Mono/Flux 描述一条惰性管道；它解决的是背压和线程切换，不是「异步就一定更快」。

适用 Reactor 3.6.x / Spring Boot 3.x / JDK 17+。HTTP 传输见 14-Java流式输出与SSE，父子流编排见 15-SSE父子流与流编排；本文只讲背压、线程切换和错误处理，不重复 SSE 帧与多路合并。

为什么需要

回调式异步能避免阻塞，但有两个问题：一是嵌套回调难读，二是生产者不知道消费者能处理多少。数据库返回一百万行、下游只能每秒处理一千行时，回调会继续推送，内存被队列撑爆。

Reactive Streams 用 request(n) 把消费能力反向传给生产者。Reactor 在这个协议上提供操作符组合、调度器和错误恢复，让「懒执行 + 背压 + 线程切换」成为一条可组合的管道。

核心机制

组装与订阅

Flux 是 0..N 个元素，Mono 是 0..1 个元素。它们都是惰性的：Flux.range(1, 5).map(...) 只是组装一条操作符链，只有 subscribe() 才触发订阅。

一次订阅的生命周期：

subscribe()
  → onSubscribe(Subscription)
  → request(n)                 // 消费者声明需求
  → onNext(item) ...           // 最多 n 个
  → onComplete() 或 onError()

没有 subscribe() 的管道不会执行；忘记订阅是 Reactor 最常见的「代码写了但没跑」。

背压：request(n) 是协议，不是建议

消费者通过 Subscription.request(n) 告诉生产者「我最多还能处理 n 个」。生产者必须尊重这个上限；如果生产者不尊重，Reactor 只能在中间加缓冲或丢弃策略。

常见边界：

• publishOn 默认预取 256 个元素（Queues.SMALL_BUFFER_SIZE），内部队列是有界的，不是无界。下游消费慢时，它只会请求到预取上限，然后停止向上游要数据。
• subscribeOn 影响订阅和请求发生的线程，不负责下游 onNext 的执行线程；需要切换下游执行线程用 publishOn。
• flatMap 默认并发度 256、预取 256；并发太高会把下游打满，太低会浪费吞吐。
• limitRate(n) 可以把请求拆成更小的批次，减少上游一次吐太多。
• onBackpressureBuffer/Drop/Latest/Error 是「上游不守规矩」时的兜底，不是背压本身。

如果上游是 Flux.create，必须在 emitter.requested 的约束下发送；不检查需求直接 next()，再多的下游背压也救不了。

线程切换

操作符	影响
subscribeOn(scheduler)	订阅、请求、取消发生的线程；通常放在源头附近
publishOn(scheduler)	下游 onNext/onComplete/onError 的执行线程；可多次出现
Schedulers.parallel()	CPU 密集型，线程数默认等于核数；不能阻塞
Schedulers.boundedElastic()	阻塞 IO 的兜底，线程有上限；适合包装 JDBC、老 HTTP 客户端

publishOn 只切换它下游的线程，不会把上游变成异步；subscribeOn 也不会让每个 onNext 都跑到另一个线程。把这两个搞混，会出现「以为切了线程，实际还在 IO 线程上阻塞」的问题。

错误与超时

错误是流的一等信号，onError 之后流终止。恢复操作符：

• onErrorResume(e -> Mono.just(fallback))：换一条流继续。
• onErrorReturn(value)：直接给默认值。
• retryWhen(Retry.backoff(3, Duration.ofMillis(100)))：有次数和退避的重试。
• timeout(Duration.ofSeconds(2))：超时转 TimeoutException，再配合 onErrorResume。

错误不会自动变成 HTTP 500；在 WebFlux 中，响应头一旦发出，后续错误只能通过流内事件或连接关闭表达。

最小上手

Maven 依赖：

<dependency>
  <groupId>io.projectreactor</groupId>
  <artifactId>reactor-core</artifactId>
  <version>3.6.5</version>
</dependency>

import org.reactivestreams.Subscription;
import reactor.core.publisher.*;
import reactor.core.scheduler.Schedulers;

public class ReactorDemo {
    public static void main(String[] args) throws Exception {
        Flux.range(1, 5)
            .publishOn(Schedulers.boundedElastic(), 2) // 有界预取 2
            .subscribe(new BaseSubscriber<Integer>() {
                @Override
                protected void hookOnSubscribe(Subscription s) {
                    System.out.println("subscribe on " + Thread.currentThread().getName());
                    request(1); // 一次只要一个
                }
                @Override
                protected void hookOnNext(Integer value) {
                    System.out.println("onNext " + value + " on " + Thread.currentThread().getName());
                    request(1);
                }
            });
        Thread.sleep(500); // 等待异步流结束；生产代码不要用 sleep
    }
}

mvn -q dependency:copy-dependencies -DoutputDirectory=lib
javac -cp "lib/*" ReactorDemo.java
java -cp ".;lib/*" ReactorDemo     # Linux/macOS 用 ":lib/*"

输出会显示 onNext 在 boundedElastic 线程上执行，并且每次只请求一个元素。把 request(1) 改成 request(5)，预取和消费节奏会变化，但队列仍然有界。

生产动作

• 阻塞操作放 boundedElastic：JDBC、文件 IO、老式 HTTP 客户端用 subscribeOn(Schedulers.boundedElastic()) 包装；不要在 parallel() 上阻塞。
• 超时和重试要成对：timeout 之后接 onErrorResume 或 retryWhen，否则错误直接终止请求；重试要限次数、加退避，避免放大下游故障。
• 控制 flatMap 并发：显式传 concurrency 和 prefetch，按下游容量设置；不要默认 256 打满数据库连接池。
• 背压策略要显式：上游可能无视需求时，选 onBackpressureBuffer(maxSize)、Drop、Latest 或 Error，并记录丢弃量。
• 调试：开发环境用 Hooks.onOperatorDebug() 或 ReactorDebugAgent，生产不要长期开启；日志里带 traceId 需要显式通过 Context 传播。
• 监控：用 Micrometer 看订阅数、取消数、错误率、排队长度；响应式链路的瓶颈常在「某个操作符偷偷阻塞」。

边界

• Reactor 不是「更快的异步」：它减少线程等待和回调嵌套，不减少下游工作量；CPU 密集任务用并行调度器也不会超过核数。
• 背压只在链路都遵守时有效：上游是消息队列、HTTP 客户端或 Flux.create 时不尊重 request(n)，就要靠缓冲/丢弃策略。
• 缓冲不是免费的：publishOn、flatMap、onBackpressureBuffer 都有队列，队列越大，内存和延迟越高。
• 调试成本更高：响应式栈是异步拼接的，异常栈不直观；需要 Reactor 的调试工具和清晰的边界日志。
• 虚拟线程是另一条路：JDK 21 可以用阻塞写法获得类似吞吐，团队不熟悉响应式时，先评估虚拟线程是否更合适。

讲清楚

Reactor 的核心是 Reactive Streams 的 request(n)：消费者声明需求，生产者按需推送，Flux/Mono 只是把这条协议包装成可组合的惰性管道。publishOn 切换下游执行线程，默认预取 256，队列有界；subscribeOn 影响订阅和请求的线程，不负责每个 onNext。

背压只有在整条链路都尊重需求时才成立，遇到不守规矩的上游要用缓冲或丢弃策略，并显式设置上限。错误和超时是流内信号，用 onErrorResume、retryWhen、timeout 处理。它适合 IO 密集、流式处理和需要背压的场景；CPU 密集、团队不熟悉异步调试时，虚拟线程可能更简单。