onBackpressureError：没有需求就终止
onBackpressureError 是最严格的策略：它在 onSubscribe 里向上游请求 Long.MAX_VALUE，但当元素到达而下游的请求计数为 0 时，不缓冲、不丢弃，直接发出 OverflowException 终止整条流。
用同步源和手动请求可以精确复现：
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import java.util.ArrayList;
import java.util.List;
public class BackpressureErrorDemo {
public static void main(String[] args) {
List<Integer> received = new ArrayList<>();
Flux.range(1, 200)
.onBackpressureError()
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
// 故意不请求
}
@Override public void onNext(Integer v) { received.add(v); }
@Override
public void onError(Throwable t) {
System.out.println("received=" + received);
System.out.println("error=" + t.getClass().getName());
System.out.println("message=" + t.getMessage());
}
@Override public void onComplete() { }
});
}
}
实测输出：
received=[]
error=reactor.core.Exceptions$OverflowException
message=The receiver is overrun by more signals than expected (bounded queue...)
三个容易记错的点：
- 异常类型不是
IllegalStateException。实际类型是reactor.core.Exceptions$OverflowException，它继承IllegalStateException；判断时应该用Exceptions.isOverflow(t)，而不是catch (IllegalStateException e)。 - 错误发生在第一个元素到达时，不是订阅时。如果上游一个元素都不发，这条流不会报错。
- 它自己不保留任何元素。这与
onBackpressureBuffer(n, ..., BufferOverflowStrategy.ERROR)不同：后者会先把队列里已有的元素发完，再在排空后报错；onBackpressureError没有队列，直接终止。
一个常见误解是「delayElements 会无界消费，所以永远不会触发这个错误」。实际相反：delayElements 内部是 delayUntil，走不带预取的 concatMap，一次只请求 1 个元素。Flux.interval(10ms).onBackpressureError().delayElements(100ms) 在 550ms 内实测 consumed=0 errors=1——第一个延迟还没完成，下一个 interval 元素就到了，需求为 0，流立即失败。
onBackpressureError 适合「不能丢、也不能缓冲，宁可整条流失败」的场景：下游必须自己控制节奏，一旦失控就暴露出来。代价是失败后没有恢复点，需要外层 retryWhen 或告警介入。
limitRate：把上游请求切成批次
limitRate 经常被当成「限流操作符」，但它控制的是请求的批次大小，不是元素的处理速率。
它的实现很短：
// Flux.limitRate(int) 的内部实现
public final Flux<T> limitRate(int prefetchRate) {
return onAssembly(this.publishOn(Schedulers.immediate(), prefetchRate));
}
也就是说，它插入了一个预取队列，把下游的请求攒成批再向上游要。publishOn 的默认补请求水位是预取量的 75%：
// Operators.unboundedOrLimit(10, 10)
10 - (10 >> 2) == 8
实测请求序列
import reactor.core.publisher.Flux;
import java.util.ArrayList;
import java.util.List;
public class LimitRateDemo {
static List<Long> requests = new ArrayList<>();
public static void main(String[] args) {
Flux.range(1, 100)
.doOnRequest(requests::add)
.limitRate(10)
.subscribe(v -> { });
System.out.println("requests=" + requests);
}
}
实测输出：
requests=[10, 8, 8, 8, 8, 8, 8, 8, 8, 8, 8, 8, 8]
第一次是初始预取 10；之后每消费到 8 个（10 的 75%，向上取整为 8），就补请求 8。整个生命周期一共向上游请求了 10 + 12 × 8 = 106 个，但源只有 100 个元素——请求总量可以超过元素总量，limitRate 给的是上限，不是精确配额。
limitRate 不会替你产生下游需求
这是最容易踩的坑。下面这段代码只会收到 10 个元素，并且永远不会完成：
Flux.range(1, 100)
.limitRate(10)
.subscribe(new BaseSubscriber<Integer>() {
@Override
protected void hookOnSubscribe(Subscription s) {
request(10); // 只请求一次
}
@Override
protected void hookOnNext(Integer v) {
// 这里不再 request
}
});
实测 received=10 completed=false。limitRate 向上游补请求，只是让上游把元素送进它的预取队列；下游自己的请求额度用完，流就停在那里。subscribe() 默认请求 Long.MAX_VALUE，所以日常写法看不出问题；一旦换成 BaseSubscriber 手动请求，就必须继续 request。
反过来，如果下游逐个请求，limitRate 仍然按批向上游要：
Flux.range(1, 15)
.doOnRequest(requests::add)
.limitRate(10)
.subscribe(new BaseSubscriber<Integer>() {
@Override
protected void hookOnSubscribe(Subscription s) {
request(1);
}
@Override
protected void hookOnNext(Integer v) {
request(1);
}
});
实测 requests=[10, 8] received=15。下游一次只要 1 个，上游仍然先被要了 10 个、再补 8 个。多出来的元素放在 limitRate 的预取队列里，所以上游已生产但下游未消费的元素最多可达到预取量。这就是它和 onBackpressureBuffer 的区别：后者是显式的缓冲算子，前者是隐式的预取窗口。
limitRate(highTide, lowTide)
两参数版本把初始预取量和后续补请求量分开：
Flux.range(1, 30).doOnRequest(requests::add).limitRate(10, 2).subscribe();
// requests=[10, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2]
Flux.range(1, 30).doOnRequest(requests::add).limitRate(10, 0).subscribe();
// requests=[10, 10, 10, 10]
规则来自 Operators.unboundedOrLimit(prefetch, lowTide)：
lowTide <= 0：关闭提前补请求，只有预取队列排空时才请求highTide；lowTide >= highTide：退化为默认的 75% 策略；- 其他情况：每次消费
lowTide个就补lowTide个。
lowTide 越小，请求与生产之间的并发窗口越大，补请求调用也越频繁。官方 javadoc 给过一个例子：下游总共请求 14 个，limitRate(10, 2) 会连续执行 7 次 request(2)；默认 lowTide=8 时只需要一次 request(8)。
什么时候用 limitRate
典型场景是下游请求量很大，但数据源按小批次更高效：数据库分页、按批拉取的 API、需要控制单次网络往返的场景。它让上游每次最多看到 prefetchRate 个请求，而不是 Long.MAX_VALUE。
它不适合用来「让快生产者慢下来」：上游请求被切小了，但 limitRate 自己的预取队列仍会吸收差额，消费端真正处理不过来时，积压只是换了个地方。控制并发要用 flatMap(mapper, concurrency)，控制总时长要用 timeout，控制采样频率要用 sample。
limitRequest：限制整个生命周期的总请求
limitRequest(n) 限制的是累计向上游请求的数量，而不是每次请求的大小。它在 3.8.8 里仍然存在，但已经标记为废弃：
// Flux.limitRequest(long) 的实现
@Deprecated
public final Flux<T> limitRequest(long n) {
return take(n, true);
}
javadoc 建议新代码用 take(n, true) / take(n) 代替。它的实际行为是「取满 n 个后正常完成」：
Flux.range(1, 100)
.doOnRequest(total::addAndGet)
.limitRequest(5)
.subscribe(values::add, e -> { }, () -> completed.set(true));
实测：
values=[1, 2, 3, 4, 5] completed=true totalRequested=5
不是「忽略后续元素」，也不是「把后续元素丢弃后继续运行」——它在第 5 个元素之后取消上游并发出 onComplete。
边界情况：limitRequest(0) 根本不订阅上游，直接完成。实测 subscribed=false completed=true（用 doOnSubscribe 观察）。
和 take(n) 的区别
take(n) 保证下游最多看到 n 个，但它依赖取消信号向上游传播；如果中间有 publishOn 这类预取操作符，上游可能在取消到达之前已经多生产了一批。limitRequest(n) 则从请求端封顶，上游累计最多只会被请求 n 个。
实测对比（异步竞态，具体数值每次会变，但趋势稳定）：
Flux.range(1, 100)
.doOnNext(v -> produced.incrementAndGet())
.publishOn(Schedulers.parallel(), 64)
.take(5)
.blockLast();
// produced=27
Flux.range(1, 100)
.doOnNext(v -> produced.incrementAndGet())
.limitRequest(5)
.publishOn(Schedulers.parallel(), 64)
.take(5)
.blockLast();
// produced=5
当上游生产本身有成本或副作用（发网络请求、查数据库、计费）时，这个差别很重要：take(5) 可能已经触发了 27 次生产，limitRequest(5) 则严格只触发 5 次。
注意 limitRequest 是总量上限，不是速率限制。要限制「每秒多少个」，应该用 sample、delayElements、timeout 或在入口做令牌桶，而不是 limitRequest。
Producer-side 与 Consumer-side
| 方式 | 谁控制 | 手段 | 适用 |
|---|---|---|---|
| Consumer-side | Subscriber | request(n) | 标准 Reactive Streams 背压 |
| Consumer-side 批量化 | 操作符 | limitRate(n) | 减少 request 往返 |
| Consumer-side 总量封顶 | 操作符 | limitRequest(n) / take(n, true) | 限制上游总生产量 |
| Producer-side | Publisher | Flux.create + onRequest | 桥接回调式/拉取式数据源 |
生产端限流的正确写法是 Flux.create 配合 sink.onRequest：每次下游请求时，按请求量发射，而不是在循环里轮询 requestedFromDownstream()。
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
public class ProducerRateDemo {
public static void main(String[] args) {
AtomicInteger emitted = new AtomicInteger();
List<Integer> received = new ArrayList<>();
Flux<Integer> flux = Flux.create(sink -> {
sink.onRequest(n -> {
System.out.println("onRequest n=" + n);
for (long i = 0; i < n && !sink.isCancelled()
&& emitted.get() < 5; i++) {
sink.next(emitted.incrementAndGet());
}
if (emitted.get() == 5) {
sink.complete();
}
});
});
flux.subscribe(new Subscriber<Integer>() {
Subscription subscription;
@Override
public void onSubscribe(Subscription s) {
subscription = s;
s.request(2);
}
@Override
public void onNext(Integer v) {
received.add(v);
if (received.size() == 2) {
subscription.request(3);
}
}
@Override public void onError(Throwable t) { }
@Override public void onComplete() { System.out.println("onComplete"); }
});
System.out.println("received=" + received);
}
}
实测输出：
onRequest n=2
onRequest n=3
onComplete
received=[1, 2, 3, 4, 5]
三个实现要点：
n是本次新增的请求量，直接按它发射即可。requestedFromDownstream()返回的是内部计数快照，适合做诊断指标，不适合当作「还能发几个」的许可证；在重入请求的场景下，它和当前回调的n并不相等。Flux.push不适用这套写法。push是纯推送模型，注册onRequest时会立刻收到Long.MAX_VALUE，无法表达按需生产。需要背压就用Flux.create。- 不要在订阅线程里
Thread.sleep等需求。下面这种写法既阻塞了subscribe()，又在BUFFER默认策略下悄悄把超量元素排进队列：
// 反面示例
Flux.create(sink -> {
int count = 0;
while (count < 100 && !sink.isCancelled()) {
if (sink.requestedFromDownstream() > 0) {
sink.next(count++);
} else {
Thread.sleep(10); // 阻塞订阅线程
}
}
sink.complete();
});
如果要主动从上游拉数据，正确做法是在 onRequest 回调里向上游 request(n)，把需求传递给真正的数据源；而不是在本地循环里轮询计数器。
小结
| 手段 | 控制对象 | 关键行为 |
|---|---|---|
onBackpressureError() | 无需求时的单个元素 | 立即 OverflowException，不缓冲 |
limitRate(n) | 向上游请求的批次 | 初始 n，之后按 75% 补请求；不产生下游需求 |
limitRate(h, l) | 批次与补请求量 | l=0 只在队列排空时补；l>=h 退化为 75% |
limitRequest(n) | 整个生命周期的总请求量 | 取满 n 个后完成；n=0 不订阅上游 |
Flux.create + onRequest | 生产端发射节奏 | 按回调参数 n 发射；Flux.push 不支持 |
记住三层区别：onBackpressureError 决定「没有需求时这个元素怎么办」；limitRate 决定「一次向上游要多少」；limitRequest 决定「一辈子最多要多少」。三者可以组合，但都不能替代对下游并发和处理能力的真实评估。
下一节：Cold Publisher 与 Hot Publisher。
阅读导航
上一篇：7.2 背压策略：Drop Latest 与 Drop Oldest
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。