java

7.1 背压策略：Buffer

讲清 onBackpressureBuffer 的真实作用：向上游请求 Long.MAX_VALUE、把下游的需求不足转成队列，而不是让上游变慢。用确定性的订阅实验实测 DROP_LATEST、DROP_OLDEST、ERROR 三种策略的丢弃数量、保留元素、钩子参数与终止信号，并说明 delayElements 一次只请求一个元素、buffer 放在它前面会真实堆积。示例基于 reactor-core 3.8.8。

Leeting Yan 5 分钟阅读 2431 字

Buffer 解决的是「下游需求不足」，不是「上游太快」

onBackpressureBuffer 的机制很容易被误解。它在 onSubscribe 里会向上游请求 Long.MAX_VALUE，然后把元素放进自己的队列，等下游 request(n) 时再按需发出：

// FluxOnBackpressureBuffer 的订阅逻辑（简化）
actual.onSubscribe(this);
s.request(Long.MAX_VALUE);   // 先向上游要满

这意味着两件事：

  1. 它不会让上游变慢。上游会以自己能跑的最快速度把元素全部推给它。
  2. 它只在下游确实没有请求时才会堆积。下游如果请求 Long.MAX_VALUE，队列会立刻被抽干。

但反过来，把 buffer 放在慢消费者前面，它就会真的堆积。delayElements(Duration) 内部是 delayUntil(d -> Mono.delay(...))，而 delayUntil 用的是不带预取的 concatMap：取一个元素，等它的延迟完成，再请求下一个。也就是说它同一时刻只向上游要 1 个元素，不是无界消费者。

用同步源可以把这一点确定下来：

Flux.range(1, 20)
    .onBackpressureBuffer(2, v -> dropped.incrementAndGet(),
        BufferOverflowStrategy.DROP_LATEST)
    .delayElements(Duration.ofMillis(100))
    .subscribe(v -> consumed.incrementAndGet());

range 在 subscribe() 返回前就把 20 个元素推给 buffer；delayElements 只取走 1 个（正在延迟中），buffer 容量是 2，其余 17 个被丢弃：

after subscribe: dropped=17 consumed=0
after 350ms: dropped=17 consumed=3

所以顺序决定了谁在积压：onBackpressureBuffer 放在慢消费者前面时，队列会持续增长；要观察溢出，这是最直接的写法。真正会主动从 buffer 里取走一批元素的是带预取的操作符（例如 publishOn 按预取窗口取数），而不是 delayElements。

重载与默认语义

重载缓冲上限溢出行为队列实现
onBackpressureBuffer()无界不丢弃，可能 OOMSpscLinkedArrayQueue，分段扩容
onBackpressureBuffer(int maxSize)maxSize延迟报 OverflowException（等价于 ERROR）Queues.get 返回的 SPSC 队列
onBackpressureBuffer(int, Consumer)maxSize同上，并回调被丢弃的元素同上
onBackpressureBuffer(int, BufferOverflowStrategy)maxSize按策略丢弃或报错ArrayDeque 子类，onNext 用 synchronized
onBackpressureBuffer(int, Consumer, BufferOverflowStrategy)maxSize按策略处理并回调同上
onBackpressureBuffer(Duration, int, Consumer[, Scheduler])maxSize过期元素移出；缓冲满时丢最老带 TTL 的队列

三个容易踩的点：

  • 无界版本只是初始链路长度有限，队列本身可以一直增长，不能用「数据量应该不大」来兜底。
  • 有界但不带策略的版本不是静默丢弃，而是报错；错误会等缓冲区排空后才送到下游。
  • 带策略的版本内部是 ArrayDeque，不是无锁队列；onNext 里有 synchronized 块。不要把它当成高并发场景的零开销结构。

三个策略的确定性实验

为了排除线程调度和「边生产边消费」的干扰，实验这样设计：

  1. 订阅时不请求任何元素；
  2. Flux.range(1, 200) 是同步源，会在 subscribe() 返回前把 200 个元素全部推给 buffer；
  3. subscribe() 返回后，再逐个 request(1) 消费缓冲区。

这样溢出一定发生在消费之前，丢弃数量和保留元素都是确定的。

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BufferOverflowStrategy;
import reactor.core.publisher.Flux;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

public class BufferStrategyDemo {

    public static void main(String[] args) throws Exception {
        for (BufferOverflowStrategy strategy : BufferOverflowStrategy.values()) {
            run(strategy);
            System.out.println();
        }
    }

    static void run(BufferOverflowStrategy strategy) throws Exception {
        List<Integer> received = new ArrayList<>();
        AtomicInteger dropped = new AtomicInteger();
        AtomicReference<Throwable> error = new AtomicReference<>();
        CountDownLatch done = new CountDownLatch(1);
        AtomicReference<Subscription> subscription = new AtomicReference<>();

        Flux.range(1, 200)
            .onBackpressureBuffer(10,
                v -> dropped.incrementAndGet(), strategy)
            .subscribe(new Subscriber<Integer>() {
                @Override
                public void onSubscribe(Subscription s) {
                    subscription.set(s);   // 先不请求
                }

                @Override
                public void onNext(Integer v) {
                    received.add(v);
                }

                @Override
                public void onError(Throwable t) {
                    error.set(t);
                    done.countDown();
                }

                @Override
                public void onComplete() {
                    done.countDown();
                }
            });

        System.out.println(strategy + " after subscribe: dropped="
            + dropped.get() + " received=" + received.size()
            + " error=" + error.get());

        for (int i = 0; i < 10; i++) {
            subscription.get().request(1);
            Thread.sleep(2);
        }
        subscription.get().request(100);   // 触发完成/错误信号

        done.await(2, TimeUnit.SECONDS);
        System.out.println(strategy + " final: received=" + received
            + " dropped=" + dropped.get()
            + " error=" + error.get());
    }
}

实测输出：

ERROR after subscribe: dropped=1 received=0 error=null
ERROR final: received=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10] dropped=1 error=reactor.core.Exceptions$OverflowException: The receiver is overrun by more signals than expected (bounded queue...)
DROP_LATEST after subscribe: dropped=190 received=0 error=null
DROP_LATEST final: received=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10] dropped=190 error=null
DROP_OLDEST after subscribe: dropped=190 received=0 error=null
DROP_OLDEST final: received=[191, 192, 193, 194, 195, 196, 197, 198, 199, 200] dropped=190 error=null

整理成表：

策略第 11 个元素到来时最终保留丢弃数终止信号
DROP_LATEST丢弃新来的 111–10190（11–200）onComplete
DROP_OLDEST移除最老的 1，放入 11191–200190（1–190）onComplete
ERROR不放入，触发溢出错误1–101（只回调 11）排空后 onError

三个细节值得单独说：

  • 钩子收到的元素不同。DROP_LATEST 的回调收到新来的元素（11–200）；DROP_OLDEST 收到被挤出的老元素（1–190）；ERROR 只回调一次，收到触发溢出的 11。
  • ERROR 不是立即终止。缓冲里已有的 1–10 仍会被下游取走，队列排空后才收到 OverflowException。异常类型是 reactor.core.Exceptions$OverflowException，它是 IllegalStateException 的子类，消息为 The receiver is overrun by more signals than expected (bounded queue...)。
  • onBackpressureBuffer(int) 与 ERROR 的行为一致，只是没有策略参数和可选的溢出回调；不要以为有界版本会自动丢弃。

队列实现与开销

版本队列并发访问
无界SpscLinkedArrayQueue单生产者单消费者，分段链表扩容
onBackpressureBuffer(int)Queues.get 返回的 SpscArrayQueue单生产者单消费者，无锁
带 BufferOverflowStrategyArrayDeque 子类onNext 内 synchronized，溢出判断与插入是原子的

DROP_OLDEST 必须同时做「移除最老」和「放入最新」，这两步不能分开，否则并发下会丢错元素，所以实现里用了锁。它的吞吐上限和 SpscArrayQueue 不在一个量级，不要因为同为「有界 buffer」就假设性能一样。

什么时候用 Buffer

Buffer 是最后一道缓冲，不是限流手段。优先考虑这些做法：

需求更合适的手段
控制并发flatMap(mapper, concurrency)
控制请求节奏limitRate(n)、自定义 request(n)
只要最新值onBackpressureLatest()
允许丢一部分onBackpressureDrop() 或有界 buffer + 策略
处理慢消费者扩容消费者、分流、降级，而不是加大队列

只有在下面这些条件同时成立时，才考虑 onBackpressureBuffer：

  • 下游确实会以有限速率请求；
  • 能给出明确的内存上限；
  • 溢出时有可接受的业务语义（丢新、丢老或报错）；
  • 有丢弃计数、队列长度等监控指标。

用无界 buffer 省掉的复杂度，最终会以 OOM 或长尾延迟的形式还回来。

陷阱：Mono 没有 onBackpressureBuffer

// 编译错误：Mono 没有这个方法
Mono.just(1).onBackpressureBuffer();

onBackpressureBuffer 定义在 Flux 上，不在 Mono 上。Mono 最多只有一个元素，不需要用队列吸收需求差异；处理慢或失败应该用 Mono.delay、timeout、onErrorResume 等操作符。

小结

策略行为数据语义风险
无界无限缓冲全保留OOM、长尾延迟
DROP_LATEST丢弃新元素保留最早的丢失新数据
DROP_OLDEST丢弃最老元素保留最新的丢失历史数据
ERROR排空后报错终止前不丢流不可用

三条结论：

  1. onBackpressureBuffer 向上游请求无界，它不抑制生产速度；下游不请求时才可能堆积。
  2. delayElements 一次只请求 1 个元素，是慢消费者；buffer 放在它前面会持续积压。publishOn 这类预取型操作符只在预取窗口内取走元素，窗口满后 buffer 同样会堆积。
  3. 生产环境默认使用有界 buffer、明确的溢出策略和丢弃监控；无界 buffer 只能用于能证明上界的场景。

下一节：Drop Latest 与 Drop Oldest 的深层对比。

阅读导航

上一篇：6.3 自定义调度器与 Worker 模型

下一篇：7.2 背压策略：Drop Latest 与 Drop Oldest

继续阅读

探索更多技术文章

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

全部文章 返回首页

「java」更多文章

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