《Netty 高性能网络编程》9.3 读背压与 AUTO_READ

本节目标：把读方向从「Netty 自动读」改成「应用按容量拉取」，用一次 read() 拉起一轮读循环，并在 channelReadComplete 决定是否继续。
适用版本：JDK 25 + Netty 4.2.19.Final。

承接的问题

9.1 和 9.2 管的是写方向：待写字节超过高水位就暂停生产者，批量写之后一次 flush，失败通过 ChannelFuture 报告。读方向默认没有对应的刹车。ChannelOption.AUTO_READ 的默认值是 true，含义是：连接激活后自动发起读，每次读循环结束、channelReadComplete 触发之后，只要通道还活跃，Netty 就自动再发起下一轮。

对 linkd-v2 的 9093 自定义协议，这意味着入站 REQUEST 的解码速度只取决于网络和事件循环，不取决于业务处理速度。业务侧把 LinkdFrame 投递给一个有界队列，队列满了以后，默认行为是继续读、继续解码、继续投递，直到队列拒绝或者内存耗尽。队列拒绝通常表现为 RejectedExecutionException，更糟的是处理线程在事件循环里做阻塞操作，把整条 EventLoop 上的连接一起拖死。

读背压要做的就是：业务队列没有余量时，不发起下一轮读。内核接收缓冲区会因此填满，TCP 接收窗口收缩，压力沿着连接传回发送方。这和 9.1 的写水位是同一个思路的两个方向，但控制点不同：写方向看 Netty 的待写字节，读方向看应用队列的剩余容量。

核心机制一：AUTO_READ 决定谁发起下一轮读

AUTO_READ=true 时，读循环由 Netty 自己维持。连接激活时发起一次读，读到没有数据可读或者达到本轮上限后触发 channelReadComplete，然后自动注册下一次读兴趣。应用不需要调用 read()，也无法通过「不调用 read」来减速，因为 Netty 已经代劳了。

AUTO_READ=false 时，读循环变成拉模式：只有调用 Channel.read() 才会发起一轮；一轮结束后不会自动继续，必须再调用一次。channelReadComplete 仍然会触发，但它的语义变成「这一轮读完了」，而不是「马上要读下一轮」。

本机用一个服务端验证这个边界：服务端设置 AUTO_READ=false，客户端先发 10 个 16 字节的定长帧，600 ms 后再发 10 个；服务端在 300 ms 和 900 ms 各手动调用一次 read()，用 FixedLengthFrameDecoder(16) 切帧，统计 channelRead 与 channelReadComplete 次数。输出是：

client sent 10 x 16B
manual read #1, before reads=0
readComplete reads=10 completes=1
client sent second 10 x 16B
manual read #2, before reads=10
readComplete reads=20 completes=2
final reads=20 completes=2

本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8，Netty 4.2.19.Final；单次运行，未做统计。第一轮手动 read() 之前，服务端一个帧都没读到，尽管客户端已经把 10 个帧发到了内核缓冲区。手动 read() 之后，一轮读循环把 10 个帧全部读完，channelRead 调用 10 次，channelReadComplete 只调用 1 次。客户端随后再发 10 个帧，服务端计数停在 10，直到第二次手动 read() 才变成 20。没有第二次 read()，第二批数据就一直留在内核里，不会自动被读走。

这里有两个容易误解的点。

第一，一次 read() 不等于一个消息。它是一轮读循环，可能读出很多帧，也可能一帧都读不到；channelReadComplete 的粒度是「轮」，不是「帧」。所以读背压的判断放在 channelReadComplete 里，而不是每读一帧就切换一次状态。

第二，channel.read() 可以从其他线程调用，它会安全地把读请求投递到通道所属的 EventLoop。但「要不要读」的状态判断应该在 EventLoop 线程做：两个线程同时看到「队列有余量」并各调用一次 read()，会多跑一轮；如果同时看到「队列满」而放弃，又可能在容量恢复后没人拉起下一轮。状态标志要么只在 EventLoop 线程读写，要么用原子变量并接受偶尔多读一轮。

核心机制二：channelReadComplete 是决定「下一轮」的位置

在 AUTO_READ=false 的拉模式里，channelReadComplete 是天然的决策点：这一轮已经读完了，现在看业务队列还有没有余量，有就 ctx.read() 拉起下一轮，没有就把状态置为暂停，等业务侧消费后再恢复。

写成代码有一个必须避免的写法：

// 反例：无条件续读，等于把 AUTO_READ=false 又打开了
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
    ctx.read();
}

这段代码里 AUTO_READ 确实是 false，但每次读完都立刻手动补一轮，效果和自动读没有区别。业务队列满的时候，读循环仍然一轮接一轮地跑，背压完全失效。channelReadComplete 里必须先判断容量，再决定是否 read()。

第二个容易写错的地方是恢复。业务线程消费完一批消息后，不应该直接改 paused 这种 EventLoop 线程私有的布尔值，也不应该让「判断能不能读」和「发起读」分散在两个线程里；否则两个线程可能同时看到有余量而多拉起一轮，或者都放弃而没人恢复。恢复动作要投递回 EventLoop：

public void resume() {
    ChannelHandlerContext c = ctx;
    if (c == null) {
        return;
    }
    c.channel().eventLoop().execute(() -> {
        if (!paused || !c.channel().isActive()) {
            return;
        }
        paused = false;
        c.read();
    });
}

paused 只在 EventLoop 线程读写，resume() 只是投递任务，不直接改状态；任务真正执行时再检查通道是否还活着。这样即使业务线程在连接关闭后仍然调用 resume()，也只会多投递一个空任务，不会向已关闭的通道继续发起读取。

还有一个时序边界：setAutoRead(false) 或「不调用 read()」只能阻止下一轮，不能撤回本轮已经读出来的消息。如果一轮读循环读出了 100 帧，业务队列只剩 10 个空位，第 11 帧到第 100 帧仍然会走完 channelRead。所以队列容量必须能吸收「一轮最坏情况」的消息数，或者业务 handler 在队列满时执行明确的拒绝策略。读背压控制的是速率，不是单批的原子性。

核心机制三：慢消费者、队列上限与拒绝策略

慢消费者的模拟不需要真的把线程睡在 EventLoop 上。用一个有界 ArrayBlockingQueue 加一个独立消费线程就够了：EventLoop 只负责把解码后的 LinkdFrame 放进队列，消费线程按自己的速度取走处理。队列的空位数和已排队的负载字节数共同构成读背压的输入信号。

package com.plume.linkd.v2.server;

import com.plume.linkd.v2.protocol.LinkdFrame;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

public final class SlowFrameConsumer implements ReadBackpressureHandler.CapacityGuard, AutoCloseable {

    private final ArrayBlockingQueue<LinkdFrame> queue;
    private final AtomicLong queuedBytes = new AtomicLong();
    private final long maxBytes;
    private final Thread worker;
    private volatile boolean running = true;
    private volatile Runnable resumeSignal;

    public SlowFrameConsumer(int capacity, long maxBytes, long perFrameMillis) {
        this.queue = new ArrayBlockingQueue<>(capacity);
        this.maxBytes = maxBytes;
        this.worker = Thread.ofVirtual().name("linkd-slow-consumer").start(() -> {
            while (running) {
                try {
                    LinkdFrame frame = queue.take();
                    Thread.sleep(perFrameMillis);
                    handle(frame);
                    queuedBytes.addAndGet(-frame.payload().length);
                    Runnable signal = resumeSignal;
                    if (signal != null) {
                        signal.run();
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        });
    }

    public void onConsumed(Runnable signal) {
        this.resumeSignal = signal;
    }

    public boolean offer(LinkdFrame frame) {
        if (!canAccept()) {
            return false;
        }
        int length = frame.payload().length;
        queuedBytes.addAndGet(length);
        if (!queue.offer(frame)) {
            queuedBytes.addAndGet(-length);
            return false;
        }
        return true;
    }

    @Override
    public boolean canAccept() {
        return queue.remainingCapacity() > 0 && queuedBytes.get() < maxBytes;
    }

    private void handle(LinkdFrame frame) {
        // 业务处理：查库、调用 libraryd、组装 RESPONSE
    }

    @Override
    public void close() {
        running = false;
        worker.interrupt();
    }
}

这个结构里，EventLoop 不会因为业务慢而阻塞；慢只体现在队列长度上。消费线程每取走一帧就执行一次 resumeSignal，它指向 ReadBackpressureHandler.resume()；恢复逻辑自己会判断是否真的需要拉起下一轮，重复调用是安全的。

队列上限怎么定，取决于三个量：一轮读最多能产生多少帧、单帧最大多大、进程愿意为每个连接花多少内存。前两个由协议和解码器决定，第三个是预算。linkd-v2 的 REQUEST 单帧上限 1 MiB，但平均只有几 KiB；队列按帧数限制会低估内存，按字节限制又要在 offer 时累加负载长度。实际做法是两者都限：帧数上限挡住小帧洪泛，字节上限挡住大帧堆积，任一超限就进入拒绝策略。

拒绝策略只有三种，必须提前选：

linkd-v2 的 9093 请求通道选「暂停 + 超时关闭」：队列满时停止读，超过配置的暂停时限仍然没有余量，就回一个 ERROR 帧并关闭连接。这样正常慢消费者只是被限速，完全 stalled 的消费者不会永久占着连接和队列。

核心机制四：RCVBUF_ALLOCATOR 与流量整形的边界

ChannelOption.RCVBUF_ALLOCATOR 经常被当成读背压开关，但它不是。它配置的是「每次 read 分配多大的缓冲区、读多少字节」的策略，影响单轮读的字节上限和内存分配，不检查应用队列，也不会因为业务处理不过来就停止读取。把它调小，能让一轮读少读一点、消息批次小一点，但 EventLoop 仍然会在 channelReadComplete 后继续读下一轮。

ChannelOption.SO_RCVBUF 调的是内核接收缓冲区大小。它确实会影响 TCP 窗口：缓冲区小，窗口收缩得快，发送方更早被拦住。但只要 Netty 还在读，内核缓冲区就会被搬空，窗口又会打开；SO_RCVBUF 改变的只是「能吸收多少突发」，不是「应用处理多快」。真正的应用层背压只有一个动作：不发起下一轮读。

ChannelTrafficShapingHandler 是另一个层级的工具。它的构造器接收写速率、读速率和检查间隔，单位是字节每秒：

pipeline.addLast(new ChannelTrafficShapingHandler(
        2 * 1024 * 1024,   // 写限速 2 MiB/s
        4 * 1024 * 1024,   // 读限速 4 MiB/s
        1000));            // 检查间隔 1s

它的限速是每连接的平均速率，按检查间隔结算配额；超配额的写会被延迟，读会被推迟到下一个配额窗口。它适合「不让单条连接占满带宽」的场景，但不能替代应用队列上限，原因有四条。

• 它是每连接的，不是全局的。 ChannelTrafficShapingHandler 为每条连接维护独立状态和定时任务；一百条连接各自 2 MiB/s，总速率仍然是 200 MiB/s。需要全局聚合时要用别的机制，不能把每连接限速当成进程级保护。
• 它限的是字节速率，不是消息积压。 业务队列里的 LinkdFrame 数量、单帧处理耗时、下游 libraryd 的延迟，都不在它的视野里。
• 它延迟写，不阻止生产。 写超配额时消息会在 handler 内排队，队列同样占内存；如果上游继续生产，延迟只会转化成更长的队列。
• 它有定时器开销。 每条连接一个检查任务，连接数很大时，定时任务的调度成本要计入容量规划。

所以顺序应该是：应用队列上限负责「业务能不能处理」，AUTO_READ 负责「要不要继续从内核读」，流量整形负责「链路带宽怎么分」。三者不能互相替代。

在 linkd-v2 里落地：ReadBackpressureHandler

ReadBackpressureHandler 放在 LinkdFrameDecoder 之后、业务 handler 之前。它不碰消息内容，只看容量信号：channelActive 时拉起第一轮读，channelReadComplete 时决定是否拉起下一轮，业务侧消费后调用 resume() 恢复。

package com.plume.linkd.v2.server;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public final class ReadBackpressureHandler extends ChannelInboundHandlerAdapter {

    public interface CapacityGuard {
        boolean canAccept();
    }

    private final CapacityGuard guard;
    private volatile ChannelHandlerContext ctx;
    private boolean paused;

    public ReadBackpressureHandler(CapacityGuard guard) {
        this.guard = guard;
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        this.ctx = ctx;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        ctx.fireChannelActive();
        ctx.read();
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        if (!guard.canAccept()) {
            paused = true;
            ctx.channel().config().setAutoRead(false);
        } else if (!paused) {
            ctx.read();
        }
        ctx.fireChannelReadComplete();
    }

    public void resume() {
        ChannelHandlerContext c = ctx;
        if (c == null) {
            return;
        }
        c.channel().eventLoop().execute(() -> {
            if (!paused || !c.channel().isActive()) {
                return;
            }
            if (!guard.canAccept()) {
                return;
            }
            paused = false;
            c.read();
        });
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        ChannelHandlerContext c = ctx;
        this.ctx = null;
        c.fireChannelInactive();
    }
}

装配时把 AUTO_READ 显式关掉，让读循环完全由 handler 控制：

ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childOption(ChannelOption.AUTO_READ, false)
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
                SlowFrameConsumer consumer = new SlowFrameConsumer(512, 8 * 1024 * 1024, 5);
                ReadBackpressureHandler backpressure = new ReadBackpressureHandler(consumer);
                consumer.onConsumed(backpressure::resume);
                ch.pipeline().addLast(new LinkdFrameDecoder());
                ch.pipeline().addLast(backpressure);
                ch.pipeline().addLast(new LinkdRequestHandler(consumer));
            }
        });

SlowFrameConsumer 先创建，ReadBackpressureHandler 再以它为容量来源；两者的反向依赖用 onConsumed 注入，避免构造器循环。consumer.onConsumed(backpressure::resume) 把「消费了一帧」这个事件接到「尝试恢复读」上，重复调用由 resume() 内部的 paused 检查过滤。

几个行为细节：

• channelActive 里的 ctx.read() 是启动开关。 AUTO_READ=false 时没有这一句，连接建立后不会自动出现第一轮读，日志里看不到任何错误，只会表现为客户端发了请求、服务端没反应。这是最常见的「静默挂起」。
• paused 只在 EventLoop 线程读写。 channelReadComplete 和 resume() 投递的任务都在同一个 EventLoop 上执行，因此不需要同步；ctx 用 volatile 是因为 resume() 可能从业务线程读取它。
• channelInactive 把 ctx 置空。 业务线程随后调用 resume() 时会直接返回，不会向已关闭的通道投递读请求。ctx = null 之后又调用 ctx.fireChannelInactive() 是错的；正确写法是先保存局部变量再 fire，或者把置空放到最后。上面的片段为了突出生命周期，保留了 ctx = null，实际实现应先 ChannelHandlerContext c = ctx; ctx = null; c.fireChannelInactive();。
• 容量判断只发生在轮末。 一轮读可能越过容量，队列要预留最坏批次的余量；linkd-v2 的 512 槽 / 8 MiB 队列对应的是「单轮读预算 + 业务消费速度」的估算值，不是拍脑袋的常量。

LinkdRequestHandler 在把帧放进 consumer 时，如果 offer 返回 false，说明队列在这一轮中途就满了。此时不能再读，但也不能把当前帧丢掉：请求/响应协议丢了请求，客户端会一直等到超时。linkd-v2 的处理是把当前帧放进一个每连接一个槽位的 pending，停止读，等消费线程腾出空间后先处理 pending；如果 pending 已经有值，说明一轮批次超过了队列容量，直接回 ERROR 并关闭连接，避免无限堆积。

常见错误与边界

错误一：设置 AUTO_READ=false 后忘记调用 read()。 连接能建立、能写响应，但一直读不到请求。没有异常，没有日志，只有超时。把第一轮 read() 放在 channelActive，并把「暂停后由谁恢复」写成明确的调用链，才能避免这个静默状态。

错误二：在 channelReadComplete 里无条件 ctx.read()。 这等于恢复了自动读，业务队列满时仍然一轮接一轮地读。读背压的判断必须包含容量条件，否则只是把 AUTO_READ 换了一种写法。

错误三：用 RCVBUF_ALLOCATOR 当背压。 调小它可以减少单轮读的字节数和内存分配，但读循环不会因此停止；业务队列仍然会被填满。背压的动作是「不读」，不是「每次少读一点」。少读一点只能减缓堆积速度，不能改变堆积趋势。

错误四：在业务线程直接改 paused 或调用 ctx.read()。 paused 是 EventLoop 线程的状态，跨线程读写会出现「业务看到已暂停、事件循环看到未暂停」的分裂；ctx.read() 本身线程安全，但如果和状态修改交错，可能多拉起一轮或少拉起一轮。统一投递到 eventLoop().execute，在同一个线程里判断和动作。

边界：读背压不能撤回已读出的消息。 一轮读循环的批量由内核可用字节、解码器和读缓冲区共同决定；channelReadComplete 只能决定下一轮。队列容量、单轮预算、拒绝策略三者要一起设计，不能指望「读完这一轮就自动停」。

边界：ChannelTrafficShapingHandler 不是全局限流器。 它按连接维护配额和定时任务，连接数一多，总速率是各连接配额之和；它也不理解业务队列。需要进程级保护时，用连接数上限、共享令牌桶或网关层限流，不要把它当成最后一道闸门。

小结

• AUTO_READ 默认是 true，Netty 在每轮 channelReadComplete 后自动续读；业务处理速度不参与这个决定。
• 设为 false 后，一次 Channel.read() 只拉起一轮读循环，读完必须再次调用；忘记第一轮会让连接静默挂起。
• channelReadComplete 是决定下一轮的位置，必须结合业务队列剩余容量判断，无条件 read() 等于没有背压。
• 一轮读可能越过容量，队列要预留最坏批次余量；拒绝策略提前选：暂停、丢弃或关闭。
• RCVBUF_ALLOCATOR 和 SO_RCVBUF 调的是读缓冲与内核窗口，不检查应用队列；应用层背压只有「不发起下一轮读」。
• ChannelTrafficShapingHandler 是每连接的平均速率整形，不阻止生产、不理解业务队列，不能替代队列上限。

读方向也停下了，接下来换一个入站协议：WebSocket。10.1 用 WebSocketServerProtocolHandler 处理握手和升级，再把文本帧交给会话表。

阅读导航：上一节：9.2 写队列与可写事件 · 下一节：10.1 WebSocketServerProtocolHandler 。