《Netty 高性能网络编程》2.2 Channel 生命周期与 ChannelFuture

本节目标：说清一条连接从注册到注销会触发哪些事件，ChannelFuture 的完成、失败和监听器在哪个线程执行，以及 closeFuture() 在优雅关闭里到底等到什么。
适用版本：JDK 25 + Netty 4.2.19.Final。

承接的问题

2.1 的 LinkdServer 能绑定 9090 和 9093，但绑定成功只说明监听 socket 就绪，不说明连接会正常走完。ConnectionRegistry 需要在连接建立时把 channel 放进去，在连接断开时移除；如果挂错事件，注册表里会留下已经关闭的 channel，broadcast 时不断抛 ClosedChannelException，size() 也会一直虚高。

关闭服务器时也有一个容易被忽略的边界：LinkdServer.awaitClose() 里的 httpChannel.closeFuture().sync() 只等到监听 socket 关闭。已经接受的连接属于 worker group，监听 channel 的 closeFuture() 不会替它们收尾。要理解这一点，必须把 Channel 的生命周期事件和 ChannelFuture 的完成语义分开看。

本节用 LinkdServer 的 9093 自定义协议连接做主线：连接建立时注册、断开时移除、写响应后关闭、连接失败时记录原因。所有代码只使用 io.netty.channel 与 io.netty.bootstrap 的公开 API，不依赖 EmbeddedChannel 或内部测试类。

本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8，Netty 4.2.19.Final。下面的输出没有标注「本机输出」的，都是结构示意，实际文本以运行结果为准。

核心机制一：一条连接的五段状态

Netty 把一条 channel 的生命周期拆成五个入站事件。它们都发生在该 channel 绑定的那个 EventLoop 线程上，顺序如下：

服务端 accepted channel 的 channelActive 在 accept 完成、channel 注册之后触发；客户端 channel 的 channelActive 在 TCP 建连成功后触发。连接失败时，channelActive 不会出现，但 channelRegistered 可能已经发生，随后是 channelInactive 和 channelUnregistered。所以「没收到 active」和「没注册过」是两种不同的失败阶段，日志要分开记录。

把事件打印出来只需要一个入站 handler：

package com.plume.linkd.v2.server;

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

public final class LifecycleLoggingHandler extends ChannelInboundHandlerAdapter {

    private static final System.Logger LOG = System.getLogger(LifecycleLoggingHandler.class.getName());

    private final String name;

    public LifecycleLoggingHandler(String name) {
        this.name = name;
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) {
        log(ctx, "registered");
        ctx.fireChannelRegistered();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        log(ctx, "active");
        ctx.fireChannelActive();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        log(ctx, "read " + msg.getClass().getSimpleName());
        ctx.fireChannelRead(msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        log(ctx, "readComplete");
        ctx.fireChannelReadComplete();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        log(ctx, "inactive");
        ctx.fireChannelInactive();
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) {
        log(ctx, "unregistered");
        ctx.fireChannelUnregistered();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOG.log(System.Logger.Level.WARNING, name + " exception", cause);
        ctx.close();
    }

    private void log(ChannelHandlerContext ctx, String event) {
        LOG.log(System.Logger.Level.INFO, "{0} {1} id={2} loop={3}",
                name, event, ctx.channel().id().asShortText(), ctx.channel().eventLoop());
    }
}

几个细节：

• 每个事件都要 fire 给下一个 handler。 ChannelInboundHandlerAdapter 不会自动传播；不调用 ctx.fireChannelRead(msg)，后面的解码器和业务 handler 永远收不到数据。ctx.fire... 只是把事件传下去，不改变消息所有权。
• channelRead 的消息所有权在传递时转移。 LifecycleLoggingHandler 把 msg 原样传给下一个 handler，由下一个 handler 负责释放；如果它决定不传，必须自己调用 ReferenceCountUtil.release(msg)，否则 ByteBuf 泄漏。
• channelReadComplete 不是「读完一条消息」。 它表示这一轮读事件处理结束，可能已经读了多个消息，也可能一个都没读到。不要在这个回调里假设 channelRead 一定发生过。
• exceptionCaught 不自动关闭连接。 这里显式 ctx.close()；如果异常可以恢复（例如单帧解码失败但连接还能继续），应该记录后决定是否关闭，而不是无条件关。

核心机制二：ChannelFuture 的完成语义

ChannelFuture 是 Netty 对异步操作结果的表示。bind、connect、writeAndFlush、close 都返回它。它不是一个「稍后自动变成结果」的占位符，而是一个可以查询、可以等待、可以挂监听器的状态对象：

• isDone() 表示操作已经结束，成功或失败都算结束。
• isSuccess() 只在成功时为 true；失败时 cause() 给出原因。
• sync() 阻塞当前线程直到完成，失败时把原因抛出来；await() 只等待，不主动抛失败原因。
• addListener 注册回调；future 完成时回调执行，已经完成时注册也会被立即调度。

监听器的执行线程是关键。ChannelFuture 的监听器由 channel 所属的 EventLoop 执行；如果注册时 future 已经完成，Netty 会通过 executor 通知，不要在监听器里假设自己一定在调用 addListener 的线程上。监听器里做阻塞操作同样会卡住这个 EventLoop。

客户端连接失败是一个典型场景：

EventLoopGroup group = new NioEventLoopGroup(1);
Bootstrap bootstrap = new Bootstrap()
        .group(group)
        .channel(NioSocketChannel.class)
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000)
        .handler(new ChannelInboundHandlerAdapter());

ChannelFuture future = bootstrap.connect("127.0.0.1", 1);
future.addListener((ChannelFutureListener) f -> {
    if (f.isSuccess()) {
        System.out.println("connected to " + f.channel().remoteAddress());
    } else {
        System.out.println("connect failed: " + f.cause());
    }
});
future.awaitUninterruptibly();
if (future.isSuccess()) {
    future.channel().closeFuture().sync();
}
group.shutdownGracefully().sync();

连接 127.0.0.1 的 1 端口通常会被拒绝，示意输出：

connect failed: java.net.ConnectException: Connection refused

具体异常文本随平台和网络栈变化，但结构是固定的：future.isSuccess() 为 false，future.cause() 是失败原因，future.channel() 仍然存在但处于未激活状态。不要因为连接失败就假设 channel() 为 null；也不要对失败的 future 调用 closeFuture().sync() 然后期待它马上返回，是否关闭要看 Bootstrap 的失败处理。

ChannelFutureListener.CLOSE 是另一个常用常量。它把「操作完成后关闭 channel」写成一个监听器：

ChannelFuture writeFuture = ctx.writeAndFlush(response);
writeFuture.addListener(ChannelFutureListener.CLOSE);

无论写成功还是失败，CLOSE 都会关闭 channel。HTTP/1.1 的 Connection: close 响应、一次性请求的自定义帧都适合这种写法；如果写失败后还需要发一条错误响应，就不能直接挂 CLOSE，要在自己的监听器里判断 isSuccess() 再决定。

写方向没有对应的入站生命周期事件。writeAndFlush 是出站操作，返回的 ChannelFuture 表示数据已经交给 Netty 的 outbound 路径，不代表对端应用已经读到；写完后的关闭用 ChannelFutureListener.CLOSE，不要用 channelReadComplete 推断写完成。

closeFuture() 则是「channel 关闭」这个事件的 future：

Channel serverChannel = bootstrap.bind(9090).sync().channel();
serverChannel.closeFuture().addListener(f ->
        System.out.println("listener socket closed"));

监听器在监听 channel 的 EventLoop 上执行，适合做关闭日志、释放监听端口相关的资源。不要在监听器里调用 sync()，也不要把 closeFuture() 当成「所有子连接都关闭」的信号。

核心机制三：closeFuture 与优雅关闭

服务端关闭要分两层：监听 channel 和已接受的连接。监听 channel 的 closeFuture() 只覆盖第一层；已接受的连接由 worker group 管理。一个可等待的关闭方法长这样：

public void shutdownAndAwait() throws InterruptedException {
    if (httpChannel != null) {
        httpChannel.close().sync();
    }
    if (customChannel != null) {
        customChannel.close().sync();
    }
    boss.shutdownGracefully().sync();
    worker.shutdownGracefully().sync();
}

2.1 的 shutdown() 用 syncUninterruptibly() 提供不可中断版本；这里用会抛 InterruptedException 的版本，让调用方决定中断后是否继续等待。

顺序是「先停止 accept，再停事件循环」：

• httpChannel.close() 和 customChannel.close() 让两个监听 socket 停止接受新连接；sync() 等待关闭完成。
• boss.shutdownGracefully() 停掉 accept 线程；worker.shutdownGracefully() 停掉处理连接的线程。
• shutdownGracefully() 的 quiet period 和 timeout 有默认值，具体数值以 4.2 实现为准；需要控制时用带参数的重载，不要背默认数字。
• sync() 只能在 main、shutdown hook 或普通业务线程调用。在 EventLoop 线程里调用会阻塞事件循环，2.2 后面给出反例。

shutdownGracefully() 不等同于「等所有响应写完」。它给事件循环一个安静期，然后关闭仍注册的 channel；在途写是否排空，取决于连接状态、quiet period 和业务层是否先停止接收请求。真正的优雅关闭要在 ConnectionRegistry 里记录在途请求，先停止接受新请求，再等待计数归零，最后关闭 group；这部分在第 8、15 章展开。

在 linkd-v2 里落地：ConnectionRegistry 与事件挂载

ConnectionRegistry 用 ConcurrentHashMap 保存 channel，key 是 channel.id().asShortText()。add 和 remove 都在 channel 的 EventLoop 上调用；broadcast 可能从业务线程调用，所以容器必须是线程安全的。

package com.plume.linkd.v2.server;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public final class ConnectionRegistry {

    private final ConcurrentMap<String, Channel> channels = new ConcurrentHashMap<>();

    public void add(Channel channel) {
        channels.put(channel.id().asShortText(), channel);
    }

    public void remove(Channel channel) {
        channels.remove(channel.id().asShortText());
    }

    public int size() {
        return channels.size();
    }

    public void broadcast(ByteBuf frame) {
        for (Channel channel : channels.values()) {
            channel.writeAndFlush(frame.retainedDuplicate())
                    .addListener((ChannelFutureListener) future -> {
                        if (!future.isSuccess()) {
                            remove(future.channel());
                        }
                    });
        }
        frame.release();
    }
}

broadcast 的所有权规则要写清楚：每个 channel 拿到一个 retainedDuplicate()，写成功后由 pipeline 释放；循环结束后原始 frame 由本方法 release()。写失败时监听器把 channel 从注册表移除，避免下次广播继续撞上已关闭连接。size() 是并发容器上的近似值，只适合监控，不适合做精确配额。

把注册和移除挂到生命周期事件上：

package com.plume.linkd.v2.server;

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

public final class RegistryHandler extends ChannelInboundHandlerAdapter {

    private final ConnectionRegistry registry;

    public RegistryHandler(ConnectionRegistry registry) {
        this.registry = registry;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        registry.add(ctx.channel());
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        registry.remove(ctx.channel());
        ctx.fireChannelInactive();
    }
}

channelActive 里先 add 再 fire，让后续 handler 看到 active 时注册表已经一致；channelInactive 里先 remove 再 fire，让后续清理逻辑看不到已关闭的 channel。两个回调都在该连接的 EventLoop 上执行，ConnectionRegistry 的 map 操作不需要额外加锁。

把 2.1 匿名 handler 里的 add/remove 抽成 RegistryHandler，管道变成：

ch.pipeline().addLast(new LinkdFrameDecoder());
ch.pipeline().addLast(new LinkdFrameEncoder());
ch.pipeline().addLast(new RegistryHandler(registry));
ch.pipeline().addLast(new LifecycleLoggingHandler("custom"));
ch.pipeline().addLast(new SimpleChannelInboundHandler<LinkdFrame>() {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, LinkdFrame frame) {
        ctx.writeAndFlush(frame);
    }
});

入站事件从 head 到 tail：decoder、RegistryHandler、LifecycleLoggingHandler、业务 handler。出站事件从 tail 到 head：业务 handler 的 writeAndFlush 会反向经过 LifecycleLoggingHandler、RegistryHandler，最后由 LinkdFrameEncoder 编码。RegistryHandler 和 LifecycleLoggingHandler 都是入站 handler，出站事件直接穿过它们，不会干扰编码。

LinkdFrame 是普通 record，不是 ByteBuf，所以 SimpleChannelInboundHandler 的自动释放对它没有实际作用；如果以后把 payload 改成 ByteBuf，要么保留自动释放并确保不把同一个 ByteBuf 传给下游，要么覆写 autoRelease() 为 false 并明确释放责任。这是第 5 章引用计数的内容，这里先把所有权规则记住。

常见错误与边界

错误一：在 ChannelInboundHandlerAdapter 里不传播事件，也不释放消息。

// 反例：读完就丢，既不 fire 也不 release
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    System.out.println("got " + msg);
}

下游 handler 收不到数据，ByteBuf 也不会被释放，泄漏检测最终会在 ResourceLeakDetector 里报出来。要么 ctx.fireChannelRead(msg) 把所有权传下去，要么自己 ReferenceCountUtil.release(msg)。

错误二：把 channelInactive 当成「对端发了 FIN」。

// 反例：本地主动 close 也会走到这里
@Override
public void channelInactive(ChannelHandlerContext ctx) {
    metrics.peerClosed();
}

本地调用 channel.close()、写失败后关闭、超时关闭都会触发 channelInactive。它只表示「这条 channel 不再 active」，不区分是谁先关的。要区分方向，需要在协议层记录关闭原因，或者在半关闭场景里检查通道的输入/输出状态。

错误三：在 EventLoop 线程里 sync()。

// 反例：等待自己所在的 EventLoop 完成写操作
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ctx.writeAndFlush(msg).sync();
}

writeAndFlush 的 future 由当前 channel 的 EventLoop 完成；sync() 阻塞的正是这个线程，完成通知没有机会执行。Netty 4.2 的 DefaultPromise 会检测这种等待并抛出 BlockingOperationException（模块 netty-common），但不要依赖这个保护：在被 EventLoop 依赖的线程上等待，或者绕过检测，仍然会死锁。正确写法是 addListener：

ctx.writeAndFlush(msg).addListener(f -> {
    if (!f.isSuccess()) {
        f.channel().close();
    }
});

错误四：对 closeFuture() 也调用 sync() 然后阻塞在 handler 里。 这和上一条是同一个问题：closeFuture() 的完成同样由 EventLoop 驱动，在 handler 里等它等于等自己。关闭逻辑应该由外部线程发起，或者在监听器里继续下一步，而不是阻塞等待。

错误五：以为 shutdownGracefully() 会等所有响应写完。 它只负责事件循环的安静期和关闭；在途请求的排空要靠业务层的计数器、ConnectionRegistry 或第 8 章的关闭语义。把 shutdownGracefully() 当成「优雅」的全部，重启时仍可能丢掉正在写的响应。

小结

• Channel 的生命周期是 registered → active → read/readComplete → inactive → unregistered，全部发生在所属 EventLoop 线程。
• channelRegistered 可能成功而 channelActive 失败；连接失败时不要假设 active 一定发生过。
• ChannelFuture 的 sync() 阻塞等待并抛出失败原因，addListener 在 EventLoop 上回调；EventLoop 线程里只能用后者。
• ChannelFutureListener.CLOSE 在写操作完成后关闭 channel，成功和失败都会关；需要错误响应时自己判断 isSuccess()。
• closeFuture() 只代表当前 channel 关闭，监听 channel 的 closeFuture() 不包含已接受的子连接。
• ConnectionRegistry 在 channelActive 注册、channelInactive 移除；broadcast 用 retainedDuplicate() 给每个连接一份引用，最后释放原始 ByteBuf。

一条连接能正确地建立、注册、读写、关闭之后，下一节看这些行为受哪些参数控制：SO_BACKLOG、TCP_NODELAY、SO_KEEPALIVE、AUTO_READ、ALLOCATOR 和水位线分别作用在哪一层，以及哪些默认值不能凭记忆写死。

阅读导航：上一节：2.1 ServerBootstrap 与 Bootstrap · 下一节：2.3 ChannelOption 与 Socket 参数 。