如何使用java.util.concurrent.Flow实现响应式编程中的背压控制

flow.publisher 的 onsubscribe 方法必须立即调用 subscription.request()，否则背压链路卡死；request(n) 中 n 必须为正数，0 会触发取消协议，负数抛异常；processor 需同步上下游许可，submissionpublisher 的 maxbuffercapacity 控制缓冲阻塞而非背压。

Flow.Publisher 的 onSubscribe 方法必须立即调用 Subscription.request()

背压失效的最常见原因，是 Publisher 在 onSubscribe 回调里没立刻调用 subscription.request(1) 或其他正数。JDK 的 Flow 规范要求：订阅建立后，除非明确等待下游拉取，否则必须主动“推”第一个请求，否则整个链路会卡死。

实操建议：

立即学习“Java免费学习笔记（深入）”；

SophNet

专业的AI开发工具平台，让AI集成变得简单高效。

下载

• Subscription 实例不可缓存或延迟使用，收到即调用 request()
• 若需支持“暂停/恢复”，应把 request(n) 封装进线程安全的队列或原子计数器，而非靠 onSubscribe 一次性发完
• 测试时故意不调用 request()，观察下游 Subscriber 的 onNext 是否永远不触发——这是背压卡死的典型信号

Subscriber.onSubscribe() 中 request(n) 的 n 值不能为 0 或负数

传入 0 不代表“不请求”，而是触发规范定义的“取消协议”：Subscription.cancel() 会被隐式调用，后续所有 onNext 都被忽略。负数则直接抛 IllegalArgumentException。

实操建议：

立即学习“Java免费学习笔记（深入）”；

• 动态计算 n 时，务必做非零校验：if (n > 0) subscription.request(n);
• 缓冲区满时，别用 request(0) “暂停”，而应暂存未处理的 request 数量，等消费后再补发
• 注意 request(Long.MAX_VALUE) 等价于“无界请求”，等同于取消背压控制，仅适用于已知数据量小、且下游处理极快的场景

Processor 实现必须同时遵守 Publisher 和 Subscriber 协议

像 SubmissionPublisher 这类内置实现能自动协调背压，但自定义 Processor（如转换流）时，容易只顾一头：要么在 onNext 里盲目转发，忘了检查下游 Subscription 是否允许接收；要么在 submit() 时无视上游背压，导致内部队列无限膨胀。

实操建议：

立即学习“Java免费学习笔记（深入）”；

• 维护一个线程安全的待处理项计数器，在 onNext 入队前判断是否
• 每次向下游 onNext 后，立即 decrementAndGet() 许可数，并在许可耗尽时停止转发
• 不要复用同一个 Subscription 实例跨线程传递；每个 Subscriber 应绑定独立的 Subscription 状态

SubmissionPublisher 的 maxBufferCapacity 参数影响背压触发时机

SubmissionPublisher 构造时传入的 maxBufferCapacity 不是“最大缓存条数”，而是“当缓冲区中待处理元素 ≥ 该值时，submit() 开始阻塞或抛 SubmissionPublisher.SubmissionException”。它和背压无关，但和响应行为强相关。

实操建议：

立即学习“Java免费学习笔记（深入）”；

• 设为 0 时，submit() 永远非阻塞，但一旦下游处理慢，就会快速触发 onError（因内部缓冲溢出）
• 设为 1 是最保守选择，能确保每次只持有一个待处理项，但可能频繁触发 request() 来回，增加调度开销
• 若下游是异步处理（如写磁盘），建议配合 ForkJoinPool.commonPool() 使用，并将 maxBufferCapacity 设为略大于平均单次处理耗时 × 吞吐预期