本节目标：用两条水位线控制停读与恢复，用硬上限兜住写队列，并把「读得太快」拆成每事件预算、令牌桶和帧边界丢弃三层来处理。
适用版本：JDK 25（Temurin 25.0.5+8）；SocketChannel、SelectionKey、ByteBuffer。
承接的问题
13.3 用 FileChannel.transferTo 把文件直接送进 socket，绕过了 7.2 的 WriteQueue。这条路径上 pendingBytes() 一直是 0，14.1 的水位永远不会触发；可内核发送缓冲区该满还是会满，transferTo 返回的字节数会一次比一次少。水位统计的是用户态排队字节，不是「对端还没收走的字节」，这两个量不是一回事。
更常见的问题出在普通写路径。7.2 的 WriteQueue 已经有 highWaterMark，超过就拒绝入队，但拒绝之后怎么办没有定义：是关连接、丢响应，还是让上游先别读？如果只用一个阈值判断「忙/不忙」，写队列在阈值附近来回摆动时，事件循环会不停摘掉又注册 OP_READ，连接在「停读」和「恢复读」之间抖动，吞吐反而更差。
这一节把背压拆成三个可独立观测的信号：用户态排队字节（写队列）、单连接读预算（每事件最多读多少）、下游是否吃得下（业务队列拒绝）。三者对应不同的动作，混在一起就会写出「一慢就关连接」的实现。
本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8。
核心机制一：为什么需要两条水位线
单阈值的问题是抖动。假设阈值 256 KiB，写队列在 255 KiB 和 257 KiB 之间波动：超过就停读，写出一个缓冲区后降到 255 KiB 又恢复读，下一次可读事件立刻把它推回 257 KiB。Selector 是水平触发的，这个循环每轮 select() 都会发生，兴趣位被反复改写，读到的数据却没增加多少。
高低水位用滞回（hysteresis）解决这个问题：超过高水位才停，降到低水位以下才恢复，中间那段是缓冲区，不产生动作。
排队字节 pending | 水位状态 | OP_READ | 动作 |
|---|---|---|---|
0 ≤ pending < high | NORMAL | 关注 | 正常读写 |
pending ≥ high | HIGH | 摘掉 | 停止读取，让 TCP 窗口收缩 |
pending ≤ low | NORMAL | 恢复 | 重新接收 |
pending > hardLimit | OVER_LIMIT | 摘掉 | 关闭连接，释放队列 |
low 和 high 的取值不是协议常量，而是按「一个事件循环轮次能写掉多少」估的。linkd 的起点是 high = 256 KiB、low = 64 KiB、hardLimit = 1 MiB，三个值都必须可配置。high - low 至少要大于一次可写事件能排空的字节数，否则恢复后立刻又越线；hardLimit 则要对齐单连接允许占用的最大堆外内存。
还有一条约束容易漏：low 不能是 0。如果只有写完整个队列才恢复读，那么一个大响应会把连接锁死到最后一字节写完，期间对端发来的任何控制帧都读不到。low = 64 KiB 表示「还剩一点尾巴，但已经有余力继续读了」。
核心机制二：暂停 OP_READ 把压力传回对端
背压的本质不是「在本地丢掉数据」，而是「让发送方慢下来」。TCP 已经提供了这个机制：本端不读，接收缓冲区填满，接收窗口缩到 0，对端内核的发送缓冲区随之填满，对端的 write 开始返回 0。压力就这样沿着连接传回上游。
所以暂停读取只需要改兴趣位：
// 暂停：只摘 OP_READ，保留写队列可能需要的 OP_WRITE
key.interestOpsAnd(~SelectionKey.OP_READ);
// 恢复：只加回 OP_READ，不动 OP_WRITE
key.interestOpsOr(SelectionKey.OP_READ);
不要写成 key.interestOps(SelectionKey.OP_READ)。绝对赋值会把写队列刚注册的 OP_WRITE 一起清掉，结果是「为了背压停读，顺手把还没写完的响应也停了」，写队列永远排不空，连接进入死锁。用 interestOpsAnd / interestOpsOr 只改自己关心的位，写队列的状态由写队列自己维护。
两个执行边界：
- 改兴趣位必须在连接所属的事件循环线程。
SelectionKey不是线程安全的；从定时线程或业务线程直接改，可能和select()内部状态竞争。跨线程时投递EventLoop.execute，必要时再selector.wakeup()。 - 摘掉
OP_READ不会丢弃已经在内核接收缓冲区里的字节。 它们还在，只是不再被搬进用户态。恢复读取后照常能读到，顺序不变。这也是「停读」比「读出来再丢」更可取的原因：不浪费一次拷贝，也不破坏帧边界。
恢复时机由写路径触发。WriteQueue.writeTo 每次写出一部分，pending 下降；降到 low 以下时把 OP_READ 加回去。不要在 onWritable 里无条件恢复，否则刚恢复就又越线。
核心机制三：硬上限、读预算与丢弃策略
高水位解决「暂时忙」，硬上限解决「已经救不回来」。pending > hardLimit 说明用户态排队的字节已经超过单连接预算，可能的原因是对端完全不读、应用层在生产速度远高于链路速度，或者响应体没有上限。这时继续接收只会把堆外内存吃光，正确动作是关闭连接并记录日志。
为什么不是丢掉队头缓冲区？因为写队列里的 ByteBuffer 是序列化后的字节流，队头可能是某个响应的前半段。丢掉它之后，后面的字节仍然会发出去，对端收到的是半截帧，长度域解析立刻错位。要么完整发完，要么关连接，不能从中间丢。
读方向的限速分三层，粒度从粗到细：
| 层 | 手段 | 作用 | 代价 |
|---|---|---|---|
| 连接级 | 摘 OP_READ | 让 TCP 窗口收缩，压力传回对端 | 恢复前完全停读 |
| 事件级 | 每事件读预算 | 防止单个连接独占事件循环 | 剩余数据留在内核 |
| 字节级 | 令牌桶 | 限制长期平均读速率 | 需要额外时钟与配额状态 |
事件级预算在 7.1 已经用 MAX_READ_PER_EVENT = 64 KiB 出现过，那里是常量；这里把它做成随水位变化的函数：NORMAL 时给满额，排队超过高水位一半时减半，HIGH 时给 0。令牌桶则负责跨事件的长期速率，避免「每个事件都读满预算」在大连接上累积成无上限吞吐。
丢弃只应该发生在帧边界，而且只对允许丢消息的协议生效。比如遥测上报、日志转发，丢掉一帧只损失一条样本；HTTP 请求/响应不能丢，丢了客户端会一直等响应。还要注意：TCP 上「丢弃一帧」仍然要把这一帧的字节从内核读出来，否则下一个帧的起点就找不到了。丢弃节省的是解码和业务处理，不是带宽。
在 linkd 里落地：WatermarkHandler
WatermarkHandler 是包在业务 Handler 外面的装饰器，自己不做解码，只负责水位状态机、兴趣位和读配额。它通过 LongSupplier 读取连接写队列的 pendingBytes()，通过注册时拿到的 SelectionKey 改兴趣位，因此不需要给 Connection 增加新方法。
package com.plume.linkd.core;
import java.nio.channels.SelectionKey;
import java.util.function.LongSupplier;
public final class WatermarkHandler implements Handler {
public enum Level { NORMAL, HIGH, OVER_LIMIT, CLOSED }
public record Limits(long highWaterMark, long lowWaterMark, long hardLimit,
int readBudget, long ingressBytesPerSecond) {
public Limits {
if (lowWaterMark <= 0 || highWaterMark <= lowWaterMark) {
throw new IllegalArgumentException("expect 0 < low < high");
}
if (hardLimit <= highWaterMark) {
throw new IllegalArgumentException("expect high < hardLimit");
}
if (readBudget <= 0 || ingressBytesPerSecond <= 0) {
throw new IllegalArgumentException("budgets must be positive");
}
}
}
public static final Limits DEFAULT_LIMITS = new Limits(
256 * 1024, 64 * 1024, 1024 * 1024, 64 * 1024, 4 * 1024 * 1024);
private final Handler next;
private final LongSupplier pendingBytes;
private final SelectionKey key;
private final Limits limits;
private final TokenBucket ingress;
private Level level = Level.NORMAL;
private long droppedFrames;
public WatermarkHandler(Handler next, LongSupplier pendingBytes,
SelectionKey key, Limits limits) {
this.next = next;
this.pendingBytes = pendingBytes;
this.key = key;
this.limits = limits;
long burst = Math.max(limits.readBudget(), limits.ingressBytesPerSecond());
this.ingress = new TokenBucket(burst, limits.ingressBytesPerSecond(), System::nanoTime);
}
@Override
public void onReadable(Connection c) {
if (level == Level.HIGH || level == Level.OVER_LIMIT || level == Level.CLOSED) {
return;
}
next.onReadable(c);
refresh(c);
}
@Override
public void onWritable(Connection c) {
next.onWritable(c);
refresh(c);
}
@Override
public void onClosed(Connection c) {
next.onClosed(c);
level = Level.CLOSED;
}
public Level refresh(Connection c) {
if (level == Level.CLOSED) {
return level;
}
long pending = pendingBytes.getAsLong();
if (pending > limits.hardLimit()) {
level = Level.OVER_LIMIT;
key.interestOpsAnd(~SelectionKey.OP_READ);
c.close();
return level;
}
if (level == Level.OVER_LIMIT) {
return level;
}
if (level == Level.NORMAL && pending >= limits.highWaterMark()) {
level = Level.HIGH;
key.interestOpsAnd(~SelectionKey.OP_READ);
} else if (level == Level.HIGH && pending <= limits.lowWaterMark()) {
level = Level.NORMAL;
key.interestOpsOr(SelectionKey.OP_READ);
}
return level;
}
public long acquireRead(int wanted) {
return ingress.tryAcquire(wanted);
}
public int readBudget() {
if (level == Level.HIGH || level == Level.OVER_LIMIT || level == Level.CLOSED) {
return 0;
}
long half = limits.highWaterMark() / 2;
return pendingBytes.getAsLong() >= half
? Math.max(1, limits.readBudget() / 2) : limits.readBudget();
}
public boolean shouldDropFrame(boolean lossTolerant) {
return lossTolerant && level == Level.HIGH;
}
public void recordDrop() {
droppedFrames++;
}
public Level level() {
return level;
}
public long droppedFrames() {
return droppedFrames;
}
public static final class TokenBucket {
private final long capacity;
private final long refillPerSecond;
private final LongSupplier clock;
private long tokens;
private long lastNanos;
public TokenBucket(long capacity, long refillPerSecond, LongSupplier clock) {
if (capacity <= 0 || refillPerSecond <= 0) {
throw new IllegalArgumentException("capacity and refill must be positive");
}
this.capacity = capacity;
this.refillPerSecond = refillPerSecond;
this.clock = clock;
this.tokens = capacity;
this.lastNanos = clock.getAsLong();
}
public long tryAcquire(long wanted) {
refill();
long granted = Math.min(wanted, tokens);
tokens -= granted;
return granted;
}
public long available() {
refill();
return tokens;
}
private void refill() {
long now = clock.getAsLong();
long elapsed = now - lastNanos;
if (elapsed <= 0) {
return;
}
lastNanos = now;
long added = elapsed / 1_000_000_000L * refillPerSecond
+ elapsed % 1_000_000_000L * refillPerSecond / 1_000_000_000L;
tokens = Math.min(capacity, tokens + added);
}
}
}
几个行为细节：
refresh是唯一的兴趣位入口。onReadable与onWritable末尾都调用它，读路径和写路径看到的是同一个状态机，不会出现「读路径以为停着、写路径以为开着」。OVER_LIMIT和CLOSED是终态。 进入后不再被refresh拉回NORMAL；硬上限触发的是关闭，不是临时暂停。readBudget()返回 0 表示本事件不要读。 即使因为竞态收到了一个可读事件，读循环看到 0 也应该直接返回，而不是先读一块再判断。TokenBucket不做同步。 它只在连接所属的事件循环线程被调用；跨线程共享同一个桶会破坏tokens和lastNanos的一致性。onClosed不释放写队列。 队列由Connection.close()内部释放（7.3 的第五步），装饰器只改自己的状态。
装配时，pendingBytes 绑定到连接自己的写队列，key 来自注册返回值：
SelectionKey key = ch.register(selector, SelectionKey.OP_READ);
WriteQueue outbound = new WriteQueue(pool, 1024 * 1024);
WatermarkHandler watermarks = new WatermarkHandler(reader, outbound::pendingBytes,
key, WatermarkHandler.DEFAULT_LIMITS);
key.attach(watermarks);
读循环在每次 read 前取配额，配额用完就把控制权还给 select()：
int budget = watermarks.readBudget();
while (budget > 0) {
int allowed = (int) Math.min(budget, watermarks.acquireRead(budget));
if (allowed == 0) {
break;
}
ByteBuffer buf = pool.acquire(allowed);
int n = ch.read(buf);
if (n <= 0) {
pool.release(buf);
break;
}
budget -= n;
// flip 后交给解码器；水位置为 HIGH 时下一轮 budget 自然是 0
}
用真实 SelectionKey 和固定时钟跑一遍状态迁移与令牌桶。下面的输出由本机 JDK 25 编译运行，水位配置为 high=1024、low=256、hard=4096、每事件预算 512 字节，令牌桶容量 1024、每秒补 1024：
pending=0 ops_read=true level=NORMAL
pending=1024 ops_read=false level=HIGH
pending=512 ops_read=false level=HIGH
pending=200 ops_read=true level=NORMAL
pending=5000 ops_read=false level=OVER_LIMIT closed=true
grant=600 left=424
grant=424 left=0
after +1s grant=800 left=224
- 1024 达到高水位，
OP_READ被摘掉；512 仍在高低水位之间，状态保持HIGH，这就是滞回。 - 降到 200 才恢复
OP_READ，低于low=256。 - 5000 超过硬上限 4096，连接被关闭，状态进入终态。
- 令牌桶第一次申请 600 给出 600，剩 424；再申请 600 只能给出 424；推进 1 秒后补满到 1024，申请 800 后剩 224。
数字来自固定配置和固定时钟，换一组参数结论不变：停读由高水位触发，恢复由低水位触发，硬上限只负责关闭。
常见错误与边界
错误一：用 interestOps(0) 暂停读。
// 反例：把写队列的 OP_WRITE 一起清掉
key.interestOps(0);
写队列非空时，OP_WRITE 被清掉，事件循环再也收不到可写事件，排队的响应永远发不出去。暂停读只摘 OP_READ，其余位保持原样。
错误二：在 onReadable 里先读空再检查水位。
// 反例：读循环不设上限，水位检查放在循环之后
while ((n = ch.read(buf)) > 0) {
outbound.enqueue(buf);
}
if (outbound.pendingBytes() > high) {
key.interestOpsAnd(~SelectionKey.OP_READ);
}
对端高速发送时，这个循环可能搬几十 MB 才退出，其他连接在这个事件循环上全部饿死；而且 enqueue 的硬上限可能在循环中途就触发关闭，后续 read 拿到的是已关闭通道。预算要在循环条件里，不是循环之后。
错误三：硬上限触发后丢掉队头缓冲区继续跑。
// 反例：丢掉最老的响应，保留后面的
outbound.pollFirst();
写队列里是字节流，不是独立消息。丢掉队头会让后续字节从帧中间开始，对端的长度域解析直接错位。超过硬上限就关闭连接，把「响应丢了」变成「连接断了」，后者客户端能立刻感知并重试（是否可重试见 14.2）。
错误四：把 transferTo 的返回值当成「已发送」。 它返回本次写进 socket 的字节数，可能小于请求数；剩余部分要循环调用或排队等可写事件。这条路径没有用户态队列，水位看不见它。别在同一连接上混用 transferTo 和 WriteQueue 而不共享计数，否则水位会低估真实在途字节。
边界：暂停读只能减慢对端，不能约束对端已发的数据。 从摘掉 OP_READ 到对端感知窗口为 0，中间还有一个 RTT 的在途数据；这部分会进入内核接收缓冲区，不受水位控制。所以 hardLimit 要按「用户态 + 内核缓冲区」的总内存预算来定，而不是只看 WriteQueue。
边界：shouldDropFrame 只对允许丢消息的协议开放。 HTTP/1.1 的请求丢了就是超时，WebSocket 文本帧丢了就是消息缺失。linkd 只在遥测通道上打开 lossTolerant，其他通道在水位为 HIGH 时停读，不做丢弃。
小结
- 单阈值会让停读/恢复在边界抖动，高低水位用滞回把动作隔开；
low不能为 0，high - low要大于一次可写事件的排空量。 - 暂停读只摘
OP_READ，保留OP_WRITE；压力通过 TCP 接收窗口传回对端，不丢已到达的字节。 - 硬上限是关闭线，不是丢弃线；写队列里是字节流，从中间丢会破坏帧边界。
WatermarkHandler用LongSupplier读排队字节，用SelectionKey改兴趣位；OVER_LIMIT与CLOSED是终态。- 读限速分连接级停读、事件级预算、字节级令牌桶三层；丢弃只在帧边界且只对允许丢消息的协议生效。
- 本机输出验证了
NORMAL → HIGH → NORMAL → OVER_LIMIT的迁移和令牌桶的配额递减，配置与时钟都是固定的。
背压让本端不再无限接收，但「等多久算失败」还没有答案。下一节把连接超时、读超时和请求期限分开，再看重试为什么必须和幂等绑在一起。
阅读导航：上一节：13.3 内存映射与文件传输 · 下一节：14.2 超时、重试与幂等边界 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。