java

1.3 HTTP 层的背压

从 TCP 滑动窗口与 Netty 写水位线出发，深入理解 Reactor Netty HTTP 层的背压机制：Flux 到 send/sendString 的异步衔接、WRITE_BUFFER_WATER_MARK 配置、isWritable 检查与慢客户端保护、超时断开与内存安全策略。

Leeting Yan 2026-11-01 5 分钟阅读 2069 字

TCP 层的背压

HTTP 运行在 TCP 之上，TCP 本身就有流控机制——滑动窗口。接收方通过 ACK 报文告知可用缓冲区大小（rwnd），发送方据此控制发送速率。如果接收方处理慢，rwnd 变小，发送方自动降速。这是操作系统内核实现的，应用层无需干预。

发送方 App → 发送缓冲区 → TCP 协议栈 → 网卡 → 网络
                                           ↑
接收方 App ← 接收缓冲区 ← TCP 协议栈 ← 网卡
         ↑
         ACK 报文携带 rwnd

但这个窗口只存在于「两个 TCP 端点」之间。如果服务端应用层（Reactor Netty）生产数据的速度超过了下游消费速度（客户端网络慢、客户端处理慢），仅靠 TCP 窗口还不够。

Netty 的写水位线

Netty 在应用层增加了第二道保护——写水位线（Write Buffer Water Mark）：

阈值	含义
低水位线（默认 32 KiB）	缓冲区低于此值时，`Channel.isWritable()` 恢复为 `true`
高水位线（默认 64 KiB）	缓冲区超过此值时，`Channel.isWritable()` 变为 `false`

isWritable() 只是一个状态查询，Netty 不会因为它自动暂停你的业务代码。真正的联动发生在 Reactor Netty 的发送路径上：写缓冲达到高水位后，发送端会感知到不可写并推迟向响应 Publisher 请求下一批数据，生产速度随之降下来，给 TCP 窗口时间消化积压。如果业务自己检查 isWritable()，也可以据此主动降速或丢弃非关键数据。

在 Reactor Netty 中，这个机制通过 ChannelOption.WRITE_BUFFER_WATER_MARK 配置：

import io.netty.channel.ChannelOption;
import io.netty.channel.WriteBufferWaterMark;
import reactor.netty.http.server.HttpServer;

HttpServer.create()
    .port(8080)
    .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
        new WriteBufferWaterMark(32 * 1024, 128 * 1024))  // 低 32K 高 128K
    .handle((req, res) -> ...)
    .bindNow();

childOption（不是 option）作用于每个连接的 Channel，option 只作用于 ServerSocket 的 accept 通道。

send / sendString 的背压衔接

Reactor Netty 1.x 的响应发送核心 API 是 HttpServerResponse.send(Publisher<? extends ByteBuf>) 及其包装方法 sendString()；旧的 writeWith() 系列已经移除。sendString() 返回 NettyOutbound（它实现了 Publisher<Void>），订阅后才表示发送完成，可在需要 Mono<Void> 的位置调用 .then() 收口；无参的 send() 则直接返回 Mono<Void>。

关键是：Publisher 的消费速度与 Netty Channel 的可写状态是联动的。

import reactor.core.publisher.Flux;
import reactor.netty.http.server.HttpServer;
import java.time.Duration;
import java.nio.charset.StandardCharsets;

public class BackpressureServer {
    public static void main(String[] args) {
        HttpServer.create().port(8080)
            .handle((req, res) -> {
                // 每 10ms 产生一个字符串，总共 10000 个
                Flux<String> data = Flux.range(1, 10000)
                    .delayElements(Duration.ofMillis(10))
                    .map(i -> "event-" + i + "\n");

                return res.sendString(data, StandardCharsets.UTF_8);
            })
            .bindNow();
        System.out.println("Slow-client test: curl http://localhost:8080");
        try { System.in.read(); } catch (Exception ignored) {}
    }
}

运行后用 curl --limit-rate 10K http://localhost:8080 模拟慢客户端。你会观察到：

delayElements 每 10ms 发出一个元素，但它同样受下游 request(n) 约束
sendString 把下游的消费节奏传回 Flux：客户端读得慢，request(n) 就来得慢
客户端极慢时，Netty 写缓冲区达到高水位线，isWritable() 变 false
Reactor Netty 推迟向 Flux 请求更多数据，delayElements 没有下游需求，也就不会继续发出新元素

这就是从响应式流到 TCP 的背压贯通：

Flux.range(1, 10000)
    ↑ 背压：slow consumer → 减少 request(n)
    delayElements(10ms)
    ↑ 背压：netty channel not writable → 暂停调度
    res.sendString()
    ↑ Netty ByteBuf 缓冲区 → isWritable() → backpressure
    TCP send buffer
    ↑ TCP sliding window (rwnd)
    客户端

手动检查 isWritable

如果你需要更细粒度的控制（例如丢弃部分事件而不是全部暂停），可以检查 isWritable()：

import reactor.netty.channel.ChannelOperations;

HttpServer.create().port(8080)
    .handle((req, res) -> {
        Flux<String> events = Flux.interval(Duration.ofMillis(1))
            .map(i -> "tick-" + i)
            .filter(tick -> {
                // 获取底层 Netty Channel
                if (res instanceof ChannelOperations<?,?> ops) {
                    return ops.channel().isWritable();
                }
                return true;
            });

        return res.sendString(events);
    })
    .bindNow();

这个方案在高频实时场景（如行情推送）中很有用：当客户端网络拥塞时，丢弃中间 ticks 而不是全部阻塞，保证最新数据优先送达。

慢客户端保护：超时与断开

背压机制再完善，也无法应对恶意或故障客户端完全不读取数据的情况。必须配置超时：

import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;
import reactor.netty.http.server.HttpServer;

HttpServer.create()
    .port(8080)
    .doOnConnection(conn ->
        conn.addHandlerLast(new ReadTimeoutHandler(30))   // 30秒无数据读取则断开
            .addHandlerLast(new WriteTimeoutHandler(30))   // 30秒写不出数据则断开
    )
    .handle((req, res) -> ...)
    .bindNow();

处理器	触发条件	行为
`ReadTimeoutHandler`	30 秒内未从 Channel 读取任何数据	抛出 `ReadTimeoutException`，关闭连接
`WriteTimeoutHandler`	30 秒内未完成写入操作	抛出 `WriteTimeoutException`，关闭连接

WriteTimeoutHandler 统计的是单次 Channel.write() 返回的 ChannelFuture 在指定时间内仍未完成。背压导致发送端「暂时不发」时根本没有发起 write，不会触发它；触发通常意味着数据已经压给内核、但对端长时间不确认（链路中断或对端不再读取），此时继续等只会累积内存。

内存安全：避免无界缓冲

永远不要假设客户端一定消费足够快。这种写法是危险的：

// 危险：先全部生产再发送，慢客户端会导致 OOM
List<String> allData = blockingFetchAllRows();  // 百万级结果
return res.sendString(Flux.fromIterable(allData));

正确做法：用响应式流本身做背压，让数据库查询与发送同速率推进。第 7 章 R2DBC 会展示如何用 Flux 直接包裹数据库结果集，无需全量缓冲。

同理，如果你用 Flux.interval 产生高频数据，务必考虑降采样或丢弃策略，不要无限制地累积在内存中。

eventd v2 网关的背压设计

eventd v2 网关层的背压策略：

HTTP 客户端请求
    → Gateway（Reactor Netty）
        → 检查 isWritable：不可写时丢弃非关键事件
        → 关键事件（如订单）进入有界队列
            → 转发至 WebFlux API 层
                → R2DBC 查询结果以 Flux 直通返回
                    ← 背压沿整条链路传递

这个设计确保：

非关键事件（心跳、日志）在拥塞时被丢弃
关键事件（交易指令）进入有界队列，队列满时返回 503 Service Unavailable
数据库查询使用响应式结果集，不缓冲全量数据

小结

机制	层级	控制点
TCP 滑动窗口	传输层	内核自动，应用不可见
Netty 写水位线	Netty Channel	`WRITE_BUFFER_WATER_MARK` + `isWritable()`
Reactor Netty send 背压	响应式流	`send()`/`sendString()` 内部 request(n) 联动
超时断开	应用层	`ReadTimeoutHandler` / `WriteTimeoutHandler`
有界队列/丢弃	业务层	`isWritable()` 检查 + 队列容量限制

HTTP 层的背压不是单点机制，而是从 TCP 窗口到业务队列的多层协作。理解每一层的职责范围，才能在真实系统中正确地配置保护参数。

第 1 章结束。第 2 章将从客户端视角出发：2.1 Reactor Netty HTTP 客户端。

继续阅读

探索更多技术文章

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

全部文章返回首页