Cao Note

外观

Netty ​

高性能网络编程框架，异步事件驱动的 NIO 框架。Netty 由 JBoss 提供，现为独立开源项目，广泛应用于互联网、大数据、游戏、通信等领域。

一、Netty 概述 ​

1.1 什么是 Netty ​

Netty 是一个异步、事件驱动的网络编程框架，它基于 Java NIO 构建，极大地简化了 TCP、UDP、HTTP、WebSocket 等协议的开发工作。

核心特性：

  • 异步非阻塞：所有 I/O 操作都是异步的，通过 Future / Callback 机制获取结果。
  • 事件驱动：将网络事件（连接建立、数据读取、异常发生等）抽象为事件，交由 Handler 处理。
  • 高性能：基于 Reactor 模型、零拷贝、池化内存等机制实现高吞吐、低延迟。
  • 易用性：API 简洁，开箱即用的编解码器，屏蔽了 Java NIO 的复杂性。

1.2 Netty vs Java NIO ​

对比维度Java NIONetty
API 复杂度直接操作 ByteBuffer、Selector、Channel，代码量大高度抽象，Handler 链式处理，代码简洁
开发效率需要手动处理粘包拆包、断线重连等内置各种编解码器和工具类
性能基础 NIO，无池化、零拷贝等优化池化内存、零拷贝、I/O 多路复用优化
可靠性需要自行处理异常重连等支持空闲检测、写缓冲区水位、流量控制
扩展性需自行实现线程模型灵活的事件循环组和 Handler 编排
社区生态无Dubbo、Spring Cloud Gateway、Elasticsearch、RocketMQ 等均基于 Netty

1.3 使用场景 ​

  • RPC 框架底层通信：Dubbo、gRPC、Thrift 的传输层
  • API 网关：Spring Cloud Gateway、Zuul 2.x
  • 即时通讯（IM）：微信、钉钉等服务端消息推送
  • 游戏服务器：MMORPG、棋牌游戏的高并发连接管理
  • 消息推送系统：WebSocket 推送、SSE
  • 大数据组件：HBase、Flink、Spark 的部分通信模块
  • 消息队列：RocketMQ、Kafka 的部分网络层

二、Reactor 模式 ​

2.1 Reactor 为什么是 Netty 的基础 ​

Reactor 是一种事件驱动的设计模式，它将 I/O 事件分发到对应的处理器。Netty 的线程模型正是基于 Reactor 模式的变体实现的。

Reactor 模式的核心角色：

  • Reactor：负责监听和分发事件
  • Acceptor：处理新的连接请求
  • Handler：处理具体的业务逻辑

2.2 单线程 Reactor（单 Reactor 单线程） ​

所有 I/O 操作（包括 accept、read、write）都在同一个线程中完成。

Reactor (单线程)
   ├── accept 连接
   ├── read 请求
   ├── 业务处理
   └── write 响应
Reactor (单线程)
   ├── accept 连接
   ├── read 请求
   ├── 业务处理
   └── write 响应

优点：实现简单，不存在并发问题。

缺点：单线程处理所有事件，若某个 Handler 阻塞，将影响整个系统的吞吐量。无法充分利用多核 CPU。

java
// 单线程 Reactor 示意
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
// 实际 Netty 不会这么用，仅作概念演示
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup) // 只有一个 EventLoopGroup
 .channel(NioServerSocketChannel.class)
 .childHandler(new ChannelInitializer<SocketChannel>() {
     @Override
     protected void initChannel(SocketChannel ch) {
         ch.pipeline().addLast(new EchoServerHandler());
     }
 });
// 单线程 Reactor 示意
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
// 实际 Netty 不会这么用，仅作概念演示
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup) // 只有一个 EventLoopGroup
 .channel(NioServerSocketChannel.class)
 .childHandler(new ChannelInitializer<SocketChannel>() {
     @Override
     protected void initChannel(SocketChannel ch) {
         ch.pipeline().addLast(new EchoServerHandler());
     }
 });

2.3 多线程 Reactor（单 Reactor 线程池） ​

Reactor 线程只负责 I/O 事件的监听和分发，业务处理交由 worker 线程池完成。

Reactor (单线程)
   ├── accept 连接
   ├── read 请求  ──→ Worker 线程池（业务处理）
   ├── write 响应  ←── Worker 线程池
   └── ...
Reactor (单线程)
   ├── accept 连接
   ├── read 请求  ──→ Worker 线程池（业务处理）
   ├── write 响应  ←── Worker 线程池
   └── ...

优点：将 I/O 与业务逻辑分离，业务处理可并行执行。

缺点：Reactor 单线程处理所有 I/O 事件，若并发连接数巨大，Reactor 线程可能成为瓶颈。

2.4 主从 Reactor（多 Reactor 多线程） ​

引入多个 Reactor：

  • Main Reactor：只负责处理新的连接请求（accept）
  • Sub Reactor：负责已连接 Channel 的读写事件
Main Reactor（accept）
   │
   ├── Sub Reactor 1（read / write）
   ├── Sub Reactor 2（read / write）
   └── Sub Reactor 3（read / write）
Main Reactor（accept）
   │
   ├── Sub Reactor 1（read / write）
   ├── Sub Reactor 2（read / write）
   └── Sub Reactor 3（read / write）

优点：主从分离，accept 与 read/write 不互相阻塞；Sub Reactor 可线性扩展，充分利用多核。

Netty 的标准线程模型正是主从 Reactor。

2.5 Netty 的 Reactor 实现 ​

java
// Netty 主从 Reactor 模型 —— 服务端标准写法
EventLoopGroup bossGroup = new NioEventLoopGroup(1);       // Main Reactor
EventLoopGroup workerGroup = new NioEventLoopGroup();       // Sub Reactor（默认 CPU 核心数 * 2）

ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
         .channel(NioServerSocketChannel.class)
         .option(ChannelOption.SO_BACKLOG, 128)
         .childOption(ChannelOption.SO_KEEPALIVE, true)
         .childHandler(new ChannelInitializer<SocketChannel>() {
             @Override
             protected void initChannel(SocketChannel ch) {
                 ch.pipeline().addLast(new MyServerHandler());
             }
         });

ChannelFuture future = bootstrap.bind(8080).sync();
future.channel().closeFuture().sync();
// Netty 主从 Reactor 模型 —— 服务端标准写法
EventLoopGroup bossGroup = new NioEventLoopGroup(1);       // Main Reactor
EventLoopGroup workerGroup = new NioEventLoopGroup();       // Sub Reactor（默认 CPU 核心数 * 2）

ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
         .channel(NioServerSocketChannel.class)
         .option(ChannelOption.SO_BACKLOG, 128)
         .childOption(ChannelOption.SO_KEEPALIVE, true)
         .childHandler(new ChannelInitializer<SocketChannel>() {
             @Override
             protected void initChannel(SocketChannel ch) {
                 ch.pipeline().addLast(new MyServerHandler());
             }
         });

ChannelFuture future = bootstrap.bind(8080).sync();
future.channel().closeFuture().sync();
  • bossGroup：处理连接接入（Main Reactor），通常线程数为 1
  • workerGroup：处理读写事件（Sub Reactor），线程数默认为 CPU核心数 * 2
  • ServerBootstrap：服务端启动引导类

三、核心组件 ​

3.1 Bootstrap / ServerBootstrap ​

引导类用途客户端/服务端
Bootstrap连接远程主机，启动客户端客户端
ServerBootstrap绑定本地端口，监听连接服务端

ServerBootstrap 常用方法：

方法说明
group(boss, worker)设置 EventLoopGroup
channel(Class)指定 Channel 类型
option(ChannelOption, Object)配置 Channel 参数（服务端 Channel）
childOption(ChannelOption, Object)配置客户端 Channel 参数
handler(ChannelHandler)处理 bossGroup 的事件
childHandler(ChannelHandler)处理 workerGroup 的事件
bind(int)绑定端口

3.2 EventLoopGroup / EventLoop ​

EventLoopGroup 是一个线程池，管理多个 EventLoop 实例。

  • 一个 EventLoop 对应一个线程，负责多个 Channel 的 I/O 事件
  • EventLoop 与 Channel 是 多对一 的关系：一个 EventLoop 可管理多个 Channel
  • EventLoop 内部维护着一个 Selector，用于监听 Channel 的状态变化
java
// 常用 EventLoopGroup 实现
EventLoopGroup group = new NioEventLoopGroup();        // NIO 模式
EventLoopGroup group = new EpollEventLoopGroup();      // Linux epoll（性能更高）
EventLoopGroup group = new OioEventLoopGroup();        // OIO 阻塞模式（已过时）
// 常用 EventLoopGroup 实现
EventLoopGroup group = new NioEventLoopGroup();        // NIO 模式
EventLoopGroup group = new EpollEventLoopGroup();      // Linux epoll（性能更高）
EventLoopGroup group = new OioEventLoopGroup();        // OIO 阻塞模式（已过时）

NioEventLoopGroup 构造函数参数说明：

java
public NioEventLoopGroup() {
    this(0); // 0 表示使用默认值：CPU核心数 * 2
}

public NioEventLoopGroup(int nThreads) {
    // 指定线程数
}
public NioEventLoopGroup() {
    this(0); // 0 表示使用默认值：CPU核心数 * 2
}

public NioEventLoopGroup(int nThreads) {
    // 指定线程数
}

3.3 Channel（NioSocketChannel / NioServerSocketChannel） ​

Channel 类型说明
NioServerSocketChannel服务端 Channel，监听新连接
NioSocketChannel客户端 Channel，用于数据传输
OioServerSocketChannelOIO 服务端（阻塞模式）
OioSocketChannelOIO 客户端（阻塞模式）
EpollServerSocketChannelLinux epoll 服务端
EpollSocketChannelLinux epoll 客户端

Channel 的生命周期状态：

未注册 → 已注册 → 活跃 → 非活跃 → 关闭
未注册 → 已注册 → 活跃 → 非活跃 → 关闭

对应事件：channelRegistered → channelActive → channelInactive → channelUnregistered

3.4 ChannelPipeline / ChannelHandlerContext ​

ChannelPipeline 是一个 Handler 链，负责管理 ChannelHandler 的执行顺序。

  • pipeline.addLast(handler)：在链尾添加 Handler
  • pipeline.addFirst(handler)：在链头添加 Handler
  • pipeline.remove(handler)：移除 Handler

ChannelHandlerContext 代表 Pipeline 中的一个节点，用于在 Handler 间传递数据：

java
// 通过 ChannelHandlerContext 转发事件
ctx.fireChannelRead(msg);     // 将消息传递给下一个 Inbound Handler
ctx.write(msg);                // 将消息写入下一个 Outbound Handler
ctx.channel().write(msg);     // 从 Pipeline 尾部开始寻找 Outbound Handler
// 通过 ChannelHandlerContext 转发事件
ctx.fireChannelRead(msg);     // 将消息传递给下一个 Inbound Handler
ctx.write(msg);                // 将消息写入下一个 Outbound Handler
ctx.channel().write(msg);     // 从 Pipeline 尾部开始寻找 Outbound Handler

3.5 ChannelHandler（Inbound / Outbound） ​

类型接口处理方向
Inbound HandlerChannelInboundHandler处理入站数据（数据从网络 → 应用）
Outbound HandlerChannelOutboundHandler处理出站数据（数据从应用 → 网络）

典型 Inbound 方法调用顺序：

channelRegistered → channelActive → channelRead → channelReadComplete
→ exceptionCaught → channelInactive → channelUnregistered
channelRegistered → channelActive → channelRead → channelReadComplete
→ exceptionCaught → channelInactive → channelUnregistered

典型 Outbound 方法调用顺序：

bind → connect → disconnect → close → deregister → read → write → flush
bind → connect → disconnect → close → deregister → read → write → flush

3.6 ChannelFuture（异步回调） ​

Netty 所有 I/O 操作都是异步的，返回 ChannelFuture 对象。

java
// 方式一：同步等待
ChannelFuture future = bootstrap.bind(8080).sync();
future.channel().closeFuture().sync();

// 方式二：异步回调（推荐）
ChannelFuture future = bootstrap.bind(8080);
future.addListener((ChannelFutureListener) f -> {
    if (f.isSuccess()) {
        System.out.println("绑定成功");
    } else {
        System.err.println("绑定失败");
        f.cause().printStackTrace();
    }
});
// 方式一：同步等待
ChannelFuture future = bootstrap.bind(8080).sync();
future.channel().closeFuture().sync();

// 方式二：异步回调（推荐）
ChannelFuture future = bootstrap.bind(8080);
future.addListener((ChannelFutureListener) f -> {
    if (f.isSuccess()) {
        System.out.println("绑定成功");
    } else {
        System.err.println("绑定失败");
        f.cause().printStackTrace();
    }
});

四、ByteBuf ​

ByteBuf 是 Netty 自带的字节缓冲区，替代 Java NIO 的 ByteBuffer。

4.1 ByteBuf 与 ByteBuffer 的对比 ​

特性Java NIO ByteBufferNetty ByteBuf
指针一个 position 指针readerIndex + writerIndex 双指针
扩容不支持，溢出抛出异常自动扩容
池化不支持支持池化（PooledByteBufAllocator）
零拷贝不支持支持 CompositeByteBuf、slice、duplicate
API 易用性需 flip/rewind/compact直观的读写方法

4.2 堆内 / 堆外（Direct）缓冲区 ​

java
// 堆内缓冲区（HeapByteBuf）
// 优点：分配和回收快，由 GC 管理
// 缺点：Socket I/O 时需拷贝到 Direct 内存
ByteBuf heapBuf = Unpooled.buffer(128);

// 堆外缓冲区（DirectByteBuf）
// 优点：Socket I/O 效率高（零拷贝）
// 缺点：分配/回收成本高，需手动释放
ByteBuf directBuf = Unpooled.directBuffer(128);
// 堆内缓冲区（HeapByteBuf）
// 优点：分配和回收快，由 GC 管理
// 缺点：Socket I/O 时需拷贝到 Direct 内存
ByteBuf heapBuf = Unpooled.buffer(128);

// 堆外缓冲区（DirectByteBuf）
// 优点：Socket I/O 效率高（零拷贝）
// 缺点：分配/回收成本高，需手动释放
ByteBuf directBuf = Unpooled.directBuffer(128);

4.3 池化（PooledByteBufAllocator） ​

Netty 4.1 默认启用池化。池化分配器通过内存池复用 ByteBuf，减少 GC 压力。

java
// 显示指定池化/非池化
ByteBufAllocator allocator;

// 池化（推荐）
allocator = PooledByteBufAllocator.DEFAULT;

// 非池化
allocator = UnpooledByteBufAllocator.DEFAULT;

// 分配 ByteBuf
ByteBuf buf = allocator.buffer(256);
// 显示指定池化/非池化
ByteBufAllocator allocator;

// 池化（推荐）
allocator = PooledByteBufAllocator.DEFAULT;

// 非池化
allocator = UnpooledByteBufAllocator.DEFAULT;

// 分配 ByteBuf
ByteBuf buf = allocator.buffer(256);

4.4 引用计数（ReferenceCountUtil） ​

ByteBuf 使用引用计数管理内存生命周期：

java
ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(128);

// 增加引用
buf.retain();

// 减少引用，计数为 0 时释放内存
buf.release();

// 工具类：安全释放
ReferenceCountUtil.safeRelease(buf);

// 判断是否只有一个引用
boolean isUnique = buf.refCnt() == 1;
ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(128);

// 增加引用
buf.retain();

// 减少引用，计数为 0 时释放内存
buf.release();

// 工具类：安全释放
ReferenceCountUtil.safeRelease(buf);

// 判断是否只有一个引用
boolean isUnique = buf.refCnt() == 1;

4.5 ByteBuf 读/写操作 ​

      +-------------------+------------------+------------------+
      |   丢弃部分         |    可读数据       |    可写空间      |
      +-------------------+------------------+------------------+
      |                   |                  |                  |
      0             readerIndex          writerIndex         capacity
      +-------------------+------------------+------------------+
      |   丢弃部分         |    可读数据       |    可写空间      |
      +-------------------+------------------+------------------+
      |                   |                  |                  |
      0             readerIndex          writerIndex         capacity

写操作（增加 writerIndex）：

java
ByteBuf buf = Unpooled.buffer(10);
buf.writeByte(1);              // 写一个字节
buf.writeShort(100);           // 写两个字节
buf.writeInt(10000);           // 写四个字节
buf.writeLong(100000L);        // 写八个字节
buf.writeBytes("Hello".getBytes()); // 写入字节数组
ByteBuf buf = Unpooled.buffer(10);
buf.writeByte(1);              // 写一个字节
buf.writeShort(100);           // 写两个字节
buf.writeInt(10000);           // 写四个字节
buf.writeLong(100000L);        // 写八个字节
buf.writeBytes("Hello".getBytes()); // 写入字节数组

读操作（增加 readerIndex）：

java
byte b = buf.readByte();       // 读一个字节
short s = buf.readShort();     // 读两个字节
int i = buf.readInt();         // 读四个字节
long l = buf.readLong();       // 读八个字节
byte[] dst = new byte[5];
buf.readBytes(dst);            // 读入字节数组
byte b = buf.readByte();       // 读一个字节
short s = buf.readShort();     // 读两个字节
int i = buf.readInt();         // 读四个字节
long l = buf.readLong();       // 读八个字节
byte[] dst = new byte[5];
buf.readBytes(dst);            // 读入字节数组

其他操作：

java
// 标记和重置
buf.markReaderIndex();
buf.markWriterIndex();
buf.resetReaderIndex();
buf.resetWriterIndex();

// discardReadBytes：释放已读空间，压缩缓冲区
buf.discardReadBytes();

// clear：重置 readerIndex 和 writerIndex
buf.clear();

// 查找分隔符
int index = buf.indexOf(buf.readerIndex(), buf.writerIndex(), (byte) '\n');

// 可读/可写字节数
int readableBytes = buf.readableBytes();
int writableBytes = buf.writableBytes();
// 标记和重置
buf.markReaderIndex();
buf.markWriterIndex();
buf.resetReaderIndex();
buf.resetWriterIndex();

// discardReadBytes：释放已读空间，压缩缓冲区
buf.discardReadBytes();

// clear：重置 readerIndex 和 writerIndex
buf.clear();

// 查找分隔符
int index = buf.indexOf(buf.readerIndex(), buf.writerIndex(), (byte) '\n');

// 可读/可写字节数
int readableBytes = buf.readableBytes();
int writableBytes = buf.writableBytes();

五、ChannelHandler（重点） ​

5.1 ChannelInboundHandler / ChannelOutboundHandler ​

InboundHandler：处理入站事件

java
public class MyInboundHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) {
        System.out.println("Channel 注册到 EventLoop");
        ctx.fireChannelRegistered();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("Channel 已激活（连接建立）");
        ctx.fireChannelActive();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("收到数据：" + msg);
        // 处理业务后，传递给下一个 Handler
        ctx.fireChannelRead(msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        System.out.println("读取完成");
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        System.out.println("连接关闭");
        ctx.fireChannelInactive();
    }
}
public class MyInboundHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) {
        System.out.println("Channel 注册到 EventLoop");
        ctx.fireChannelRegistered();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("Channel 已激活（连接建立）");
        ctx.fireChannelActive();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("收到数据：" + msg);
        // 处理业务后，传递给下一个 Handler
        ctx.fireChannelRead(msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        System.out.println("读取完成");
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        System.out.println("连接关闭");
        ctx.fireChannelInactive();
    }
}

OutboundHandler：处理出站事件

java
public class MyOutboundHandler extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        System.out.println("写出数据：" + msg);
        ctx.write(msg, promise);
    }

    @Override
    public void flush(ChannelHandlerContext ctx) {
        System.out.println("刷新缓冲区");
        ctx.flush();
    }

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
        System.out.println("关闭连接");
        ctx.close(promise);
    }
}
public class MyOutboundHandler extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        System.out.println("写出数据：" + msg);
        ctx.write(msg, promise);
    }

    @Override
    public void flush(ChannelHandlerContext ctx) {
        System.out.println("刷新缓冲区");
        ctx.flush();
    }

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
        System.out.println("关闭连接");
        ctx.close(promise);
    }
}

5.2 SimpleChannelInboundHandler（自动释放） ​

SimpleChannelInboundHandler 会在 channelRead 结束后自动释放消息（调用 ReferenceCountUtil.release），避免内存泄漏。

java
public class AutoReleaseHandler extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        // 此处无需手动 release
        System.out.println("收到消息：" + msg);
        ctx.channel().writeAndFlush("已收到：" + msg);
    }
}
public class AutoReleaseHandler extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        // 此处无需手动 release
        System.out.println("收到消息：" + msg);
        ctx.channel().writeAndFlush("已收到：" + msg);
    }
}

对比 ChannelInboundHandlerAdapter，它不会自动释放消息：

java
public class ManualReleaseHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            // 业务处理
            System.out.println("收到消息：" + msg);
        } finally {
            // 必须手动释放
            ReferenceCountUtil.release(msg);
        }
    }
}
public class ManualReleaseHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            // 业务处理
            System.out.println("收到消息：" + msg);
        } finally {
            // 必须手动释放
            ReferenceCountUtil.release(msg);
        }
    }
}

5.3 Handler 执行顺序 ​

Pipeline 中 Handler 的组织是一个双向链表：

                    ┌───────┐    ┌───────┐    ┌───────┐
入站 (Inbound)  →   │ H1(In)│ → │ H2(In)│ → │ H3(In)│ →  ...
                    └───────┘    └───────┘    └───────┘
                          ↑
                    ┌───────┐    ┌───────┐    ┌───────┐
出站 (Outbound) ←   │ H3(Out)│ ← │ H2(Out)│ ← │ H1(Out)│
                    └───────┘    └───────┘    └───────┘
                    ┌───────┐    ┌───────┐    ┌───────┐
入站 (Inbound)  →   │ H1(In)│ → │ H2(In)│ → │ H3(In)│ →  ...
                    └───────┘    └───────┘    └───────┘
                          ↑
                    ┌───────┐    ┌───────┐    ┌───────┐
出站 (Outbound) ←   │ H3(Out)│ ← │ H2(Out)│ ← │ H1(Out)│
                    └───────┘    └───────┘    └───────┘
  • Inbound Handler：按添加顺序正序执行（addFirst 在前，addLast 在后）
  • Outbound Handler：按添加顺序倒序执行（addLast 在前，addFirst 在后）
java
// 添加顺序：H1 → H2 → H3（Inbound）
//          H1 → H2 → H3（Outbound）
pipeline.addLast("H1", new InboundHandler1());
pipeline.addLast("H2", new InboundHandler2());
pipeline.addLast("H3", new InboundHandler3());

pipeline.addLast("H1", new OutboundHandler1());
pipeline.addLast("H2", new OutboundHandler2());
pipeline.addLast("H3", new OutboundHandler3());

// Inbound 执行顺序: H1 → H2 → H3
// Outbound 执行顺序: H3 → H2 → H1
// 添加顺序：H1 → H2 → H3（Inbound）
//          H1 → H2 → H3（Outbound）
pipeline.addLast("H1", new InboundHandler1());
pipeline.addLast("H2", new InboundHandler2());
pipeline.addLast("H3", new InboundHandler3());

pipeline.addLast("H1", new OutboundHandler1());
pipeline.addLast("H2", new OutboundHandler2());
pipeline.addLast("H3", new OutboundHandler3());

// Inbound 执行顺序: H1 → H2 → H3
// Outbound 执行顺序: H3 → H2 → H1

5.4 异常处理（exceptionCaught） ​

异常在 Pipeline 中传播：

java
// 在任意 Handler 中抛出异常
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    throw new RuntimeException("业务异常");
}
// 在任意 Handler 中抛出异常
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    throw new RuntimeException("业务异常");
}

异常会跳过当前 Handler 之后的所有同类型 Handler，沿 Pipeline 反向传播：

Inbound 异常传播方向：
H1 → H2 → H3（抛出异常）
     ↑
   异常往回传播
     ↑
H1.exceptionCaught ← H2.exceptionCaught ← 异常源头 H3
Inbound 异常传播方向：
H1 → H2 → H3（抛出异常）
     ↑
   异常往回传播
     ↑
H1.exceptionCaught ← H2.exceptionCaught ← 异常源头 H3

推荐在 Pipeline 尾部添加异常处理 Handler：

java
public class ExceptionHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 记录日志
        log.error("连接异常，关闭 Channel：{}", ctx.channel().remoteAddress(), cause);

        // 关闭连接
        ctx.close();
    }
}
public class ExceptionHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 记录日志
        log.error("连接异常，关闭 Channel：{}", ctx.channel().remoteAddress(), cause);

        // 关闭连接
        ctx.close();
    }
}

六、编解码器 ​

6.1 ByteToMessageDecoder / MessageToByteEncoder ​

解码器：将字节流转换为业务对象

java
public class MyDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 从 ByteBuf 中读取数据，解析为业务对象
        if (in.readableBytes() < 4) {
            return; // 数据不足，等待下次
        }
        int length = in.readInt();
        byte[] content = new byte[length];
        in.readBytes(content);
        out.add(new MyMessage(content)); // 将解析结果添加到 out 列表
    }
}
public class MyDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 从 ByteBuf 中读取数据，解析为业务对象
        if (in.readableBytes() < 4) {
            return; // 数据不足，等待下次
        }
        int length = in.readInt();
        byte[] content = new byte[length];
        in.readBytes(content);
        out.add(new MyMessage(content)); // 将解析结果添加到 out 列表
    }
}

编码器：将业务对象转换为字节流

java
public class MyEncoder extends MessageToByteEncoder<MyMessage> {

    @Override
    protected void encode(ChannelHandlerContext ctx, MyMessage msg, ByteBuf out) {
        // 将业务对象写入 ByteBuf
        out.writeInt(msg.getContent().length);
        out.writeBytes(msg.getContent());
    }
}
public class MyEncoder extends MessageToByteEncoder<MyMessage> {

    @Override
    protected void encode(ChannelHandlerContext ctx, MyMessage msg, ByteBuf out) {
        // 将业务对象写入 ByteBuf
        out.writeInt(msg.getContent().length);
        out.writeBytes(msg.getContent());
    }
}

6.2 ReplayingDecoder ​

ReplayingDecoder 是 ByteToMessageDecoder 的便捷子类，无需手动检查可读字节数：

java
public class MyReplayingDecoder extends ReplayingDecoder<Void> {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 无需检查 readableBytes，若数据不足将自动抛出异常并等待
        int length = in.readInt();
        byte[] content = new byte[length];
        in.readBytes(content);
        out.add(new MyMessage(content));
    }
}
public class MyReplayingDecoder extends ReplayingDecoder<Void> {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 无需检查 readableBytes，若数据不足将自动抛出异常并等待
        int length = in.readInt();
        byte[] content = new byte[length];
        in.readBytes(content);
        out.add(new MyMessage(content));
    }
}

注意：ReplayingDecoder 内部通过异常机制处理数据不足的情况，性能略低于 ByteToMessageDecoder。

6.3 内置编解码器 ​

6.3.1 LineBasedFrameDecoder（行分割） ​

以换行符（\n 或 \r\n）作为消息分隔符：

java
pipeline.addLast(new LineBasedFrameDecoder(1024));  // 最大行长度 1024
pipeline.addLast(new StringDecoder());               // 转换为字符串
pipeline.addLast(new MyBusinessHandler());
pipeline.addLast(new LineBasedFrameDecoder(1024));  // 最大行长度 1024
pipeline.addLast(new StringDecoder());               // 转换为字符串
pipeline.addLast(new MyBusinessHandler());

6.3.2 DelimiterBasedFrameDecoder（自定义分隔符） ​

java
ByteBuf delimiter = Unpooled.copiedBuffer("$$".getBytes());
pipeline.addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
pipeline.addLast(new StringDecoder());
pipeline.addLast(new MyBusinessHandler());
ByteBuf delimiter = Unpooled.copiedBuffer("$$".getBytes());
pipeline.addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
pipeline.addLast(new StringDecoder());
pipeline.addLast(new MyBusinessHandler());

6.3.3 FixedLengthFrameDecoder（定长消息） ​

java
// 每条消息固定 100 字节
pipeline.addLast(new FixedLengthFrameDecoder(100));
pipeline.addLast(new MyBusinessHandler());
// 每条消息固定 100 字节
pipeline.addLast(new FixedLengthFrameDecoder(100));
pipeline.addLast(new MyBusinessHandler());

6.3.4 LengthFieldBasedFrameDecoder（长度字段） ​

最灵活的协议解码器，通过消息头中的长度字段确定消息体：

java
// 参数：maxFrameLength, lengthFieldOffset, lengthFieldLength
//       lengthAdjustment, initialBytesToStrip
pipeline.addLast(new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 4));
// 参数：maxFrameLength, lengthFieldOffset, lengthFieldLength
//       lengthAdjustment, initialBytesToStrip
pipeline.addLast(new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 4));

协议格式示例：

+--------+----------------+
| 长度(4) |    消息体      |
+--------+----------------+
+--------+----------------+
| 长度(4) |    消息体      |
+--------+----------------+

常见参数组合：

协议格式参数
长度(2) + 消息体new LengthFieldBasedFrameDecoder(1024, 0, 2, 0, 2)
长度(4) + 消息体new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 4)
固定头(4) + 长度(2) + 消息体new LengthFieldBasedFrameDecoder(1024, 4, 2, 0, 6)
长度(4) + 消息体，保留长度字段new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 0)

6.4 TCP 粘包 / 拆包问题及解决方案 ​

问题现象：

  • 粘包：发送方发送 ABC 和 DEF，接收方收到 ABCDEF
  • 拆包：发送方发送 ABCDEF，接收方收到 ABC 和 DEF

原因：TCP 是流式协议，没有消息边界。

Netty 解决方案 —— 使用编解码器划定消息边界：

方案适用场景推荐
LineBasedFrameDecoder文本协议，行分隔简单
DelimiterBasedFrameDecoder自定义分隔符灵活
FixedLengthFrameDecoder固定长度消息简单
LengthFieldBasedFrameDecoder通用二进制协议最常用
java
// 解决粘包/拆包的完整 Pipeline
ChannelPipeline pipeline = ch.pipeline();

// 方案一：行分割（文本协议）
pipeline.addLast(new LineBasedFrameDecoder(8192));
pipeline.addLast(new StringDecoder());
pipeline.addLast(new BusinessHandler());

// 方案二：长度字段（二进制协议，推荐）
pipeline.addLast(new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 4));
pipeline.addLast(new ByteArrayDecoder());
pipeline.addLast(new BusinessHandler());
// 解决粘包/拆包的完整 Pipeline
ChannelPipeline pipeline = ch.pipeline();

// 方案一：行分割（文本协议）
pipeline.addLast(new LineBasedFrameDecoder(8192));
pipeline.addLast(new StringDecoder());
pipeline.addLast(new BusinessHandler());

// 方案二：长度字段（二进制协议，推荐）
pipeline.addLast(new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 4));
pipeline.addLast(new ByteArrayDecoder());
pipeline.addLast(new BusinessHandler());

七、实战示例 ​

7.1 HTTP 服务器 ​

java
public class HttpServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                     .channel(NioServerSocketChannel.class)
                     .childHandler(new ChannelInitializer<SocketChannel>() {
                         @Override
                         protected void initChannel(SocketChannel ch) {
                             ChannelPipeline p = ch.pipeline();
                             // HTTP 编解码器
                             p.addLast(new HttpServerCodec());
                             // HTTP 聚合器（将多个 HttpContent 聚合为 FullHttpRequest）
                             p.addLast(new HttpObjectAggregator(65536));
                             // 业务处理
                             p.addLast(new HttpServerHandler());
                         }
                     });

            ChannelFuture future = bootstrap.bind(8080).sync();
            System.out.println("HTTP 服务器启动，端口：8080");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
        // 打印请求信息
        System.out.println("请求方法：" + request.method());
        System.out.println("请求 URI：" + request.uri());
        System.out.println("请求头：" + request.headers());

        // 构造响应
        String responseContent = "<html><body><h1>Hello, Netty HTTP!</h1></body></html>";
        FullHttpResponse response = new DefaultFullHttpResponse(
                HttpVersion.HTTP_1_1,
                HttpResponseStatus.OK,
                Unpooled.copiedBuffer(responseContent, CharsetUtil.UTF_8));

        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=utf-8");
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());

        // 写入响应
        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}
public class HttpServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                     .channel(NioServerSocketChannel.class)
                     .childHandler(new ChannelInitializer<SocketChannel>() {
                         @Override
                         protected void initChannel(SocketChannel ch) {
                             ChannelPipeline p = ch.pipeline();
                             // HTTP 编解码器
                             p.addLast(new HttpServerCodec());
                             // HTTP 聚合器（将多个 HttpContent 聚合为 FullHttpRequest）
                             p.addLast(new HttpObjectAggregator(65536));
                             // 业务处理
                             p.addLast(new HttpServerHandler());
                         }
                     });

            ChannelFuture future = bootstrap.bind(8080).sync();
            System.out.println("HTTP 服务器启动，端口：8080");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
        // 打印请求信息
        System.out.println("请求方法：" + request.method());
        System.out.println("请求 URI：" + request.uri());
        System.out.println("请求头：" + request.headers());

        // 构造响应
        String responseContent = "<html><body><h1>Hello, Netty HTTP!</h1></body></html>";
        FullHttpResponse response = new DefaultFullHttpResponse(
                HttpVersion.HTTP_1_1,
                HttpResponseStatus.OK,
                Unpooled.copiedBuffer(responseContent, CharsetUtil.UTF_8));

        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=utf-8");
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());

        // 写入响应
        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

7.2 WebSocket 服务端 ​

java
public class WebSocketServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                     .channel(NioServerSocketChannel.class)
                     .childHandler(new ChannelInitializer<SocketChannel>() {
                         @Override
                         protected void initChannel(SocketChannel ch) {
                             ChannelPipeline p = ch.pipeline();
                             // HTTP 编解码器（用于 WebSocket 握手）
                             p.addLast(new HttpServerCodec());
                             // 聚合 HTTP 消息
                             p.addLast(new HttpObjectAggregator(65536));
                             // WebSocket 协议处理器
                             // 参数：访问路径 /ws，最大帧大小，子协议，允许扩展
                             p.addLast(new WebSocketServerProtocolHandler("/ws"));
                             // 业务处理
                             p.addLast(new WebSocketHandler());
                         }
                     });

            ChannelFuture future = bootstrap.bind(8081).sync();
            System.out.println("WebSocket 服务器启动，端口：8081，路径：/ws");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        System.out.println("新连接：" + ctx.channel().remoteAddress());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        System.out.println("连接断开：" + ctx.channel().remoteAddress());
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
        System.out.println("收到消息：" + msg.text());

        // 回复消息
        TextWebSocketFrame response = new TextWebSocketFrame("服务端回复：" + msg.text());
        ctx.channel().writeAndFlush(response);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}
public class WebSocketServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                     .channel(NioServerSocketChannel.class)
                     .childHandler(new ChannelInitializer<SocketChannel>() {
                         @Override
                         protected void initChannel(SocketChannel ch) {
                             ChannelPipeline p = ch.pipeline();
                             // HTTP 编解码器（用于 WebSocket 握手）
                             p.addLast(new HttpServerCodec());
                             // 聚合 HTTP 消息
                             p.addLast(new HttpObjectAggregator(65536));
                             // WebSocket 协议处理器
                             // 参数：访问路径 /ws，最大帧大小，子协议，允许扩展
                             p.addLast(new WebSocketServerProtocolHandler("/ws"));
                             // 业务处理
                             p.addLast(new WebSocketHandler());
                         }
                     });

            ChannelFuture future = bootstrap.bind(8081).sync();
            System.out.println("WebSocket 服务器启动，端口：8081，路径：/ws");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        System.out.println("新连接：" + ctx.channel().remoteAddress());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        System.out.println("连接断开：" + ctx.channel().remoteAddress());
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
        System.out.println("收到消息：" + msg.text());

        // 回复消息
        TextWebSocketFrame response = new TextWebSocketFrame("服务端回复：" + msg.text());
        ctx.channel().writeAndFlush(response);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

7.3 心跳检测（IdleStateHandler） ​

java
public class HeartbeatServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                     .channel(NioServerSocketChannel.class)
                     .childHandler(new ChannelInitializer<SocketChannel>() {
                         @Override
                         protected void initChannel(SocketChannel ch) {
                             ChannelPipeline p = ch.pipeline();
                             // 空闲检测：
                             // readerIdleTime = 5s（5秒未读到数据触发事件）
                             // writerIdleTime = 0（不检测写空闲）
                             // allIdleTime = 0（不检测全部空闲）
                             p.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
                             p.addLast(new HeartbeatHandler());
                         }
                     });

            ChannelFuture future = bootstrap.bind(8082).sync();
            System.out.println("心跳检测服务器启动，端口：8082");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

class HeartbeatHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            switch (event.state()) {
                case READER_IDLE:
                    System.out.println("读空闲，关闭连接：" + ctx.channel().remoteAddress());
                    ctx.close();
                    break;
                case WRITER_IDLE:
                    System.out.println("写空闲");
                    break;
                case ALL_IDLE:
                    System.out.println("读写都空闲");
                    break;
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("收到心跳包：" + msg);
        ctx.fireChannelRead(msg);
    }
}
public class HeartbeatServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                     .channel(NioServerSocketChannel.class)
                     .childHandler(new ChannelInitializer<SocketChannel>() {
                         @Override
                         protected void initChannel(SocketChannel ch) {
                             ChannelPipeline p = ch.pipeline();
                             // 空闲检测：
                             // readerIdleTime = 5s（5秒未读到数据触发事件）
                             // writerIdleTime = 0（不检测写空闲）
                             // allIdleTime = 0（不检测全部空闲）
                             p.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
                             p.addLast(new HeartbeatHandler());
                         }
                     });

            ChannelFuture future = bootstrap.bind(8082).sync();
            System.out.println("心跳检测服务器启动，端口：8082");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

class HeartbeatHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            switch (event.state()) {
                case READER_IDLE:
                    System.out.println("读空闲，关闭连接：" + ctx.channel().remoteAddress());
                    ctx.close();
                    break;
                case WRITER_IDLE:
                    System.out.println("写空闲");
                    break;
                case ALL_IDLE:
                    System.out.println("读写都空闲");
                    break;
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("收到心跳包：" + msg);
        ctx.fireChannelRead(msg);
    }
}

客户端心跳发送示例：

java
// 客户端定时发送心跳
ctx.executor().scheduleAtFixedRate(() -> {
    if (ctx.channel().isActive()) {
        ctx.writeAndFlush(new HeartbeatMessage());
    }
}, 0, 3, TimeUnit.SECONDS);
// 客户端定时发送心跳
ctx.executor().scheduleAtFixedRate(() -> {
    if (ctx.channel().isActive()) {
        ctx.writeAndFlush(new HeartbeatMessage());
    }
}, 0, 3, TimeUnit.SECONDS);

八、Netty 高级 ​

8.1 EventLoop 与线程模型 ​

Netty 的线程模型核心思想：

  1. 一个 EventLoop 绑定一个线程，在其生命周期内不会变更
  2. 一个 Channel 在生命周期内只注册到一个 EventLoop，保证线程安全
  3. 一个 EventLoop 可管理多个 Channel，提升线程利用率
  4. EventLoop 内部维护了一个 Selector，循环处理 I/O 事件
EventLoopGroup（线程池）
   │
   ├── EventLoop-1（线程1）── Channel A, Channel B, Channel C
   ├── EventLoop-2（线程2）── Channel D, Channel E
   └── EventLoop-3（线程3）── Channel F, Channel G, Channel H, Channel I
EventLoopGroup（线程池）
   │
   ├── EventLoop-1（线程1）── Channel A, Channel B, Channel C
   ├── EventLoop-2（线程2）── Channel D, Channel E
   └── EventLoop-3（线程3）── Channel F, Channel G, Channel H, Channel I

线程模型的优势：

  • 通过 Channel 绑定固定 EventLoop 避免了锁竞争
  • 将 I/O 事件与业务任务分离，各自在独立线程中执行
  • 支持非阻塞 I/O（NIO）、异步回调、定时任务

EventLoop 任务队列：

java
// 提交普通任务到 EventLoop 执行
ctx.channel().eventLoop().execute(() -> {
    System.out.println("在 EventLoop 线程中执行");
});

// 提交定时任务
ctx.channel().eventLoop().schedule(() -> {
    System.out.println("延迟 5 秒执行");
}, 5, TimeUnit.SECONDS);

// 提交周期任务
ctx.channel().eventLoop().scheduleAtFixedRate(() -> {
    System.out.println("每 3 秒执行一次");
}, 0, 3, TimeUnit.SECONDS);
// 提交普通任务到 EventLoop 执行
ctx.channel().eventLoop().execute(() -> {
    System.out.println("在 EventLoop 线程中执行");
});

// 提交定时任务
ctx.channel().eventLoop().schedule(() -> {
    System.out.println("延迟 5 秒执行");
}, 5, TimeUnit.SECONDS);

// 提交周期任务
ctx.channel().eventLoop().scheduleAtFixedRate(() -> {
    System.out.println("每 3 秒执行一次");
}, 0, 3, TimeUnit.SECONDS);

8.2 内存管理（Arena 分配器） ​

Netty 内存分配的核心是 PoolArena，采用 jemalloc 分配算法思想。

分配机制：

PoolArena
   ├── Tiny 区域（0 ~ 512B，步进 16B）
   ├── Small 区域（512B ~ 8KB，步进 512B）
   ├── Normal 区域（8KB ~ 16MB，使用 Page/Subpage）
   └── Huge 区域（> 16MB，直接分配堆外内存）
PoolArena
   ├── Tiny 区域（0 ~ 512B，步进 16B）
   ├── Small 区域（512B ~ 8KB，步进 512B）
   ├── Normal 区域（8KB ~ 16MB，使用 Page/Subpage）
   └── Huge 区域（> 16MB，直接分配堆外内存）

内存分配流程：

  1. 优先从 ThreadLocal 缓存获取内存
  2. 若缓存未命中，从 Arena 分配
  3. 若 Arena 也无法分配，从操作系统分配
java
// 查看内存池统计信息
PooledByteBufAllocator allocator = PooledByteBufAllocator.DEFAULT;

// 获取分配器指标
System.out.println("堆内存使用：" + allocator.metric().heapChunks());
System.out.println("直接内存使用：" + allocator.metric().directChunks());
// 查看内存池统计信息
PooledByteBufAllocator allocator = PooledByteBufAllocator.DEFAULT;

// 获取分配器指标
System.out.println("堆内存使用：" + allocator.metric().heapChunks());
System.out.println("直接内存使用：" + allocator.metric().directChunks());

配置内存池：

java
// 在启动时配置
System.setProperty("io.netty.allocator.maxOrder", "9");       // 默认 11
System.setProperty("io.netty.allocator.pageSize", "8192");    // 默认 8192
System.setProperty("io.netty.allocator.numHeapArenas", "2");  // 默认 CPU*2
System.setProperty("io.netty.allocator.numDirectArenas", "2");
System.setProperty("io.netty.allocator.tinyCacheSize", "512");
System.setProperty("io.netty.allocator.useCacheForAllThreads", "true");
// 在启动时配置
System.setProperty("io.netty.allocator.maxOrder", "9");       // 默认 11
System.setProperty("io.netty.allocator.pageSize", "8192");    // 默认 8192
System.setProperty("io.netty.allocator.numHeapArenas", "2");  // 默认 CPU*2
System.setProperty("io.netty.allocator.numDirectArenas", "2");
System.setProperty("io.netty.allocator.tinyCacheSize", "512");
System.setProperty("io.netty.allocator.useCacheForAllThreads", "true");

8.3 零拷贝（Zero-Copy） ​

Netty 的零拷贝不是指操作系统层面的零拷贝（如 sendfile），而是在用户态层面减少不必要的内存拷贝。

（1）CompositeByteBuf —— 组合多个 ByteBuf 为一个逻辑视图

java
// 将两个 ByteBuf 组合成逻辑上的一个，无需物理拷贝
CompositeByteBuf composite = Unpooled.compositeBuffer();
ByteBuf header = Unpooled.copiedBuffer("Header", CharsetUtil.UTF_8);
ByteBuf body = Unpooled.copiedBuffer("Body", CharsetUtil.UTF_8);

// 添加组件，不进行拷贝（需设置 index 为 true）
composite.addComponents(true, header, body);
// 将两个 ByteBuf 组合成逻辑上的一个，无需物理拷贝
CompositeByteBuf composite = Unpooled.compositeBuffer();
ByteBuf header = Unpooled.copiedBuffer("Header", CharsetUtil.UTF_8);
ByteBuf body = Unpooled.copiedBuffer("Body", CharsetUtil.UTF_8);

// 添加组件，不进行拷贝（需设置 index 为 true）
composite.addComponents(true, header, body);

（2）Unpooled.wrappedBuffer —— 包装现有数组

java
byte[] array1 = "Hello ".getBytes();
byte[] array2 = "World".getBytes();

// 将多个字节数组包装为一个 ByteBuf，无拷贝
ByteBuf wrapped = Unpooled.wrappedBuffer(array1, array2);
byte[] array1 = "Hello ".getBytes();
byte[] array2 = "World".getBytes();

// 将多个字节数组包装为一个 ByteBuf，无拷贝
ByteBuf wrapped = Unpooled.wrappedBuffer(array1, array2);

（3）slice —— 切片操作

java
ByteBuf buf = Unpooled.copiedBuffer("Hello World", CharsetUtil.UTF_8);

// 创建原缓冲区的切片，共享同一块内存
ByteBuf slice = buf.slice(0, 5); // "Hello"
// slice 和 buf 共享底层内存，修改 slice 会影响 buf
ByteBuf buf = Unpooled.copiedBuffer("Hello World", CharsetUtil.UTF_8);

// 创建原缓冲区的切片，共享同一块内存
ByteBuf slice = buf.slice(0, 5); // "Hello"
// slice 和 buf 共享底层内存，修改 slice 会影响 buf

（4）FileRegion（sendfile 系统调用）

java
// 使用 FileRegion 实现文件零拷贝传输
// 底层利用操作系统的 sendfile 系统调用
public class FileServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        File file = new File("/path/to/large/file.zip");
        FileRegion region = new DefaultFileRegion(
                new FileInputStream(file).getChannel(), 0, file.length());

        // 直接发送 FileRegion，数据不经过用户态
        ctx.writeAndFlush(region).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                System.out.println("文件传输完成");
            }
        });
    }
}
// 使用 FileRegion 实现文件零拷贝传输
// 底层利用操作系统的 sendfile 系统调用
public class FileServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        File file = new File("/path/to/large/file.zip");
        FileRegion region = new DefaultFileRegion(
                new FileInputStream(file).getChannel(), 0, file.length());

        // 直接发送 FileRegion，数据不经过用户态
        ctx.writeAndFlush(region).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                System.out.println("文件传输完成");
            }
        });
    }
}

8.4 流量控制（writeAndFlush 高水位） ​

防止生产者过快写入导致 OOM，Netty 提供了写缓冲区水位机制。

java
// 服务端配置写水位
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
        new WriteBufferWaterMark(32 * 1024,      // 低水位：32KB
                                  64 * 1024));    // 高水位：64KB

// 客户端判断水位
public class WaterMarkHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        Channel channel = ctx.channel();

        // 判断是否达到高水位
        if (!channel.isWritable()) {
            System.out.println("写缓冲区已满，进行背压处理");
            // 停止生产或丢弃消息
            return;
        }

        channel.writeAndFlush(msg);
    }
}
// 服务端配置写水位
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
        new WriteBufferWaterMark(32 * 1024,      // 低水位：32KB
                                  64 * 1024));    // 高水位：64KB

// 客户端判断水位
public class WaterMarkHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        Channel channel = ctx.channel();

        // 判断是否达到高水位
        if (!channel.isWritable()) {
            System.out.println("写缓冲区已满，进行背压处理");
            // 停止生产或丢弃消息
            return;
        }

        channel.writeAndFlush(msg);
    }
}

水位工作原理：

未达到高水位：正常写入
达到高水位  → Channel.isWritable() 返回 false → 触发 channelWritabilityChanged
降到低水位  → Channel.isWritable() 返回 true  → 恢复写入
未达到高水位：正常写入
达到高水位  → Channel.isWritable() 返回 false → 触发 channelWritabilityChanged
降到低水位  → Channel.isWritable() 返回 true  → 恢复写入

背压处理示例：

java
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) {
    if (ctx.channel().isWritable()) {
        // 恢复数据发送
        System.out.println("缓冲区可写，恢复发送");
    } else {
        // 暂停数据发送
        System.out.println("缓冲区满，暂停发送");
    }
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) {
    if (ctx.channel().isWritable()) {
        // 恢复数据发送
        System.out.println("缓冲区可写，恢复发送");
    } else {
        // 暂停数据发送
        System.out.println("缓冲区满，暂停发送");
    }
}

九、实际应用场景 ​

9.1 Dubbo 底层 RPC 通信 ​

Dubbo 的远程调用基于 Netty 实现，在 2.7+ 版本中默认使用 Netty 作为传输层。

Dubbo 协议编解码：

  • 使用 LengthFieldBasedFrameDecoder 解决粘包拆包
  • 请求头包含魔数（0xdabb）、序列化 ID、请求 ID、数据长度等字段
  • 支持多种序列化方式：Hessian2、JSON、Kryo、Protobuf
java
// Dubbo 解码器简化示意
public class DubboDecoder extends LengthFieldBasedFrameDecoder {

    public DubboDecoder() {
        super(8 * 1024 * 1024,   // 最大帧长度 8MB
              12,                // 长度字段偏移
              4,                 // 长度字段长度
              0,                 // 长度调整
              16);               // 剥离头部长度
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) {
        ByteBuf frame = (ByteBuf) super.decode(ctx, in);
        if (frame == null) return null;

        // 解析 Dubbo 协议头
        short magic = frame.readShort();        // 魔数
        byte serializationId = frame.readByte(); // 序列化方式
        long requestId = frame.readLong();       // 请求 ID
        int bodyLength = frame.readInt();        // 消息体长度

        // 反序列化 body
        byte[] body = new byte[bodyLength];
        frame.readBytes(body);
        return deserialize(body, serializationId);
    }
}
// Dubbo 解码器简化示意
public class DubboDecoder extends LengthFieldBasedFrameDecoder {

    public DubboDecoder() {
        super(8 * 1024 * 1024,   // 最大帧长度 8MB
              12,                // 长度字段偏移
              4,                 // 长度字段长度
              0,                 // 长度调整
              16);               // 剥离头部长度
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) {
        ByteBuf frame = (ByteBuf) super.decode(ctx, in);
        if (frame == null) return null;

        // 解析 Dubbo 协议头
        short magic = frame.readShort();        // 魔数
        byte serializationId = frame.readByte(); // 序列化方式
        long requestId = frame.readLong();       // 请求 ID
        int bodyLength = frame.readInt();        // 消息体长度

        // 反序列化 body
        byte[] body = new byte[bodyLength];
        frame.readBytes(body);
        return deserialize(body, serializationId);
    }
}

9.2 Spring Cloud Gateway（WebFlux + Netty） ​

Spring Cloud Gateway 基于 Spring WebFlux，底层使用 Netty 作为异步 I/O 引擎。

请求处理流程：

客户端请求 → Netty 接收 → WebFlux Handler → Gateway Filter Chain → 转发到后端
     ↑                                                                  │
     └────────────────────────── 响应返回 ←──────────────────────────────┘
客户端请求 → Netty 接收 → WebFlux Handler → Gateway Filter Chain → 转发到后端
     ↑                                                                  │
     └────────────────────────── 响应返回 ←──────────────────────────────┘

Gateway 与 Netty 的关系：

  • Reactor Netty 是 WebFlux 的默认底层实现
  • 每个 HTTP 连接对应一个 Netty Channel
  • 使用 Netty 的 EventLoop 模型处理 I/O
  • 通过 reactor.netty 包与 Netty 交互
yaml
# application.yml 中配置 Netty 参数
spring:
  cloud:
    gateway:
      httpclient:
        connect-timeout: 5000
        response-timeout: 10s
        pool:
          type: elastic
          max-connections: 1000
          max-idle-time: 60000
# application.yml 中配置 Netty 参数
spring:
  cloud:
    gateway:
      httpclient:
        connect-timeout: 5000
        response-timeout: 10s
        pool:
          type: elastic
          max-connections: 1000
          max-idle-time: 60000

9.3 Elasticsearch 通信层 ​

Elasticsearch 使用 Netty 作为节点间通信（Transport）和 HTTP REST 请求的基础。

Elasticsearch Node
   ├── HTTP Server（Netty） ← REST 请求（9200 端口）
   │     └── NettyHttpServerTransport
   │
   └── Transport（Netty） ← 节点间通信（9300 端口）
         └── NettyTransport
Elasticsearch Node
   ├── HTTP Server（Netty） ← REST 请求（9200 端口）
   │     └── NettyHttpServerTransport
   │
   └── Transport（Netty） ← 节点间通信（9300 端口）
         └── NettyTransport

ES 对 Netty 的定制：

  • 自实现 EsMessageDecoder/EsMessageEncoder 编解码
  • 管理多个 Netty 线程池（HTTP 与 Transport 分离）
  • 通过 NetworkModule 注册自定义 ChannelHandler

9.4 RocketMQ 通信层 ​

RocketMQ 使用 Netty 处理 Broker 与 Producer/Consumer 之间的通信。

RocketMQ Netty 配置：

java
// RocketMQ Netty 配置示意
public class RocketMQNettyConfig {
    // 服务端配置
    public static final int SERVER_WORKER_THREADS = 8;
    public static final int SERVER_CHANNEL_MAX_IDLE_SECONDS = 120;

    // 客户端配置
    public static final int CLIENT_WORKER_THREADS = 4;
    public static final int CLIENT_CONNECT_TIMEOUT_MILLIS = 3000;
    public static final int CLIENT_CHANNEL_MAX_IDLE_SECONDS = 60;
}
// RocketMQ Netty 配置示意
public class RocketMQNettyConfig {
    // 服务端配置
    public static final int SERVER_WORKER_THREADS = 8;
    public static final int SERVER_CHANNEL_MAX_IDLE_SECONDS = 120;

    // 客户端配置
    public static final int CLIENT_WORKER_THREADS = 4;
    public static final int CLIENT_CONNECT_TIMEOUT_MILLIS = 3000;
    public static final int CLIENT_CHANNEL_MAX_IDLE_SECONDS = 60;
}

RocketMQ 的 Netty 特性使用：

  • 使用 LengthFieldBasedFrameDecoder 处理消息边界
  • 使用 IdleStateHandler 做心跳检测
  • 使用 ChannelPipeline 链式处理编解码、业务处理
  • 使用 ChannelFutureListener 处理异步回调

十、Netty 开发最佳实践 ​

10.1 内存泄漏排查 ​

Netty 提供了内存泄漏检测机制，通过 JVM 参数启用：

bash
# 内存泄漏检测级别
-Dio.netty.leakDetectionLevel=paranoid
# 级别：disabled < simple < advanced < paranoid

# 推荐开发环境使用 paranoid，生产环境使用 advanced
-Dio.netty.leakDetectionLevel=advanced
# 内存泄漏检测级别
-Dio.netty.leakDetectionLevel=paranoid
# 级别：disabled < simple < advanced < paranoid

# 推荐开发环境使用 paranoid，生产环境使用 advanced
-Dio.netty.leakDetectionLevel=advanced
java
// 代码中检测 ByteBuf 是否泄漏
ReferenceCountUtil.retain(msg);   // 需要延长生命周期时
ReferenceCountUtil.release(msg);  // 使用完毕后释放

// 确保 finally 中释放
ByteBuf buf = allocator.buffer();
try {
    // 使用 buf
} finally {
    buf.release();
}
// 代码中检测 ByteBuf 是否泄漏
ReferenceCountUtil.retain(msg);   // 需要延长生命周期时
ReferenceCountUtil.release(msg);  // 使用完毕后释放

// 确保 finally 中释放
ByteBuf buf = allocator.buffer();
try {
    // 使用 buf
} finally {
    buf.release();
}

10.2 常用优化参数 ​

java
ServerBootstrap bootstrap = new ServerBootstrap();

// TCP 参数优化
bootstrap.option(ChannelOption.SO_BACKLOG, 1024)           // 连接队列大小
         .option(ChannelOption.TCP_NODELAY, true)           // 禁用 Nagle 算法
         .option(ChannelOption.SO_KEEPALIVE, true)          // TCP Keep-Alive
         .option(ChannelOption.SO_REUSEADDR, true)          // 地址复用
         .option(ChannelOption.SO_RCVBUF, 1024 * 1024)     // 接收缓冲区 1MB
         .option(ChannelOption.SO_SNDBUF, 1024 * 1024)     // 发送缓冲区 1MB

         // Netty 特有参数
         .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
             new WriteBufferWaterMark(32 * 1024, 64 * 1024))
         .childOption(ChannelOption.ALLOCATOR,
             PooledByteBufAllocator.DEFAULT)                 // 池化分配器
         .childOption(ChannelOption.RCVBUF_ALLOCATOR,
             new AdaptiveRecvByteBufAllocator());            // 自适应缓冲区
ServerBootstrap bootstrap = new ServerBootstrap();

// TCP 参数优化
bootstrap.option(ChannelOption.SO_BACKLOG, 1024)           // 连接队列大小
         .option(ChannelOption.TCP_NODELAY, true)           // 禁用 Nagle 算法
         .option(ChannelOption.SO_KEEPALIVE, true)          // TCP Keep-Alive
         .option(ChannelOption.SO_REUSEADDR, true)          // 地址复用
         .option(ChannelOption.SO_RCVBUF, 1024 * 1024)     // 接收缓冲区 1MB
         .option(ChannelOption.SO_SNDBUF, 1024 * 1024)     // 发送缓冲区 1MB

         // Netty 特有参数
         .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK,
             new WriteBufferWaterMark(32 * 1024, 64 * 1024))
         .childOption(ChannelOption.ALLOCATOR,
             PooledByteBufAllocator.DEFAULT)                 // 池化分配器
         .childOption(ChannelOption.RCVBUF_ALLOCATOR,
             new AdaptiveRecvByteBufAllocator());            // 自适应缓冲区

10.3 Pipeline 编排建议 ​

java
// 推荐的 Pipeline 编排顺序
ChannelPipeline p = ch.pipeline();

// 1. 解码器（最前）
p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(...));
p.addLast("msgDecoder", new MessageDecoder());

// 2. 编码器（在解码器之后）
p.addLast("msgEncoder", new MessageEncoder());

// 3. 空闲检测
p.addLast("idleHandler", new IdleStateHandler(60, 0, 0));

// 4. 业务 Handler
p.addLast("businessHandler", new BusinessHandler());

// 5. 异常处理（最后）
p.addLast("exceptionHandler", new ExceptionHandler());
// 推荐的 Pipeline 编排顺序
ChannelPipeline p = ch.pipeline();

// 1. 解码器（最前）
p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(...));
p.addLast("msgDecoder", new MessageDecoder());

// 2. 编码器（在解码器之后）
p.addLast("msgEncoder", new MessageEncoder());

// 3. 空闲检测
p.addLast("idleHandler", new IdleStateHandler(60, 0, 0));

// 4. 业务 Handler
p.addLast("businessHandler", new BusinessHandler());

// 5. 异常处理（最后）
p.addLast("exceptionHandler", new ExceptionHandler());

10.4 线程模型选择 ​

场景推荐线程模型说明
高并发 HTTP 服务boss=1, worker=CPU*2标准主从 Reactor
长连接 IM 服务boss=1, worker=CPU*4更多 worker 处理读写
文件传输服务boss=1, worker=CPUI/O 密集型，worker 不宜过多
网关代理boss=1, worker=CPU*2转发为主，计算较少

参考资源 ​

Copyright © 2024-present xiaocao