java

13.3 背压演示面板：请求率 vs 消费率

用同一套指标对比 BUFFER、DROP、LATEST 三种策略：上游 10ms 一个元素，下游每次请求 1 个并处理 50ms。实测 BUFFER(10, DROP_LATEST) 丢弃 76 个、仍有 10 个留在缓冲区，DROP 丢弃 85 个，LATEST 通过 doOnDiscard 计数丢弃 84 个。重点说明 onBackpressureBuffer 两参数重载在溢出时会发 OverflowException，而不是静默丢弃。

Leeting Yan 4 分钟阅读 1916 字

指标定义

面板统计五个值，每个策略一份，不能跨策略共享：

指标计数位置含义
emitted源之后的 doOnNext上游实际产生了多少元素
delivered消费者 onNext下游真正收到多少元素
dropped各策略的丢弃回调或 doOnDiscard被策略丢弃多少元素
requests消费者侧的 doOnRequest下游累计请求了多少次
latency消费者 onNext 内计时处理一个元素的耗时

注意 emitted 不等于 delivered，两者之差也不一定等于 dropped：BUFFER 策略在取消时可能还有元素留在队列里，这部分既不叫「已交付」，也不叫「已丢弃」。

代码

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.BufferOverflowStrategy;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiFunction;

public class BackpressureDashboard {

    static final class Stats {
        final AtomicInteger emitted = new AtomicInteger();
        final AtomicInteger delivered = new AtomicInteger();
        final AtomicInteger dropped = new AtomicInteger();
        final AtomicInteger requests = new AtomicInteger();
        final AtomicLong latencyNanos = new AtomicLong();
    }

    public static void main(String[] args) throws Exception {
        run("BUFFER(10)", (flux, stats) -> flux.onBackpressureBuffer(
            10,
            v -> stats.dropped.incrementAndGet(),
            BufferOverflowStrategy.DROP_LATEST));

        run("DROP", (flux, stats) ->
            flux.onBackpressureDrop(v -> stats.dropped.incrementAndGet()));

        run("LATEST", (flux, stats) ->
            flux.onBackpressureLatest()
                .doOnDiscard(Long.class, v -> stats.dropped.incrementAndGet()));
    }

    static void run(String name, BiFunction<Flux<Long>, Stats, Flux<Long>> strategy)
            throws Exception {
        Stats stats = new Stats();
        CountDownLatch done = new CountDownLatch(1);

        Flux<Long> producer = Flux.interval(Duration.ofMillis(10))
            .doOnNext(v -> stats.emitted.incrementAndGet());

        Flux<Long> flux = strategy.apply(producer, stats)
            .publishOn(Schedulers.boundedElastic(), 1)
            .doOnRequest(n -> stats.requests.addAndGet((int) Math.min(n, Integer.MAX_VALUE)));

        flux.subscribe(new BaseSubscriber<Long>() {
            long start;

            @Override
            protected void hookOnSubscribe(Subscription s) {
                start = System.nanoTime();
                request(1);
            }

            @Override
            protected void hookOnNext(Long value) {
                stats.delivered.incrementAndGet();
                long t0 = System.nanoTime();
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    cancel();
                    done.countDown();
                    return;
                }
                stats.latencyNanos.addAndGet(System.nanoTime() - t0);

                if (System.nanoTime() - start < Duration.ofSeconds(1).toNanos()) {
                    request(1);
                } else {
                    cancel();
                    done.countDown();
                }
            }

            @Override
            protected void hookOnError(Throwable t) {
                t.printStackTrace();
                done.countDown();
            }
        });

        done.await(5, TimeUnit.SECONDS);

        System.out.println("=== " + name + " ===");
        System.out.println("Requests: " + stats.requests.get());
        System.out.println("Emitted: " + stats.emitted.get());
        System.out.println("Delivered: " + stats.delivered.get());
        System.out.println("Dropped: " + stats.dropped.get());
        System.out.println("Avg Latency: "
            + (stats.latencyNanos.get() / Math.max(stats.delivered.get(), 1) / 1_000_000) + "ms");
        System.out.println("Produced - Delivered: "
            + (stats.emitted.get() - stats.delivered.get()));
        System.out.println();
    }
}

publishOn(..., 1) 的预取量设为 1，是为了让背压信号快速传回上游；默认预取 256 会把下游的慢消费掩盖在队列里，1 秒的演示根本触发不了丢弃。

一次实测

本机 JDK 25、Reactor 3.8.8 的一次运行（数字随时序和调度抖动变化，不是基准）：

=== BUFFER(10) ===
Requests: 18
Emitted: 104
Delivered: 18
Dropped: 76
Avg Latency: 57ms
Produced - Delivered: 86

=== DROP ===
Requests: 16
Emitted: 101
Delivered: 16
Dropped: 85
Avg Latency: 57ms
Produced - Delivered: 85

=== LATEST ===
Requests: 18
Emitted: 102
Delivered: 18
Dropped: 84
Avg Latency: 56ms
Produced - Delivered: 84

逐项解释：

  • BUFFER(10)：Emitted 104，Delivered 18，差值是 86。其中 76 个被 DROP_LATEST 丢弃，另外 10 个在取消时仍留在缓冲区。Dropped 与 Produced - Delivered 不相等是正常的，差额就是队列里没来得及消费的元素。
  • DROP：Dropped 85，Produced - Delivered 85，两者相等。onBackpressureDrop 不保留队列，没有「滞留元素」。
  • LATEST：Dropped 84 来自 doOnDiscard。onBackpressureLatest 只保留一个最新值，被新值替换的旧值走 discard 钩子，所以能精确计数。
  • Requests 是消费者发出的请求次数，不是请求的元素总数。消费者每次 request(1)，因此它和 Delivered 很接近；多出来的一两次是取消前已经发出的请求。
  • Avg Latency 用 Delivered 做分母，不是 Emitted。拿「处理耗时」除以「上游产量」会得到明显偏小的假数字。

关键陷阱：两参数的 onBackpressureBuffer 不是「丢弃」

很多示例写成：

flux.onBackpressureBuffer(50, dropped::incrementAndGet)

看起来像「溢出时调用 dropped 并丢弃」，实际语义是：溢出时调用这个消费者，然后以 OverflowException 终止整条流。用容量 50 实测时，缓冲区刚满、Dropped 变成 1，流就结束了；后面的元素不会再产生。

要让流继续，必须显式指定溢出策略：

flux.onBackpressureBuffer(
    50,
    dropped::incrementAndGet,
    BufferOverflowStrategy.DROP_LATEST);

三个可选策略：

策略行为
DROP_LATEST丢弃新来的元素，保留队列里的旧元素
DROP_OLDEST丢弃队首，给新元素腾位置
ERROR默认行为，发 OverflowException

onBackpressureBuffer(50) 单参数重载同样使用 ERROR。它的「缓冲」只是延迟溢出，不是消除溢出。

为什么 Flux.interval 需要策略

Flux.interval 是定时源：时钟不会为下游变慢，到点就有一个 tick 要交付。它确实遵守 request(n)，但只在 tick 到来时检查额度；如果这一刻没有未满足的需求，它不会静默丢弃，而是直接以 OverflowException 终止整条流：

reactor.core.Exceptions$OverflowException: Could not emit tick 0 due to lack of requests
(interval doesn't support small downstream requests that replenish slower than the ticks)

本机用 .subscribe(..., s -> s.request(0)) 实测即可复现。也就是说，interval 的默认背压策略等价于 ERROR。要让它接慢消费者，必须在中间显式选择一种策略：

  1. onBackpressureBuffer：加有界队列，队列满后按 BufferOverflowStrategy 丢弃或报错；
  2. onBackpressureDrop / onBackpressureLatest：向上游请求无界，由策略在交付前丢弃。

onBackpressureBuffer、onBackpressureDrop、onBackpressureLatest 就是三种显式选择。不要用无界缓冲掩盖问题：无界队列只是把 OOM 推迟到更晚。

如果源本身能降速（例如 Flux.range、数据库游标），优先让背压一路传回源，而不是在中途加策略。

面板的读数方法

现象说明
Delivered 远小于 Emitted，Dropped 接近差值有损策略在起作用
Delivered 远小于 Emitted，Dropped 很小元素滞留在缓冲区，取消时被丢弃
Requests 接近 Delivered下游逐个请求，没有预取
Requests 远大于 Delivered下游预取较多，或请求了但上游没产出
Avg Latency 接近消费者的 sleep 时间处理耗时是主要成本，不是调度

这些读数只能说明「这一秒发生了什么」，不能直接外推成吞吐基准。要比较策略优劣，至少固定源速率、消费者耗时、运行时长和调度器，并重复多次取分位数。

小结

策略保留什么丢弃计数方式
onBackpressureBuffer(n, consumer, DROP_LATEST)最多 n 个旧元素溢出消费者
onBackpressureDrop(consumer)不保留丢弃消费者
onBackpressureLatest()最新 1 个doOnDiscard
onBackpressureBuffer(n)最多 n 个，溢出即错误无（流终止）

阅读导航

上一篇：13.2 传感器模拟器：带背压的随机数据流

下一篇：14.1 用 Project Reactor 重写 eventd

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南