背压:request(n)、limitRate 与缓冲区溢出

不支持背压的源

实测 · 本机执行输出

消费者实际处理了 4 个元素就收到错误:OverflowException: Could not emit tick 4 due to lack of requests (interval doesn't support small downstream requests that replenish slower than the ticks)

Flux.interval 下游跟不上就抛 OverflowException。

溢出策略

实测 · 本机执行输出

onBackpressureBuffer()(无界)               消费者收到 5000 个,最后一个是 5000,回调里统计的丢弃数    0,错误:无
onBackpressureBuffer(100)(满了报错)          消费者收到  101 个,最后一个是  101,回调里统计的丢弃数    0,错误:OverflowException
onBackpressureBuffer(100, DROP_OLDEST)   消费者收到  116 个,最后一个是 5000,回调里统计的丢弃数 4884,错误:无
onBackpressureDrop()                     消费者收到   16 个,最后一个是   16,回调里统计的丢弃数 4984,错误:无
onBackpressureLatest()                   消费者收到   17 个,最后一个是 5000,回调里统计的丢弃数    0,错误:无

缓冲、丢弃、保留最新各有取舍,无界缓冲不丢但吃内存。

需求的传播

实测 · 本机执行输出

直接 subscribe(v -> ...)             上游收到的 request(n) 序列(共 1 次):[无限]
publishOn(单线程)(默认预取 256)           上游收到的 request(n) 序列(共 6 次):[256, 192, 192, 192, 192, 192]
publishOn(单线程, 32)                 上游收到的 request(n) 序列(共 42 次):[32, 24, 24, 24, 24, 24, 24, 24, 24, 24, 24, 24] …
limitRate(100)                     上游收到的 request(n) 序列(共 14 次):[100, 75, 75, 75, 75, 75, 75, 75, 75, 75, 75, 75] …
limitRate(100, 10)                 上游收到的 request(n) 序列(共 101 次):[100, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10] …
flatMap(x -> Mono.just(x))         上游收到的 request(n) 序列(共 6 次):[256, 192, 192, 192, 192, 192]
flatMap(..., 4)(并发 4)              上游收到的 request(n) 序列(共 334 次):[4, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3] …
concatMap(x -> Mono.just(x))       上游收到的 request(n) 序列(共 1001 次):[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1] …
buffer(10)                         上游收到的 request(n) 序列(共 1 次):[无限]
BaseSubscriber 手动控制:每次 request(3),处理完 3 个再 request 3 个:
  上游收到 request(3) | 收到 1 | 收到 2 | 收到 3 | 上游收到 request(3) | 收到 4 | 收到 5 | 收到 6 | 上游收到 request(3) | 收到 7 | 收到 8 | 收到 9 | 上游收到 request(3) | 收到 10

每个操作符有自己的预取和补充节奏。

无界缓冲

实测 · 本机执行输出

   200 ms:已生产 200000,已消费  148,堆积在缓冲里 199852
   400 ms:已生产 200000,已消费  289,堆积在缓冲里 199711
   600 ms:已生产 200000,已消费  427,堆积在缓冲里 199573
   800 ms:已生产 200000,已消费  570,堆积在缓冲里 199430
  对照:不加 onBackpressureBuffer,publishOn(预取 16)
   200 ms:已生产    148,已消费  142,生产领先消费      6
   400 ms:已生产    292,已消费  283,生产领先消费      9
   600 ms:已生产    424,已消费  420,生产领先消费      4

生产者全速跑,元素堆积在内存。

Sinks

实测 · 本机执行输出

  30 次 tryEmitNext 的返回值统计:{FAIL_OVERFLOW=21, OK=9}
  1.5 秒后订阅者实际收到 9 个

tryEmitNext 缓冲满了返回 FAIL_OVERFLOW,不检查就丢。

分步演示

背压

本步的对象

本文用到的工具与边界

本机 macOS 27 arm64 / OpenJDK 21.0.12(Homebrew) / Spring Boot 4.0.8(Spring Framework 7.0.9、Tomcat 11.0.24) / Spring Cloud 2025.1.3;用 javac 编译、java 直接运行,依赖由 Maven 从 Maven Central 解析;数字每次运行会有小幅抖动。用 reactor-core 3.8.7 的单个 Java 程序,不依赖 Spring 容器;消费者慢用 Thread.sleep 模拟(每个 1 ms 或 20 ms、50 ms),异步部分用 publishOn 切到单独线程;堆积数量按已生产减已消费估算,不是直接测的内存字节数
没有验证:真实内存占用(堆积的元素对象大小);WebFlux 里响应体写出时的背压(Netty 的可写状态);onBackpressureBuffer 与 Schedulers 不同线程组合的更多情形;Flux.create 的 push 与 pull 区别;windows、groupBy 的预取;Hooks.onNextDropped 的使用

下一篇 · SPRING 57

publishOn 与 subscribeOn:线程在哪里切换