背压（Backpressure）详解：响应式流的核心机制

背压（Backpressure）详解：响应式流的核心机制

在响应式编程和异步流处理系统中，背压（Backpressure）是一个核心概念。它描述的是下游消费者处理速度跟不上上游生产者数据发送速度时，下游向上游反馈压力，从而控制数据流速的机制。

一、背压的本质

1.1 什么是背压？

想象一个场景：水管工正在向水桶里注水（上游生产数据），而水桶底部有一个小孔在放水（下游消费数据）。如果进水速度大于出水速度，水桶最终会溢出，导致数据丢失或系统崩溃。

背压就是让上游知道“下游忙不过来了，请放慢一点”的信号机制。

技术定义： 背压是响应式流规范（Reactive Streams）中定义的一种机制，允许消费者向生产者发出信号，表明其当前能够处理的数据量，从而实现生产者与消费者之间的速率匹配。

1.2 为什么需要背压？

在传统同步编程中，方法调用是阻塞的——调用方等待被调用方执行完毕，速率天然匹配。但在异步/响应式系统中，生产者可能在消费者尚未准备好时持续推送数据，导致：

二、背压的实现机制

2.1 响应式流规范（Reactive Streams）

背压是 Reactive Streams 规范的核心部分。该规范定义了四个核心接口：

// 发布者：生产数据
public interface Publisher<T> {
void subscribe(Subscriber<? super T> subscriber);
}

// 订阅者：消费数据
public interface Subscriber<T> {
void onSubscribe(Subscription subscription);
void onNext(T item);
void onError(Throwable throwable);
void onComplete();
}

// 订阅：控制背压的关键
public interface Subscription {
void request(long n); // 请求 n 个数据
void cancel(); // 取消订阅
}

// 处理器：既是订阅者又是发布者
public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {}

关键机制： Subscription.request(long n) 是背压的核心。它让消费者告诉生产者“我还能处理 n 个数据”，生产者据此控制发送速度。

2.2 两种背压策略

策略一：推模式（Push）

生产者主动推送数据，消费者被动接收。如果消费者速度慢，数据会在缓冲区堆积。

// 模拟推模式：生产速度固定 100/秒，消费速度只有 10/秒
// 数据会迅速堆积，最终 OOM

策略二：拉模式（Pull）

消费者主动拉取数据，生产者按需提供。消费者每次拉取自己能处理的数据量。

// 模拟拉模式：消费者每次请求 10 条，处理完后再请求下一批
// 生产速度自动与消费速度匹配

响应式流的背压本质上是“推拉结合”：生产者可以推送数据，但必须遵守消费者的 request 信号——消费者请求多少，生产者就发送多少。

三、Reactor 中的背压实现

Reactor 是 Spring WebFlux 的底层实现，完整实现了 Reactive Streams 规范。

3.1 背压操作符

Flux.range(1, 1000000)
.onBackpressureBuffer() // 策略1：缓冲
// .onBackpressureDrop() // 策略2：丢弃
// .onBackpressureLatest() // 策略3：只保留最新
.subscribe(new BaseSubscriber<Integer>() {
@Override
protected void hookOnSubscribe(Subscription subscription) {
// 初始请求 10 个数据
subscription.request(10);
}

@Override
protected void hookOnNext(Integer value) {
// 处理数据…
// 处理完后继续请求 10 个
request(10);
}
});

3.2 背压策略详解

3.3 实践示例

// 1. 有界缓冲
Flux.interval(Duration.ofMillis(10)) // 每 10ms 生成一个数据
.onBackpressureBuffer(100) // 最多缓冲 100 个
.subscribe(new SlowConsumer()); // 消费速度很慢

// 2. 丢弃策略
Flux.interval(Duration.ofMillis(10))
.onBackpressureDrop(dropped -> {
System.out.println("丢弃数据：" + dropped);
})
.subscribe();

// 3. 只保留最新
Flux.interval(Duration.ofMillis(10))
.onBackpressureLatest()
.subscribe();

// 4. 自定义拉取速率
Flux.range(1, 1000)
.subscribe(new BaseSubscriber<Integer>() {
private int count = 0;
private final int BATCH_SIZE = 10;

@Override
protected void hookOnSubscribe(Subscription subscription) {
subscription.request(BATCH_SIZE);
}

@Override
protected void hookOnNext(Integer value) {
process(value);
count++;
if (count % BATCH_SIZE == 0) {
request(BATCH_SIZE);
}
}
});

四、背压与传统流控的对比

五、背压的局限与挑战

5.1 适用性限制

• 阻塞 I/O 场景：JDBC、阻塞 HTTP 客户端等无法有效支持背压
• 无背压的数据源：文件系统、网络套接字等无法响应背压信号
• 单线程消费者：即使有背压，单线程消费也无法超过 CPU 处理极限

5.2 误区与陷阱

// ❌ 误区：认为背压自动解决所有性能问题
// 背压只是让慢消费者不被压垮，不能让快消费者变慢

// ❌ 误区：使用无界缓冲区
onBackpressureBuffer() // 可能导致 OOM

// ✅ 正确：使用有界缓冲区
onBackpressureBuffer(1000)

// ✅ 更好：明确设置溢出策略
onBackpressureBuffer(1000, BufferOverflowStrategy.DROP_OLDEST)

六、Spring WebFlux 中的背压

Spring WebFlux 基于 Reactor 构建，天然支持背压：

@RestController
public class UserController {

@GetMapping("/users")
public Flux<User> getUsers() {
// 数据库返回的 Flux 会自动应用背压
return userRepository.findAll();
// WebFlux 会根据客户端消费速度自动控制数据发送
}

@GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent> streamEvents() {
return Flux.interval(Duration.ofSeconds(1))
.map(i -> ServerSentEvent.builder()
.data("Event " + i)
.build())
// 当客户端消费慢时，服务端会应用背压策略
.onBackpressureDrop(); // 丢弃无法发送的数据
}
}

七、总结

背压是响应式编程中处理速率不匹配问题的核心机制。它通过 Subscription.request(n) 实现下游对上游的流量控制，从根本上避免了数据堆积导致的系统崩溃。

核心要点：

• 背压是一种反馈机制：下游告诉上游“我能处理多少”
• Reactive Streams 是标准：定义了发布者、订阅者、订阅之间的契约
• Reactor 提供多种策略：缓冲、丢弃、最新、错误
• 不是万能的：阻塞 I/O 和无背压数据源需要额外处理
• 在分布式系统中尤为重要：微服务间的异步通信依赖背压防止级联故障

在构建高吞吐、低延迟的异步系统时，背压是一个必须理解的概念。它让系统在面对突发流量时，不是被压垮，而是优雅地降速，保持稳定运行。