摘要：Netty 是 Java 网络编程的事实标准框架，从 Dubbo、RocketMQ 到 Spring Cloud Gateway 都离不开它。本文从 NioEventLoop 的 Selector 轮询出发，深入解析 ChannelPipeline 的事件传播机制、AbstractUnsafe 的 I/O 委托模型、PooledByteBufAllocator 的 jemalloc 内存池实现，以及百万级长连接服务端、心跳保活、粘包拆包、大文件零拷贝传输等生产级实战方案。

---

一、引言

Java 原生 NIO 编程面临三大痛点：

1. API 复杂：Selector、Channel、Buffer 的交互涉及大量样板代码
2. 粘包拆包：TCP 流式传输不保留消息边界，需自行处理
3. 内存管理：ByteBuffer 无法扩展，堆外内存分配/回收成本高

Netty 通过 EventLoop 线程模型、Pipeline 责任链、ByteBuf 内存池三大核心设计，将网络编程的复杂度封装在框架内部。Spring Cloud Gateway、Dubbo、gRPC-java 等主流框架都深度依赖 Netty。

---

二、核心架构：EventLoop、Channel、Pipeline 三角模型

2.1 整体架构图

┌──────────────────────────────────────────────────────────────────┐
│                       Netty Application                           │
│                                                                   │
│  ┌──────────────────────────────────────────────────────────┐    │
│  │              EventLoopGroup (Boss Group)                  │    │
│  │    ┌──────────────────────────────────────────────┐      │    │
│  │    │         NioEventLoop (Select Loop)            │      │    │
│  │    │  ┌────────────────────────────────────────┐   │      │    │
│  │    │  │  Selector.select()                    │   │      │    │
│  │    │  │  ┌──────────┐  ┌──────────┐          │   │      │    │
│  │    │  │  │OP_ACCEPT │  │OP_ACCEPT │ ...      │   │      │    │
│  │    │  │  │  :8080   │  │  :8443   │          │   │      │    │
│  │    │  │  └────┬─────┘  └────┬─────┘          │   │      │    │
│  │    │  │       └─────────────┘                │   │      │    │
│  │    │  │  检测到新连接 → 创建 SocketChannel      │   │      │    │
│  │    │  │  → 注册到 Worker EventLoop 的 Selector │   │      │    │
│  │    │  └────────────────────────────────────────┘   │      │    │
│  │    └──────────────────────────────────────────────┘      │    │
│  └──────────────────────────────────────────────────────────┘    │
│                              │                                    │
│  ┌───────────────────────────▼──────────────────────────────┐    │
│  │            EventLoopGroup (Worker Group)                  │    │
│  │  ┌──────────────┐ ┌──────────────┐ ┌──────────────┐     │    │
│  │  │ NioEventLoop │ │ NioEventLoop │ │ NioEventLoop │     │    │
│  │  │ (Thread-1)   │ │ (Thread-2)   │ │ (Thread-N)   │     │    │
│  │  │              │ │              │ │              │     │    │
│  │  │ Selector     │ │ Selector     │ │ Selector     │     │    │
│  │  │ ┌──────────┐ │ │ ┌──────────┐ │ │ ┌──────────┐ │     │    │
│  │  │ │OP_READ   │ │ │ │OP_READ   │ │ │ │OP_READ   │ │     │    │
│  │  │ │OP_WRITE  │ │ │ │OP_WRITE  │ │ │ │OP_WRITE  │ │     │    │
│  │  │ └──────────┘ │ │ └──────────┘ │ │ └──────────┘ │     │    │
│  │  │              │ │              │ │              │     │    │
│  │  │ Channel-1    │ │ Channel-3    │ │ Channel-5    │     │    │
│  │  │ Channel-2    │ │ Channel-4    │ │ Channel-6    │     │    │
│  │  └──────────────┘ └──────────────┘ └──────────────┘     │    │
│  │          ↑               ↑               ↑               │    │
│  │          └── 一个 Channel 只绑定一个 EventLoop ───────────┘    │
│  └──────────────────────────────────────────────────────────┘    │
│                              │                                    │
│  ┌───────────────────────────▼──────────────────────────────┐    │
│  │              ChannelPipeline (每个 Channel 一个)           │    │
│  │                                                            │    │
│  │  HeadContext ──► Handler1 ──► Handler2 ──► ... ──► Tail  │    │
│  │       │              │             │                      │    │
│  │  ┌────┴────┐    ┌────┴────┐   ┌────┴────┐                │    │
│  │  │Inbound  │    │Inbound  │   │Outbound │                │    │
│  │  │Decoder  │    │Business │   │Encoder  │                │    │
│  │  │         │    │Handler  │   │         │                │    │
│  │  └─────────┘    └─────────┘   └─────────┘                │    │
│  │                                                            │    │
│  │  入站事件: channelRead → fireChannelRead → ...            │    │
│  │  出站事件: write → findContextOutbound → ...              │    │
│  └──────────────────────────────────────────────────────────┘    │
└──────────────────────────────────────────────────────────────────┘

2.2 三大核心组件

组件	职责	类比
EventLoop	I/O 事件循环线程，负责 Selector 轮询和任务调度	NIO 中的 Selector 线程
Channel	网络连接的抽象，封装 Socket 的读写操作	NIO 中的 SocketChannel
ChannelPipeline	处理器责任链，管理 Inbound/Outbound 事件传播	Servlet Filter 链

---

三、源码深度解析

3.1 EventLoop 线程模型：NioEventLoop 的事件循环

// io.netty.channel.nio.NioEventLoop
public final class NioEventLoop extends SingleThreadEventLoop {

    // ★ 每个 EventLoop 绑定一个 Selector
    private Selector selector;
    private final SelectorProvider provider;

    // ★ 任务队列：外部提交的任务（如 channel.write）先进入队列
    private final Queue<Runnable> taskQueue;

    // ★ 延迟任务队列：用于调度定时任务（如心跳超时检测）
    private final Queue<ScheduledFutureTask<?>> delayedTaskQueue = new PriorityQueue<>();

    @Override
    protected void run() {
        // ★ 核心事件循环（一直运行直到 EventLoop 关闭）
        for (;;) {
            try {
                // ★ 1. 计算下次定时任务的延迟，决定 select 的阻塞策略
                //    - 有定时任务即将到期: selectNow() 非阻塞
                //    - 无定时任务: select(timeout) 阻塞等待 I/O
                //    - wakenUp 被标记: 立即返回处理任务
                int strategy = selectStrategy.calculateStrategy(
                        selectNowSupplier, hasTasks());

                switch (strategy) {
                    case SelectStrategy.CONTINUE:
                        continue;  // 需要重试
                    case SelectStrategy.BUSY_WAIT:
                        // 忙等待策略（仅部分平台支持）
                    case SelectStrategy.SELECT:
                        // ★ 核心: Selector.select(timeout)
                        // timeout 由最近的定时任务决定
                        strategy = select(curDeadlineNanos);
                    default:
                        // select 返回了（有 I/O 事件或 wakenUp）
                }

                // ★ 2. 处理就绪的 I/O 事件
                if (strategy > 0) {
                    processSelectedKeys();
                }

                // ★ 3. 执行所有提交的任务（包括普通任务和定时任务）
                final long ioTime = System.nanoTime() - ioStartTime;
                ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);

            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }

    // ★ 处理 Selector 返回的就绪事件
    private void processSelectedKeys() {
        Set<SelectionKey> selectedKeys = selector.selectedKeys();
        for (SelectionKey k : selectedKeys) {
            NioUnsafe unsafe = (NioUnsafe) k.attachment();
            if (!k.isValid()) {
                unsafe.close(unsafe.voidPromise());
                continue;
            }
            try {
                int readyOps = k.readyOps();
                // ★ 处理 ACCEPT 事件（服务端）
                if ((readyOps & SelectionKey.OP_ACCEPT) != 0) {
                    unsafe.read();
                }
                // ★ 处理 READ 事件
                if ((readyOps & SelectionKey.OP_READ) != 0) {
                    unsafe.read();
                }
                // ★ 处理 WRITE 事件（发送缓冲区可写）
                if ((readyOps & SelectionKey.OP_WRITE) != 0) {
                    unsafe.forceFlush();
                }
                // ★ 处理 CONNECT 事件（客户端连接完成）
                if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                    unsafe.finishConnect();
                }
            } catch (CancelledKeyException e) {
                unsafe.close(unsafe.voidPromise());
            }
        }
    }
}

EventLoop 的关键设计点：

1. 线程绑定：一个 NioEventLoop 对应一个线程，一个 Channel 只注册到一个 EventLoop 上，保证 I/O 事件的顺序处理
2. I/O 与任务混排：事件循环中既处理 I/O 事件，也处理提交的任务，通过 ioRatio（默认 50）控制两者的时间比例
3. wakenUp 机制：外部提交任务时通过 selector.wakeup() 唤醒阻塞的 select 调用，避免任务延迟

3.2 ChannelPipeline 事件传播机制

// io.netty.channel.DefaultChannelPipeline
public class DefaultChannelPipeline implements ChannelPipeline {

    // ★ Pipeline 的头尾节点（双向链表）
    final AbstractChannelHandlerContext head;
    final AbstractChannelHandlerContext tail;

    // Inbound 事件传播（从 Head 到 Tail）
    // channelRegistered → channelActive → channelRead → channelReadComplete
    //                           ↑
    //    ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
    //    │   Head   │───►│ Decoder  │───►│  Biz     │───►│   Tail   │
    //    └──────────┘    └──────────┘    └──────────┘    └──────────┘

    // Outbound 事件传播（从 Tail 到 Head）
    // write → flush → bind → connect → disconnect → close
    //                           ↓
    //    ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
    //    │   Head   │◄───│ Encoder  │◄───│  Biz     │◄───│   Tail   │
    //    └──────────┘    └──────────┘    └──────────┘    └──────────┘

    // ★ channelRead 入站传播源码
    @Override
    public final ChannelPipeline fireChannelRead(Object msg) {
        // 从 Head 开始传播
        AbstractChannelHandlerContext.invokeChannelRead(head, msg);
        return this;
    }

    // AbstractChannelHandlerContext 的 invokeChannelRead
    static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
        final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            // ★ 如果在 EventLoop 线程中，直接调用
            next.invokeChannelRead(m);
        } else {
            // ★ 如果不在 EventLoop 线程，提交任务到 EventLoop 的任务队列
            executor.execute(() -> next.invokeChannelRead(m));
        }
    }

    // 查找下一个 Inbound Handler
    private AbstractChannelHandlerContext findContextInbound(int mask) {
        AbstractChannelHandlerContext ctx = this;
        do {
            ctx = ctx.next;
        } while ((ctx.executionMask & mask) == 0); // 跳过不处理该事件的 Handler
        return ctx;
    }

    // 查找下一个 Outbound Handler
    private AbstractChannelHandlerContext findContextOutbound(int mask) {
        AbstractChannelHandlerContext ctx = this;
        do {
            ctx = ctx.prev;
        } while ((ctx.executionMask & mask) == 0);
        return ctx;
    }
}

Pipeline 的关键设计点：

1. 双向链表：ChannelHandlerContext 组成双向链表，支持双向遍历
2. 事件分离：Inbound 事件（数据流入）和 Outbound 事件（操作流出）有独立的传播方向
3. 线程安全：所有 Handler 回调都在绑定的 EventLoop 线程中执行，无需额外同步
4. 动态修改：支持运行时添加/移除 Handler（如协议切换时替换编解码器）

3.3 ByteBuf 内存池：PooledByteBufAllocator

// io.netty.buffer.PoolArena (jemalloc 算法的 Java 实现)
abstract class PoolArena<T> implements PoolArenaMetric {

    // ★ 内存块分类
    // Small: [16B, 28KB] → 分成 32 个 Size Class
    // Normal: [32KB, 16MB] → 按 chunkSize 分配
    // Huge: > 16MB → 直接分配，不归池管理

    // ★ 核心数据结构
    private final PoolSubpage<T>[] smallSubpagePools;  // 32 个链表，按 Size Class 组织
    private final PoolChunkList<T> q050;  // 50% < 使用率 < 100%
    private final PoolChunkList<T> q025;  // 25% < 使用率 < 75%
    private final PoolChunkList<T> q000;  // 1%  < 使用率 < 50%
    private final PoolChunkList<T> qInit; // 0%  < 使用率 < 25%
    private final PoolChunkList<T> q075;  // 75% < 使用率 < 100%
    private final PoolChunkList<T> q100;  // 使用率 = 100%

    // PoolChunkList 形成链表，分配时按使用率从低到高查找
    // qInit → q000 → q025 → q050 → q075 → q100

    // ★ 分配内存
    private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf,
                          int reqCapacity) {
        final int sizeIdx = size2SizeIdx(reqCapacity);

        // 1. 尝试从线程本地缓存分配（无锁）
        if (cache.allocateSmall(this, buf, reqCapacity, sizeIdx)) {
            return;
        }

        // 2. 小内存: 从 PoolSubpage 分配
        if (isSmall(reqCapacity)) {
            allocateSmall(buf, reqCapacity, sizeIdx);
        }
        // 3. 普通内存: 从 PoolChunk 分配
        else if (reqCapacity <= chunkSize) {
            allocateNormal(buf, reqCapacity, sizeIdx);
        }
        // 4. 大内存: 直接分配
        else {
            allocateHuge(buf, reqCapacity);
        }
    }

    // ★ PoolSubpage: 管理 Chunk 内的小内存分配（slab 分配器）
    // 一个 PoolChunk (16MB) 被划分为多个 PoolSubpage
    // 每个 PoolSubpage 管理同一 Size Class 的内存块
    // 使用 bitmap 记录分配状态
}

ByteBuf 内存池的关键设计点：

1. PoolThreadCache：每个线程有本地缓存，小内存分配无锁，性能极高
2. Size Class：小内存按固定 Size Class 分配（16B, 32B, 48B…），减少内存碎片
3. PoolChunkList：按使用率组织 Chunk，优先从高使用率 Chunk 分配，提高内存利用率
4. Recycler：ByteBuf 对象本身也通过对象池复用，减少 GC 压力

---

四、ChannelHandler 深度解析

4.1 Inbound/Outbound Handler 分类

// ChannelInboundHandler: 处理入站事件
public interface ChannelInboundHandler extends ChannelHandler {
    void channelRegistered(ChannelHandlerContext ctx);   // Channel 注册到 EventLoop
    void channelActive(ChannelHandlerContext ctx);       // Channel 就绪（连接建立）
    void channelRead(ChannelHandlerContext ctx, Object msg); // 收到数据
    void channelReadComplete(ChannelHandlerContext ctx); // 本次读取完成
    void exceptionCaught(ChannelHandlerContext ctx, Throwable cause); // 异常
}

// ChannelOutboundHandler: 处理出站操作
public interface ChannelOutboundHandler extends ChannelHandler {
    void bind(ChannelHandlerContext ctx, SocketAddress local, ChannelPromise promise);
    void connect(ChannelHandlerContext ctx, RemoteAddress remote, local, promise);
    void disconnect(ChannelHandlerContext ctx, ChannelPromise promise);
    void close(ChannelHandlerContext ctx, ChannelPromise promise);
    void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise);
    void flush(ChannelHandlerContext ctx);
}

4.2 ChannelHandlerContext 的线程安全保证

// ★ 所有 I/O 操作最终都委托给绑定的 EventLoop
public class AbstractChannelHandlerContext implements ChannelHandlerContext {

    @Override
    public ChannelFuture write(Object msg) {
        return write(msg, newPromise());
    }

    @Override
    public ChannelFuture write(final Object msg, final ChannelPromise promise) {
        // ★ 检查当前线程是否是 EventLoop 线程
        if (executor.inEventLoop()) {
            // 直接执行
            next.invokeWrite(m, promise);
        } else {
            // ★ 提交到 EventLoop 的任务队列，保证线程安全
            executor.execute(() -> next.invokeWrite(m, promise));
        }
        return promise;
    }
}

---

五、生产级实战：百万级连接服务端

5.1 服务端完整启动代码

/**
 * ★ 百万级连接 Netty 服务端
 * 核心优化点：
 * 1. Boss/Worker 线程数合理配置
 * 2. TCP 参数优化（SO_BACKLOG/TCP_NODELAY/SO_KEEPALIVE）
 * 3. ByteBuf 内存池（PooledByteBufAllocator）
 * 4. 接收缓冲区自动调整（RecvByteBufAllocator）
 * 5. 写入缓冲区高低水位（WRITE_BUFFER_WATER_MARK）
 */
@Component
@Slf4j
public class NettyServer {

    @Autowired
    private NacosDiscoveryProperties nacosProperties;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private Channel serverChannel;

    @PostConstruct
    public void start() throws InterruptedException {
        // ★ Boss Group: 只处理 ACCEPT 事件，通常 1 个线程足够
        this.bossGroup = new NioEventLoopGroup(1,
            new ThreadFactoryBuilder()
                .setNameFormat("netty-boss-%d")
                .setDaemon(true)
                .build());

        // ★ Worker Group: 处理 I/O 读写
        // 公式: CPU 核数 * 2（可适当增大到 16-32 应对大量连接）
        int workerThreads = Runtime.getRuntime().availableProcessors() * 2;
        this.workerGroup = new NioEventLoopGroup(workerThreads,
            new ThreadFactoryBuilder()
                .setNameFormat("netty-worker-%d")
                .setDaemon(true)
                .build());

        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)

            // ★ TCP 参数优化
            .option(ChannelOption.SO_BACKLOG, 8192)          // 全连接队列大小
            .option(ChannelOption.SO_REUSEADDR, true)        // 端口复用
            .option(ChannelOption.TCP_NODELAY, true)         // 禁用 Nagle
            .childOption(ChannelOption.SO_KEEPALIVE, true)   // TCP 保活探测
            .childOption(ChannelOption.SO_RCVBUF, 65536)     // 接收缓冲区
            .childOption(ChannelOption.SO_SNDBUF, 65536)     // 发送缓冲区

            // ★ ByteBuf 配置
            .childOption(ChannelOption.ALLOCATOR,
                PooledByteBufAllocator.DEFAULT)               // 内存池
            .childOption(ChannelOption.RCVBUF_ALLOCATOR,
                new AdaptiveRecvByteBufAllocator(64, 1024, 65536))  // 自适应
            .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
                new WriteBufferWaterMark(32 * 1024, 64 * 1024))    // 水位控制

            // ★ 连接初始化
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ChannelPipeline pipeline = ch.pipeline();

                    // 1. 空闲检测（心跳超时）
                    pipeline.addLast(new IdleStateHandler(
                        60,   // readIdleTime: 60秒未收到数据
                        30,   // writeIdleTime: 30秒未发送数据
                        0,    // allIdleTime
                        TimeUnit.SECONDS));

                    // 2. 粘包拆包: 长度字段解码器
                    // 协议: | 4B length | 1B version | 1B cmd | 2B seq | N payload |
                    pipeline.addLast(new LengthFieldBasedFrameDecoder(
                        8 * 1024 * 1024,   // maxFrameLength: 最大 8MB
                        0,                  // lengthFieldOffset: length 字段偏移
                        4,                  // lengthFieldLength: length 字段 4 字节
                        8,                  // lengthAdjustment: 跳过 header
                        0,                  // initialBytesToStrip: 不剥离
                        true));             // failFast: 快速失败

                    // 3. 编解码器
                    pipeline.addLast(new ProtocolEncoder());
                    pipeline.addLast(new ProtocolDecoder());

                    // 4. 业务 Handler
                    pipeline.addLast(new HeartBeatHandler());
                    pipeline.addLast(businessGroup,
                        new BusinessLogicHandler());
                }
            });

        // ★ 绑定端口并启动
        ChannelFuture future = bootstrap.bind(8080).sync();
        this.serverChannel = future.channel();

        // ★ 注册到 Nacos 服务发现
        registerToNacos(8080);

        log.info("Netty Server started on port 8080, workers: {}", workerThreads);
    }

    @PreDestroy
    public void stop() {
        if (serverChannel != null) {
            serverChannel.close();
        }
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
        deregisterFromNacos();
        log.info("Netty Server stopped");
    }

    private void registerToNacos(int port) {
        // Nacos 服务注册
        Instance instance = new Instance();
        instance.setIp(nacosProperties.getIp());
        instance.setPort(port);
        instance.setServiceName("netty-server");
        instance.setClusterName("DEFAULT");
        instance.setMetadata(Map.of(
            "preserved.register.source", "SPRING_CLOUD",
            "protocol", "tcp",
            "max-connections", "1000000"
        ));
        try {
            nacosProperties.namingServiceInstance()
                .registerInstance("netty-server", instance);
        } catch (NacosException e) {
            log.error("Nacos register failed", e);
        }
    }

    // ★ 业务线程池（隔离 I/O 线程与业务线程）
    private final DefaultEventExecutorGroup businessGroup =
        new DefaultEventExecutorGroup(64,
            new ThreadFactoryBuilder()
                .setNameFormat("netty-biz-%d")
                .setDaemon(true)
                .build());
}

5.2 自定义协议：编解码器实现

/**
 * ★ 自定义协议消息结构
 * | 字段       | 类型   | 长度   | 说明           |
 * |-----------|--------|--------|---------------|
 * | length    | int    | 4B     | 消息总长度(含header) |
 * | version   | byte   | 1B     | 协议版本(0x01) |
 * | cmd       | byte   | 1B     | 命令类型       |
 * | seq       | short  | 2B     | 序列号         |
 * | payload   | byte[] | N      | protobuf/json  |
 */
@Data
public class ProtocolMessage {
    private int length;       // 自动计算
    private byte version = 0x01;
    private byte cmd;
    private short seq;
    private byte[] payload;

    // 命令类型枚举
    public static final byte CMD_HEARTBEAT = 0x00;
    public static final byte CMD_REQUEST   = 0x01;
    public static final byte CMD_RESPONSE  = 0x02;
    public static final byte CMD_NOTIFY    = 0x03;
}

/**
 * ★ 编码器: ProtocolMessage → ByteBuf
 */
@ChannelHandler.Sharable  // 无状态，可被多个 Channel 共享
public class ProtocolEncoder extends MessageToByteEncoder<ProtocolMessage> {

    @Override
    protected void encode(ChannelHandlerContext ctx,
                          ProtocolMessage msg, ByteBuf out) {
        byte[] payload = msg.getPayload();
        int length = 4 + 1 + 1 + 2 + (payload != null ? payload.length : 0);

        // ★ 写入长度（LengthFieldBasedFrameDecoder 依赖此字段拆包）
        out.writeInt(length);
        out.writeByte(msg.getVersion());
        out.writeByte(msg.getCmd());
        out.writeShort(msg.getSeq());

        if (payload != null) {
            out.writeBytes(payload);
        }

        // length 字段在此被写入，已经包含在上面的计算中
    }
}

/**
 * ★ 解码器: ByteBuf → ProtocolMessage
 */
@ChannelHandler.Sharable
public class ProtocolDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx,
                          ByteBuf in, List<Object> out) {
        // LengthFieldBasedFrameDecoder 已经处理了粘包
        // 到达这里时，in 中已经是一个完整的消息帧

        // 标记读位置（如果解析失败需要回滚）
        in.markReaderIndex();

        try {
            int length = in.readInt();
            byte version = in.readByte();

            // ★ 协议版本校验
            if (version != 0x01) {
                throw new ProtocolException(
                    "Unsupported version: " + version);
            }

            byte cmd = in.readByte();
            short seq = in.readShort();

            // 读取 payload
            int payloadLen = length - 4 - 1 - 1 - 2;
            byte[] payload = null;
            if (payloadLen > 0) {
                payload = new byte[payloadLen];
                in.readBytes(payload);
            }

            ProtocolMessage msg = new ProtocolMessage();
            msg.setLength(length);
            msg.setVersion(version);
            msg.setCmd(cmd);
            msg.setSeq(seq);
            msg.setPayload(payload);

            out.add(msg);
        } catch (Exception e) {
            in.resetReaderIndex();
            throw e;
        }
    }
}

5.3 心跳保活：IdleStateHandler + 心跳 Handler

/**
 * ★ 心跳 Handler: 配合 IdleStateHandler 实现双向心跳
 */
@ChannelHandler.Sharable
public class HeartBeatHandler extends ChannelDuplexHandler {

    private static final ProtocolMessage HEARTBEAT_MSG;
    static {
        HEARTBEAT_MSG = new ProtocolMessage();
        HEARTBEAT_MSG.setCmd(ProtocolMessage.CMD_HEARTBEAT);
        HEARTBEAT_MSG.setPayload(new byte[0]);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
            throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;

            switch (event.state()) {
                case READER_IDLE:
                    // ★ 读超时: 60秒未收到客户端任何数据
                    log.warn("Read idle timeout, channel: {}", ctx.channel());
                    // 再发一次心跳探测
                    ctx.writeAndFlush(HEARTBEAT_MSG.duplicate())
                        .addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                    break;

                case WRITER_IDLE:
                    // ★ 写超时: 30秒未向客户端发送数据
                    // 发送心跳包维持连接
                    ctx.writeAndFlush(HEARTBEAT_MSG.duplicate())
                        .addListener(future -> {
                            if (!future.isSuccess()) {
                                log.warn("Heartbeat send failed");
                            }
                        });
                    break;

                case ALL_IDLE:
                    // 双端都空闲，关闭连接
                    log.warn("All idle, closing channel: {}", ctx.channel());
                    ctx.close();
                    break;
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof ProtocolMessage) {
            ProtocolMessage message = (ProtocolMessage) msg;
            // ★ 收到心跳包，不传递给业务 Handler
            if (message.getCmd() == ProtocolMessage.CMD_HEARTBEAT) {
                return;
            }
        }
        // 非心跳包，继续传播
        ctx.fireChannelRead(msg);
    }
}

5.4 业务逻辑 Handler

/**
 * ★ 业务逻辑 Handler: 线程安全与背压控制
 */
@ChannelHandler.Sharable
public class BusinessLogicHandler
        extends SimpleChannelInboundHandler<ProtocolMessage> {

    @Autowired
    private RequestProcessor requestProcessor;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx,
                                  ProtocolMessage request) {
        // ★ 异步处理业务（不阻塞 I/O 线程）
        requestProcessor.process(request)
            .subscribeOn(Schedulers.boundedElastic())
            .subscribe(response -> {
                // 写回响应（自动在 EventLoop 线程执行）
                ctx.writeAndFlush(response);
            }, error -> {
                log.error("Process request failed", error);
                // 构造错误响应
                ProtocolMessage errorResponse = buildErrorResponse(
                    request.getSeq(), error.getMessage());
                ctx.writeAndFlush(errorResponse);
            });
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("Channel exception: {}", ctx.channel(), cause);
        ctx.close();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        log.info("Channel connected: {}", ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        log.info("Channel disconnected: {}", ctx.channel());
    }
}

5.5 Netty 客户端完整代码

/**
 * ★ Netty 客户端: 连接管理 + 自动重连 + 连接池
 */
@Component
@Slf4j
public class NettyClient {

    private EventLoopGroup workerGroup;
    private Bootstrap bootstrap;
    private volatile Channel channel;

    // ★ 重连参数
    private static final int MAX_RECONNECT_DELAY = 30;
    private volatile int reconnectDelay = 1;
    private final AtomicBoolean reconnecting = new AtomicBoolean(false);

    @PostConstruct
    public void init() {
        this.workerGroup = new NioEventLoopGroup(
            Runtime.getRuntime().availableProcessors(),
            new ThreadFactoryBuilder()
                .setNameFormat("netty-client-%d")
                .build());

        this.bootstrap = new Bootstrap()
            .group(workerGroup)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast(new IdleStateHandler(0, 25, 0,
                        TimeUnit.SECONDS));  // 25s 发一次心跳
                    p.addLast(new LengthFieldBasedFrameDecoder(
                        8 * 1024 * 1024, 0, 4, 8, 0, true));
                    p.addLast(new ProtocolEncoder());
                    p.addLast(new ProtocolDecoder());
                    p.addLast(new HeartBeatHandler());
                    p.addLast(new ClientBusinessHandler());
                    // ★ 重连 Handler（放在最后）
                    p.addLast(new ReconnectHandler(NettyClient.this));
                }
            });
    }

    public ChannelFuture connect(String host, int port) {
        return doConnect(host, port);
    }

    private ChannelFuture doConnect(String host, int port) {
        ChannelFuture future = bootstrap.connect(host, port);
        future.addListener((ChannelFutureListener) f -> {
            if (f.isSuccess()) {
                this.channel = f.channel();
                reconnectDelay = 1;  // 重置重连延迟
                log.info("Connected to {}:{}", host, port);
            } else {
                log.error("Connect to {}:{} failed", host, port);
                scheduleReconnect(host, port);
            }
        });
        return future;
    }

    // ★ 指数退避重连
    void scheduleReconnect(String host, int port) {
        if (reconnecting.compareAndSet(false, true)) {
            workerGroup.schedule(() -> {
                reconnecting.set(false);
                doConnect(host, port);
            }, reconnectDelay, TimeUnit.SECONDS);

            reconnectDelay = Math.min(reconnectDelay * 2, MAX_RECONNECT_DELAY);
            log.info("Reconnect scheduled in {}s", reconnectDelay);
        }
    }

    public Channel getChannel() {
        Channel ch = channel;
        if (ch != null && ch.isActive()) {
            return ch;
        }
        throw new IllegalStateException("Channel not connected");
    }

    @PreDestroy
    public void destroy() {
        if (channel != null) {
            channel.close();
        }
        workerGroup.shutdownGracefully();
    }
}

/**
 * ★ 客户端重连 Handler
 */
@ChannelHandler.Sharable
public class ReconnectHandler extends ChannelInboundHandlerAdapter {

    private final NettyClient client;
    private final String host;
    private final int port;

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        log.warn("Channel inactive, will reconnect");
        client.scheduleReconnect(host, port);
    }
}

5.6 大文件零拷贝传输

/**
 * ★ 大文件零拷贝传输: FileRegion
 * 适用场景: 视频流、日志文件、备份文件下载
 */
public class FileTransferHandler
        extends SimpleChannelInboundHandler<ProtocolMessage> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx,
                                  ProtocolMessage msg) {
        if (msg.getCmd() == ProtocolMessage.CMD_REQUEST) {
            String filePath = parseFilePath(msg);
            transferFile(ctx, filePath);
        }
    }

    private void transferFile(ChannelHandlerContext ctx, String filePath) {
        RandomAccessFile raf = null;
        try {
            File file = new File(filePath);
            raf = new RandomAccessFile(file, "r");

            // ★ FileRegion: 零拷贝传输
            // 底层调用 FileChannel.transferTo() → sendfile() 系统调用
            FileRegion region = new DefaultFileRegion(
                raf.getChannel(), 0, file.length());

            // 先发送文件头（元数据）
            ProtocolMessage header = buildFileHeader(file);
            ctx.write(header);

            // ★ 零拷贝传输文件内容
            ctx.writeAndFlush(region)
                .addListener(future -> {
                    // 关闭文件（无论成功与否）
                    if (raf != null) {
                        try { raf.close(); } catch (IOException ignored) {}
                    }
                    if (!future.isSuccess()) {
                        log.error("File transfer failed", future.cause());
                    }
                });

        } catch (Exception e) {
            log.error("File transfer error", e);
            if (raf != null) {
                try { raf.close(); } catch (IOException ignored) {}
            }
            ctx.writeAndFlush(buildErrorResponse(e.getMessage()));
        }
    }
}

5.7 百万级连接调优参数汇总

# application.yml (Spring Boot 2.x + Netty)
server:
  netty:
    # 这些参数适用于 Spring Boot 内嵌 Netty（WebFlux）
    connection-timeout: 2s

# Netty 服务端参数（编程式配置）
netty:
  server:
    port: 8080
    boss-threads: 1              # Boss 线程数
    worker-threads: 16           # Worker 线程数 = CPU * 2
    so-backlog: 8192            # 全连接队列
    so-reuseaddr: true
    tcp-nodelay: true
    so-keepalive: true
    so-rcvbuf: 65536
    so-sndbuf: 65536

    # ★ 内存池配置
    pooled-allocator:
      enabled: true
      # 每线程缓存大小 (-Dio.netty.allocator.cacheSize)
      cache-size: 512
      # Tiny 缓存大小 (-Dio.netty.allocator.tinyCacheSize)
      tiny-cache-size: 512
      # Small 缓存大小
      small-cache-size: 256
      # Normal 缓存大小
      normal-cache-size: 64
      # 是否使用直存 (堆外内存)
      prefer-direct: true

    # ★ 写入水位线
    write-buffer-water-mark:
      low: 32768        # 32KB
      high: 65536       # 64KB

    # ★ 心跳配置
    heartbeat:
      reader-idle: 60    # 读空闲超时(秒)
      writer-idle: 30    # 写空闲超时(秒)

    # ★ 连接管理
    max-connections: 1000000    # 最大连接数
    max-frame-length: 8388608   # 8MB 最大帧长度

# JVM 参数（百万连接必需）
# -Xms4g -Xmx4g
# -XX:MaxDirectMemorySize=2g
# -Dio.netty.maxDirectMemory=0  # 使用 JVM 的 MaxDirectMemorySize
# -Dio.netty.leakDetectionLevel=disabled  # 生产关闭泄漏检测
# -Dio.netty.recycler.maxCapacityPerThread=4096  # Recycler 容量
# -Dio.netty.allocator.numDirectArenas=16  # Direct Arena 数量
# -Dio.netty.allocator.numHeapArenas=16    # Heap Arena 数量
# -Dio.netty.noPreferDirect=false          # 优先使用堆外内存

# Linux 系统参数（/etc/sysctl.conf）
# net.core.somaxconn = 8192
# net.ipv4.tcp_max_syn_backlog = 8192
# net.ipv4.ip_local_port_range = 1024 65535
# net.ipv4.tcp_tw_reuse = 1
# net.ipv4.tcp_fin_timeout = 15
# net.core.netdev_max_backlog = 65536
# net.ipv4.tcp_keepalive_time = 60
# net.ipv4.tcp_keepalive_intvl = 10
# net.ipv4.tcp_keepalive_probes = 6
# fs.file-max = 2097152
# fs.nr_open = 2097152
# soft nofile 1048576
# hard nofile 1048576

---

六、ByteBuf 使用最佳实践

6.1 引用计数规则

/**
 * ★ ByteBuf 引用计数规则
 * 原则: 谁最后使用，谁负责 release()
 */
public class ByteBufBestPractice {

    // 1. 入站消息: 如果 Handler 不传递给下一个，必须 release
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf buf = (ByteBuf) msg;
        try {
            // 处理 buf...
            if (shouldForward) {
                // 传递给下一个 Handler → 不需要 release
                ctx.fireChannelRead(buf);
                buf = null;  // 不再持有引用
            }
            // 如果不传递，finally 会 release
        } finally {
            if (buf != null) {
                buf.release();
            }
        }
    }

    // 2. 出站消息: writeAndFlush 会自动 release
    public void sendMessage(ChannelHandlerContext ctx, byte[] data) {
        ByteBuf buf = ctx.alloc().buffer(data.length);  // refCnt = 1
        buf.writeBytes(data);
        ctx.writeAndFlush(buf);  // writeAndFlush 后会自动 release
        // 不要手动 buf.release()！
    }

    // 3. 使用 ReferenceCountUtil 安全释放
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // ... 释放可能未处理的消息
        ReferenceCountUtil.release(msg);
        ctx.close();
    }

    // 4. 使用 Unpooled 简化非性能敏感场景
    public ByteBuf createSimpleBuffer() {
        // 非池化，由 GC 回收，无需 release（但性能差）
        return Unpooled.wrappedBuffer("hello".getBytes());
    }
}

6.2 ByteBuf 类型选择

类型	创建方式	存储位置	自动释放	适用场景
Pooled Direct	ctx.alloc().buffer()	堆外	release()	高性能 I/O（默认）
Pooled Heap	ctx.alloc().heapBuffer()	堆	release()	需访问数组
Unpooled Direct	Unpooled.directBuffer()	堆外	GC	简单场景
Unpooled Heap	Unpooled.buffer()	堆	GC	测试/简单场景
Composite	Unpooled.wrappedBuffer(buf1, buf2)	混合	组件释放	零拷贝组合

---

七、Netty 内存泄漏检测

// 启动参数设置泄漏检测级别
// -Dio.netty.leakDetectionLevel=advanced

// 级别: DISABLED < SIMPLE < ADVANCED < PARANOID
// SIMPLE:   1% 采样，只报告泄漏（默认）
// ADVANCED: 1% 采样，报告访问记录
// PARANOID: 100% 采样，开发调试

// 泄漏报告示例:
// ERROR io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called
// Recent access records:
// #1: io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(...)
// Created at:
//   io.netty.buffer.PoolArena.allocate(PoolArena.java:200)
//   com.example.handler.MyHandler.channelRead(MyHandler.java:45)

---

八、Netty 与 Spring Boot 集成

/**
 * ★ Spring Boot 集成 Netty 服务端
 * 使用 @PostConstruct 启动，@PreDestroy 优雅关闭
 */
@Configuration
public class NettyServerConfig {

    @Bean(destroyMethod = "stop")
    public NettyServer nettyServer() {
        return new NettyServer();
    }
}

/**
 * ★ Spring 事件集成: 监听应用状态
 */
@Component
public class NettyLifecycleListener {

    @Autowired
    private NettyServer nettyServer;

    @EventListener(ApplicationReadyEvent.class)
    public void onApplicationReady() {
        // 应用启动完成后启动 Netty
        // （如果 NettyServer 使用 @PostConstruct 则不需要）
    }

    @EventListener(ContextClosedEvent.class)
    public void onContextClosed() {
        // Spring 上下文关闭时触发 Netty 优雅关闭
        nettyServer.stop();
    }
}

/**
 * ★ 使用 Nacos 作为服务注册中心
 * Netty TCP 服务注册到 Nacos
 */
@Component
public class NettyNacosRegistrar {

    @Autowired
    private NacosDiscoveryProperties nacosProperties;

    @Autowired
    private NettyServerProperties serverProperties;

    public void register() throws NacosException {
        NamingService namingService =
            nacosProperties.namingServiceInstance();

        Instance instance = new Instance();
        instance.setIp(serverProperties.getIp());
        instance.setPort(serverProperties.getPort());
        instance.setServiceName(serverProperties.getServiceName());
        instance.setClusterName("DEFAULT");
        instance.setWeight(1.0);
        instance.setMetadata(Map.of(
            "preserved.register.source", "SPRING_CLOUD",
            "protocol", "tcp",
            "version", "1.0.0",
            "max-connections", String.valueOf(serverProperties.getMaxConnections()),
            "heartbeat-interval", String.valueOf(serverProperties.getHeartbeatInterval())
        ));

        namingService.registerInstance(
            serverProperties.getServiceName(),
            nacosProperties.getGroup(),
            instance);

        // 启动心跳维持
        startHeartbeat(namingService, instance);
    }

    private void startHeartbeat(NamingService namingService, Instance instance) {
        Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "nacos-heartbeat");
            t.setDaemon(true);
            return t;
        }).scheduleAtFixedRate(() -> {
            try {
                namingService.registerInstance(
                    instance.getServiceName(),
                    nacosProperties.getGroup(),
                    instance);
            } catch (NacosException e) {
                log.error("Nacos heartbeat failed", e);
            }
        }, 5, 5, TimeUnit.SECONDS);
    }
}

---

九、性能调优与监控

9.1 关键性能指标

指标	健康阈值	说明
连接数	目标承载量	ChannelGroup.size()
内存池使用率	< 80%	PooledByteBufAllocator.DEFAULT.metric()
EventLoop 任务队列	< 1000	eventLoop.pendingTasks()
直接内存使用	< MaxDirectMemorySize	PlatformDependent.usedDirectMemory()
写入缓冲区水位	正常	超过高水位会停止读取

9.2 内存池监控

@Component
public class NettyMetricsCollector {

    @Scheduled(fixedRate = 60000)
    public void reportMetrics() {
        PooledByteBufAllocatorMetric metric =
            PooledByteBufAllocator.DEFAULT.metric();

        // Direct 内存池
        for (PoolArenaMetric arena : metric.directArenas()) {
            log.info("Direct Arena: numAllocations={}, numDeallocations={}, " +
                    "numSmallAllocations={}, numNormalAllocations={}, " +
                    "numHugeAllocations={}, chunkLists={}",
                arena.numAllocations(),
                arena.numDeallocations(),
                arena.numSmallAllocations(),
                arena.numNormalAllocations(),
                arena.numHugeAllocations(),
                arena.chunkLists().size());
        }

        // 线程本地缓存
        for (PoolThreadCacheMetric cache : metric.threadCaches()) {
            log.debug("ThreadCache: tiny={}, small={}, normal={}",
                cache.tinySubPagesSize(),
                cache.smallSubPagesSize(),
                cache.normalAllocationsSize());
        }
    }
}

---

十、总结

维度	Java NIO	Netty
线程模型	手动管理 Selector 线程	EventLoop 自动管理
内存管理	ByteBuffer（固定大小）	ByteBuf（可扩容 + 内存池）
粘包拆包	手动实现	LengthFieldBasedFrameDecoder 等
编解码	手动处理	ChannelCodec 组合
连接管理	手动处理	ChannelGroup 统一管理
心跳检测	手动 Timer	IdleStateHandler 内置
性能	中等	极高（内存池 + 零拷贝）
学习曲线	高	中

---