先确认前提：丢弃只发生在「下游没有需求」时
onBackpressureDrop、onBackpressureLatest、onBackpressureError 有一个共同点：它们都会向上游请求 Long.MAX_VALUE，把上游变成无界生产；区别只在于「下游此刻没有请求」时怎么处理当前元素。
// FluxOnBackpressureLatest 的订阅逻辑（简化）
actual.onSubscribe(this);
s.request(Long.MAX_VALUE);
因此，只要下游没有持续请求，钩子就会被调用。下面这个写法会触发大量丢弃：
Flux.interval(Duration.ofMillis(10))
.onBackpressureDrop(v -> dropped.incrementAndGet())
.delayElements(Duration.ofMillis(100))
.subscribe();
delayElements 内部走不带预取的 concatMap，一次只向上游请求 1 个元素，等前一个延迟完成才请求下一个。10ms 一个的 interval 在 100ms 的延迟窗口里连续到达，而需求一直是 0，大部分元素被丢弃。实测 550ms：
consumed=4 dropped=50
「延迟」不等于「无界消费」。要精确控制丢弃的数量和时机，还是要用同步源加手动请求。
确定性实验
实验仍然采用「同步源 + 手动请求」的方式：
Flux.range(1, 200)在subscribe()返回前同步发射完；- 订阅时可以选择先请求 1 个，或者完全不请求；
subscribe()返回后再请求 10 个，观察还能拿到什么。
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.AtomicReference;
public class DropLatestDemo {
public static void main(String[] args) {
runDrop(0);
runDrop(1);
runLatest(0);
runLatest(1);
runError(0);
}
static void runDrop(int initialRequest) {
List<Integer> received = new ArrayList<>();
List<Integer> dropped = new ArrayList<>();
AtomicReference<Subscription> subscription = new AtomicReference<>();
Flux.range(1, 200)
.onBackpressureDrop(dropped::add)
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
subscription.set(s);
if (initialRequest > 0) {
s.request(initialRequest);
}
}
@Override public void onNext(Integer v) { received.add(v); }
@Override public void onError(Throwable t) { }
@Override public void onComplete() { }
});
subscription.get().request(10);
System.out.println("drop initial=" + initialRequest
+ " received=" + received
+ " dropped=" + dropped.size()
+ " first=" + (dropped.isEmpty() ? null : dropped.get(0))
+ " last=" + (dropped.isEmpty() ? null
: dropped.get(dropped.size() - 1)));
}
static void runLatest(int initialRequest) {
List<Integer> received = new ArrayList<>();
AtomicReference<Subscription> subscription = new AtomicReference<>();
Flux.range(1, 200)
.onBackpressureLatest()
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
subscription.set(s);
if (initialRequest > 0) {
s.request(initialRequest);
}
}
@Override public void onNext(Integer v) { received.add(v); }
@Override public void onError(Throwable t) { }
@Override public void onComplete() { }
});
subscription.get().request(10);
System.out.println("latest initial=" + initialRequest
+ " received=" + received);
}
static void runError(int initialRequest) {
List<Integer> received = new ArrayList<>();
AtomicReference<Throwable> error = new AtomicReference<>();
Flux.range(1, 200)
.onBackpressureError()
.subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
if (initialRequest > 0) {
s.request(initialRequest);
}
}
@Override public void onNext(Integer v) { received.add(v); }
@Override public void onError(Throwable t) { error.set(t); }
@Override public void onComplete() { }
});
System.out.println("error initial=" + initialRequest
+ " received=" + received
+ " error=" + (error.get() == null ? null
: error.get().getClass().getName()));
}
}
实测输出：
drop initial=0 received=[] dropped=200 first=1 last=200
drop initial=1 received=[1] dropped=199 first=2 last=200
latest initial=0 received=[200]
latest initial=1 received=[1, 200]
error initial=0 received=[] error=reactor.core.Exceptions$OverflowException
对照着看：
| 场景 | Drop | Latest | Error |
|---|---|---|---|
| 订阅时完全不请求 | 200 个全部丢弃，后续请求拿不到任何元素 | 只保留 200，后续请求拿到 200 | 发射第 1 个元素时立即 OverflowException |
| 订阅时先请求 1 个 | 收到 1；2–200 全部丢弃 | 收到 1；2–199 被覆盖，之后收到 200 | 收到 1 后继续发射，无需求时再报错 |
两个关键结论：
- Drop 不保留任何元素。没有需求时元素当场丢弃，之后的
request不会把它们「捞回来」。 - Latest 只保留一个元素。新元素覆盖旧元素，下游恢复请求时只拿到最后一个。它没有任何丢弃回调，覆盖了多少个元素，操作符本身不会告诉你。
onBackpressureDrop：适合有独立价值、可降级的事件
日志采集是典型场景：每条日志都有价值，但下游写入能力有限，写不进去的至少要进死信队列。
Flux<LogEvent> logs = logCollector.flux();
logs.onBackpressureDrop(event -> {
dropCounter.increment();
deadLetterQueue.offer(event); // 必须是非阻塞的有界队列
})
.flatMap(event -> indexInElasticsearch(event)
.subscribeOn(Schedulers.boundedElastic()), 10)
.subscribe();
这里 flatMap 的第二个参数 10 会向上游施加有界需求：最多 10 个内层订阅同时进行，超出的部分才会让 onBackpressureDrop 丢弃。也就是说，丢弃是「下游并发已满」的结果，不是「上游太快」的自动保护。
两个实现细节：
- 没有参数的
onBackpressureDrop()是静默丢弃；需要计数或死信就必须用带Consumer的重载。 - 回调运行在上游发射线程上。写磁盘、发网络请求、加锁等待都会直接拖慢上游；放到有界队列里异步处理，队列满时也要有明确的降级策略。
- 回调抛异常会终止整条流：实现里会把异常交给
Operators.onOperatorError并触发onError。监控代码本身要足够简单，不要在回调里做可能失败的 I/O。
onBackpressureLatest：适合「只关心当前状态」
价格面板、传感器读数、进度百分比这类数据，中间值没有保留价值，下游恢复时只应该看到最新状态。
Flux<Price> prices = priceFeed.flux();
prices.onBackpressureLatest()
.subscribe(this::renderDashboard);
如果渲染本身很慢，renderDashboard 处理期间到达的价格会被不断覆盖，下一次请求只会拿到最后一个。消费端不需要为每个中间值付出渲染成本。
但要注意：如果目标是「每 5 秒刷新一次面板」，直接用 sample 就够了：
priceFeed.flux()
.sample(Duration.ofSeconds(5))
.subscribe(this::updateDashboard);
sample 会向上游请求无界需求，在每个时间窗口结束时发出窗口内的最后一个值；窗口内没有新值时不会发射。它已经完成了「只保留最新」的筛选，再叠加 onBackpressureLatest 是多余的，而且两层语义容易让人误以为丢弃发生在 onBackpressureLatest。
onBackpressureLatest 的监控也有盲区：它没有丢弃回调。可以分别统计上游发射数和下游接收数，用差值近似被覆盖的数量，但这两个计数不在同一个原子操作里，只适合做趋势监控，不适合精确对账。对精确性要求高的场景，用 onBackpressureDrop 的显式回调，或者改造成 sample。
onBackpressureError：没有需求就快速失败
onBackpressureError() 在没有下游需求时直接发出 OverflowException：
Flux.range(1, 200)
.onBackpressureError()
.subscribe(...);
实测异常类型是 reactor.core.Exceptions$OverflowException，消息为 The receiver is overrun by more signals than expected (bounded queue...)。它继承 IllegalStateException，可以用 Exceptions.isOverflow(t) 判断。
它适合「丢数据不可接受，但下游又必须显式控制节奏」的场景：与其静默丢或无限堆积，不如让流失败，由外层 retryWhen 或告警介入。代价是整条流终止，已经缓冲的元素不会保留——onBackpressureError 不是带错误出口的 buffer。
Drop vs Latest vs Buffer
| 场景 | 选择 | 原因 |
|---|---|---|
| 只关心最新状态，中间值可覆盖 | onBackpressureLatest | 只保留一个元素，内存恒定 |
| 每条数据都有价值，但允许降级到死信 | onBackpressureDrop + 回调 | 丢弃可观测，可补传 |
| 短期抖动，消费能追上 | onBackpressureBuffer | 保留全部，但必须有界 |
| 不能丢，也不能无限堆积 | onBackpressureError | 快速失败，交给外层处理 |
| 固定周期采样 | sample(Duration) | 窗口内只取最后一个，无需额外操作符 |
一个常见的误判是把 onBackpressureDrop 当成「自动限流」。它不控制上游速度，只是在需求不足时丢掉当前元素；真正的并发控制仍然在 flatMap(concurrency)、limitRate 或入口限流。
监控与告警
至少要能回答三个问题：
- 丢了多少：
onBackpressureDrop的回调计数。 - 为什么丢：下游并发是否打满、处理延迟是否上升、死信队列是否也在堆积。
- 丢了什么：死信队列或采样日志，保留足够的上下文用于补传和排查。
onBackpressureLatest 没有丢弃回调，只能通过「上游发射数 − 下游接收数」估算覆盖量；onBackpressureError 则应该把 OverflowException 接入告警，因为它意味着整条流已经终止。
阈值不要照搬固定百分比。同样是 1% 的丢弃率，支付流水和 UI 刷新完全是两种事故等级；阈值应该由业务对丢失的容忍度和补传能力决定。
小结
| 操作符 | 无需求时 | 保留元素 | 可观测性 |
|---|---|---|---|
onBackpressureDrop | 丢弃当前元素 | 不保留 | 有回调，可计数、可进死信 |
onBackpressureLatest | 覆盖旧值 | 只保留最新一个 | 无回调，只能估算 |
onBackpressureError | 发出 OverflowException | 不保留 | 通过 onError 告警 |
onBackpressureBuffer | 入队，满后按策略处理 | 最多 maxSize 个 | 溢出回调 + 队列指标 |
记住一点：这些操作符都不会让上游变慢。它们只是决定了「下游没有需求时，当前这个元素怎么办」。真正的流量控制要回到请求端。
下一节：Error 策略与速率控制。
阅读导航
上一篇：7.1 背压策略：Buffer
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。