《Netty 高性能网络编程》9.2 写队列与可写事件

本节目标：分清 write() 的入队语义与 flush() 的提交语义，用 ChannelFuture 判断真正的完成时机，并把广播写成「多次 write、一次 flush、显式处理失败」。
适用版本：JDK 25 + Netty 4.2.19.Final。

承接的问题

9.1 的 WritabilityPump 在循环里调用 ctx.write(frame)，循环结束后才 ctx.flush()。这个顺序不是随手写的：如果每次 write 都跟一次 flush，慢连接上的系统调用次数会随帧数线性增长，事件循环大部分时间花在 write 系统调用和 selector 唤醒上。linkd-v2 早期的广播实现就是后者：

for (Channel ch : channels) {
    ch.writeAndFlush(frame.retainedDuplicate());
}

这段代码在 5000 条连接上广播一帧，就是 5000 次写、5000 次 flush。更麻烦的是失败处理：writeAndFlush 返回的 ChannelFuture 被直接丢掉，广播失败多少条、哪些连接队列已经满了，没有任何记录。9.2 把这三个问题拆开：write() 和 flush() 各自做什么、批量写怎么组织、ChannelFuture 完成时到底意味着什么。

先说结论：write() 只负责把消息送进 pipeline 和待写队列，flush() 才请求事件循环真正写 socket；ChannelFuture 在消息完整写进 socket（或写失败）时完成，不是在对端收到时完成。 后面用本机输出把这个分界钉住。

核心机制一：write 入队，flush 提交

Netty 的写路径分两段。第一段是 write(msg)：消息从调用点沿 pipeline 向 head 方向传播，经过编码器变成 ByteBuf，最后进入连接的待写队列，同时累加待写字节。第二段是 flush()：它沿同一方向传播一个 flush 请求，到达 head 后触发一次真正的 socket 写循环，把队列里的字节尽可能交给内核。

write() 返回的 ChannelFuture 覆盖的是整个两段过程，所以只 write 不 flush 时，future 一直不完成。本机用一个会正常读取的服务端做对照：客户端在 channelActive 里先连续 write 20 个 1 KiB 的 ByteBuf，不 flush，打印监听器计数；然后 flush，再打印；最后连续 20 次 writeAndFlush，再打印。输出是：

after 20 write(), completed=0 writable=true
after flush(), completed=20
after 20 writeAndFlush(), completed=40

本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8，Netty 4.2.19.Final；单次运行，未做统计。第一行说明 20 次 write() 都已经入队，但没有任何一个 future 完成；此时通道仍然可写，因为 20 KiB 还没超过默认的 64 KiB 高水位。第二行说明一次 flush() 把队列里的 20 个消息全部写进 socket，20 个监听器同步执行完毕。第三行说明 writeAndFlush 等价于「写一个、flush 一次」，20 次调用又完成了 20 个 future。

这里有两个容易忽略的细节。

第一，future 完成的线程是通道所属的 EventLoop 线程。在 channelActive 里调用时本来就在事件循环线程，监听器同步执行；从业务线程调用 write，任务会被投递到事件循环，future 也在那里完成。监听器里不要做阻塞操作，否则会卡住这个 EventLoop 上的所有连接。

第二，future 成功只代表字节交给了内核，不代表对端应用读到了。内核发送缓冲区、网络、对端接收缓冲区都在这之后。要确认对端收到，只能靠协议层 ACK：自定义协议的 PONG、HTTP 响应、WebSocket 的应用层回执。把「future 成功」当成「消息送达」是分布式系统里最常见的过度承诺。

ctx.write 和 ctx.channel().write 的起点也不同。ctx.write 从当前 handler 的位置向 head 找下一个出站 handler，会经过当前 handler 之前的编码器；ctx.channel().write 从 pipeline 的 tail 开始，会经过全部出站 handler。在 WritabilityPump 里用 ctx.write 是故意的：帧必须经过它前面的 LinkdFrameEncoder，不能从 tail 重新绕一圈把已经编码过的消息再编码一次。

核心机制二：批量写，末尾一次 flush

批量写的目标不是「少调用几个方法」，而是把 N 次 socket 写合并成尽量少的系统调用。Netty 的 NIO 传输在 flush 时会把待写队列里连续的 ByteBuf 收集起来，尽量用一次 gathering write 交给内核；队列里的消息越多，单次 flush 能合并的字节越多。对广播场景，这直接决定了每帧的固定开销。

批量写有一个上限问题。一次 publish 如果塞进十万帧，待写队列可能在一次事件循环任务里膨胀到几十 MB，首帧的发送延迟也被拉长；对端在此期间什么都收不到。批次要按字节数或时间切片，而不是按「一次业务调用里的全部消息」。linkd-v2 的广播批次上限是 256 KiB 或 5 ms，先到哪个就切，切完立即 flush，再排下一批。

还有一个顺序细节：flush() 是每连接独立的操作，不能跨连接合并。广播给 5000 条连接，最少也是 5000 次 flush；能省的是「每条连接内每帧一次 flush」，不是「所有连接共享一次 flush」。把 flush 提到连接循环外面没有意义，因为 flush 只作用于调用它的那个 channel。

核心机制三：不可写时继续写，队列只会更长

9.1 说过 isWritable() 是缓存状态，不是写操作的闸门。write() 不会因为通道不可写而拒绝消息，也不会抛异常；它照样编码、入队、累加待写字节。这是 Netty 的刻意设计：写操作不阻塞调用方，失败通过 ChannelFuture 异步报告。

本机用不读的对端、调小发送缓冲区、把水位设为 (8192, 16384) 做了一次对照：客户端连续写 8 个 64 KiB 的 ByteBuf。第一次写入后通道就已经不可写，但后续 7 次 writeAndFlush 全部被接受。输出是：

  event writable=false
write 1x64KiB writable=false ok=0 failed=0
write 2x64KiB writable=false ok=0 failed=0
write 3x64KiB writable=false ok=0 failed=0
write 4x64KiB writable=false ok=0 failed=0
write 5x64KiB writable=false ok=0 failed=0
write 6x64KiB writable=false ok=0 failed=0
write 7x64KiB writable=false ok=0 failed=0
write 8x64KiB writable=false ok=0 failed=0
before close writable=false ok=6 failed=0
after close ok=6 failed=2

前 8 行说明「不可写」不是「不能写」：8 次调用都成功返回，future 没有立刻失败，监听器计数也还是 0。到关闭前，内核慢慢接收了其中 6 个，future 依次成功；剩下 2 个还在队列里，关闭时被标记为失败。队列增长不会以异常的形式出现，只会以内存占用和延迟的形式出现。

每个排队消息都占着它的大小，ByteBuf 还可能占着池化的堆外内存。没有上限的写入最终会把 direct memory 吃完，表现为 OutOfMemoryError: Direct buffer memory 或者进程被系统杀掉；在那之前，GC 日志和堆内存看起来都很正常，因为问题不在堆上。linkd-v2 因此给每个连接设了「待写字节硬上限」，超过就关闭连接并记录丢弃数，而不是继续接受。

失败处理要挂在 ChannelFuture 上：

ch.writeAndFlush(frame).addListener((ChannelFutureListener) f -> {
    if (!f.isSuccess()) {
        failed.incrementAndGet();
        LOG.log(System.Logger.Level.WARNING, "broadcast failed", f.cause());
    }
});

三个边界必须说清楚。第一，失败的写操作会由 Netty 释放消息（如果消息实现了 ReferenceCounted），监听器里再 release() 就是二次释放；要重试，必须在写之前 retainedDuplicate() 留一份。第二，监听器不承诺在调用线程执行，计数要用原子类型。第三，future 失败不等于连接已经关闭，可能只是单次写被拒绝；重试前检查 ch.isActive()，并给重试次数设上限，否则一个坏连接会无限占用事件循环。

在 linkd-v2 里落地：BroadcastPublisher

BroadcastPublisher 负责把一批 LinkdFrame 发给 ChannelGroup 里的所有连接。它做四件事：跳过非活跃和不可写连接、每连接批量 write、每连接一次 flush、用监听器统计失败。批次切片由调用方完成，publish 只发送已经切好的一批；这样切片策略可以按业务调整，发布器本身不用知道帧的语义。跳过不可写连接是 9.1 的状态信号在这里的直接应用；不跳过就会把队列继续推高。

package com.plume.linkd.v2.server;

import com.plume.linkd.v2.protocol.LinkdFrame;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.group.ChannelGroup;

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

public final class BroadcastPublisher {

    private static final System.Logger LOG = System.getLogger("linkd.broadcast");

    private final ChannelGroup group;
    private final AtomicLong delivered = new AtomicLong();
    private final AtomicLong failed = new AtomicLong();
    private final AtomicLong skipped = new AtomicLong();

    public BroadcastPublisher(ChannelGroup group) {
        this.group = group;
    }

    public int publish(List<LinkdFrame> frames) {
        if (frames.isEmpty()) {
            return 0;
        }
        int targets = 0;
        for (Channel ch : group) {
            if (!ch.isActive()) {
                continue;
            }
            if (!ch.isWritable()) {
                skipped.incrementAndGet();
                continue;
            }
            ChannelFuture last = null;
            for (LinkdFrame frame : frames) {
                last = ch.write(frame);
            }
            ch.flush();
            if (last != null) {
                last.addListener((ChannelFutureListener) f -> {
                    if (f.isSuccess()) {
                        delivered.incrementAndGet();
                    } else {
                        failed.incrementAndGet();
                        LOG.log(System.Logger.Level.WARNING,
                                "publish failed on " + ch.remoteAddress(), f.cause());
                    }
                });
            }
            targets++;
        }
        return targets;
    }

    public long delivered() {
        return delivered.get();
    }

    public long failed() {
        return failed.get();
    }

    public long skipped() {
        return skipped.get();
    }
}

代码里有两个刻意的取舍。

监听器只挂在最后一个 future 上。 广播一批帧时，如果逐帧挂监听器，5000 条连接 × 20 帧就是 10 万个监听器对象，每次广播都分配一次，GC 压力不小。只挂最后一个 future，能用一次回调观察这条连接的整批写是否走完；代价是它只代表最后一次写，前面的帧如果因为消息本身的问题失败，计数不会反映在 last 上。编码阶段的异常通常在 write 调用里直接抛出，也不会经过这个监听器。要精确到帧，就得逐帧挂监听器，或者让编码器在编码阶段就把错误暴露给发布器。

跳过不可写连接只记录 skipped，不在这里重试。 重试需要知道「这个连接什么时候恢复可写」，那是 9.1 的 channelWritabilityChanged 的职责。发布器只做快照式发送：本轮不可写就跳过，等恢复事件或下一轮快照再补。把重试塞进发布循环会让一次广播的耗时取决于最慢的连接，广播线程会被拖住。

如果消息本身是可共享的预编码 ByteBuf，可以让所有连接共享底层内存：写之前 frame.retainedDuplicate()，每条连接拿到独立读写索引、共享引用计数，写完由 Netty 释放各自的 duplicate。注意共享的前提是「所有连接发的内容完全相同」，只要有一帧按连接定制，就不能共享，必须各自编码。ConnectionRegistry.broadcast(ByteBuf) 用的就是这条路径；BroadcastPublisher 接收 LinkdFrame，让编码器在每条连接上各编码一次，牺牲一点 CPU 换实现简单。

重试和丢弃的策略要分开：

private void sendOrDrop(Channel ch, LinkdFrame frame, int attempt) {
    if (!ch.isActive()) {
        failed.incrementAndGet();
        return;
    }
    if (!ch.isWritable()) {
        skipped.incrementAndGet();
        return;
    }
    ch.writeAndFlush(frame).addListener((ChannelFutureListener) f -> {
        if (f.isSuccess()) {
            delivered.incrementAndGet();
        } else if (attempt < 2 && ch.isActive()) {
            ch.eventLoop().execute(() -> sendOrDrop(ch, frame, attempt + 1));
        } else {
            failed.incrementAndGet();
        }
    });
}

这里能安全重试，是因为 LinkdFrame 是 record LinkdFrame(int type, int flags, int requestId, byte[] payload)，消息体是普通 byte[]，不是引用计数对象；重试时重新编码即可。如果消息是 ByteBuf，失败后原对象已经被 Netty 释放，必须用 retainedDuplicate() 预留副本，或者在写之前就复制一份。重试次数上限和「仅对可重试消息重试」两条同时成立，才不会被一个坏连接拖死。

常见错误与边界

错误一：在 broadcast 循环里逐帧 writeAndFlush。

// 反例：N 帧就是 N 次 flush
for (LinkdFrame frame : frames) {
    ch.writeAndFlush(frame);
}

每次 flush 都会触发一次 socket 写循环和一次 selector 唤醒；帧很小时，固定开销远大于负载本身。批量写的正确形态是「帧循环里只 write，循环外一次 flush」，flush 的次数从帧数降到连接数。

错误二：在 EventLoop 线程里调用 future.sync()。 sync() 会阻塞等待 future 完成，而 future 的完成依赖同一个 EventLoop 继续跑；在 EventLoop 线程里等自己，轻则超时，重则死锁。事件循环里只用 addListener，需要等待结果时把后续逻辑写进监听器。

错误三：在失败监听器里 ReferenceCountUtil.release(msg)。

// 反例：写失败时 Netty 已经释放过 msg
ch.write(buf).addListener(f -> {
    if (!f.isSuccess()) {
        ReferenceCountUtil.release(buf);
    }
});

write 失败时，Netty 会释放消息，监听器再释放就是二次释放，引用计数变成负数，可能触发 IllegalReferenceCountException 或者破坏池化内存。要重试就先 retainedDuplicate()，让写操作消耗副本，原对象留给自己。

错误四：只 write 不 flush，然后疑惑对端为什么没收到。 消息已经编码、已经入队，但没有任何人请求写 socket。它可能一直待到连接关闭才被释放。write() 不是「发出去」，只是「排好队」；每个写路径都必须有明确的 flush 点，或者由上游的批量逻辑统一 flush。

错误五：把 future 成功当成对端已处理。 future 在字节交给内核时完成；对端可能还没读、读到一半连接断了、应用层处理失败。业务上的「送达」要靠协议回执，不能靠 ChannelFuture.isSuccess()。

边界：批量大小没有万能值。 批次太小，flush 次数多；批次太大，首帧延迟高、单连接待写字节容易越线。按「单连接待写字节上限」和「可接受的首帧延迟」两个约束反推批次，再用 9.1 的水位线做保护，而不是抄一个固定的帧数。

边界：flush 不承诺一次写完。 内核发送缓冲区满时，flush 只写出去一部分，剩余消息留在队列，等下一次可写事件继续。ChannelFuture 在剩余部分也写完后才完成；这也是为什么慢连接上的 future 会长时间不完成，但也不会报错。

小结

• write() 编码并入队，flush() 才触发 socket 写；只 write 不 flush，ChannelFuture 一直不完成。
• writeAndFlush 等于每次写都提交一次；批量场景应在循环里 write、循环外按连接 flush 一次。
• ChannelFuture 成功表示消息完整交给内核，不表示对端收到或处理；送达要靠协议层回执。
• 通道不可写时 write 仍然接受消息，队列继续增长；没有硬上限时最终会耗尽 direct memory。
• 失败处理用监听器；Netty 会释放失败的消息，重试要用 retainedDuplicate() 或可重新编码的不可变消息。
• BroadcastPublisher 跳过不可写连接、每连接批量写、一次 flush，并用监听器统计成功与失败。

写方向的水位和批量策略已经有了，读方向还默认开着自动读取。9.3 把 AUTO_READ 关掉，用一次手动 read() 拉起一轮读循环，再把慢消费者的队列上限接进 channelReadComplete。

阅读导航：上一节：9.1 Channel.isWritable 与高低水位 · 下一节：9.3 读背压与 AUTO_READ 。