--- url: /java/netty.md --- # Netty 全面知识体系 ## 一、Netty 概述 Netty 是一个**异步事件驱动的网络应用框架**，用于快速开发可维护的高性能协议服务器和客户端。基于 Java NIO（Non-blocking I/O）实现，采用 Reactor 模式，提供高效、灵活的网络编程抽象。 ### 核心特性 * **高性能**：零拷贝、内存池、自适应分配器 * **高可维护性**：清晰的 API 设计、完善的文档 * **丰富的协议支持**：HTTP、HTTP/2、HTTP/3、WebSocket、SSL/TLS * **跨平台传输层**：NIO、OIO、Epoll（Linux）、KQueue（BSD/macOS） * **强大的生态**：Spring Cloud、Dubbo、gRPC 等框架底层依赖 *** ## 二、核心架构 ### 2.1 Reactor 线程模型 Netty 采用经典的**主从 Reactor 多线程模型**： ``` ┌─────────────┐ ┌──────────────────┐ ┌─────────────────┐ │ Boss │────▶│ Accept Event │────▶│ Worker │ │ EventLoop │ │ (注册连接) │ │ EventLoop │ │ Group │ └──────────────────┘ │ Group │ └─────────────┘ │ ┌────────────┐ │ │ │ IO Events │ │ │ │ (读写操作) │ │ │ └────────────┘ │ └─────────────────┘ ``` #### BossGroup（ Selector 组） * 负责**接收客户端连接请求** * 通常只需 1 个线程（单核即可处理 accept 操作） * 将已建立的连接注册到 WorkerGroup 的 Selector #### WorkerGroup（IO 线程组） * 负责**网络 IO 读写操作** * 默认线程数 = CPU 核心数 \* 2 * 每个 Channel 绑定到一个 EventLoop，且生命周期内不变（线程安全） ### 2.2 EventLoop 核心机制 EventLoop 是 Netty 的心脏，关键特性： ```java // EventLoop 核心接口 public interface EventLoop extends Executor { boolean inEventLoop(Thread thread); // 判断当前线程是否属于该 EventLoop Future submit(Runnable task); // 提交任务 void shutdownGracefully(); // 优雅关闭 ChannelFuture register(Channel channel); // 注册 Channel } ``` **重要特性：** 1. **一个 EventLoop 对应一个线程**（NioEventLoop） 2. **Channel 与 EventLoop 一对一绑定**，保证线程安全 3. **运行队列 + Selector**：既处理 IO 事件也处理定时任务/普通任务 4. **无限循环**：`while (!shutdown)` 不断轮询 selector 和处理任务队列 ### 2.3 Channel 通道 Channel 是 Netty 对网络连接的抽象，类似 JDK NIO 的 SocketChannel 但更强大： ```java public interface Channel { ChannelConfig config(); // 获取配置 boolean isRegistered(); // 是否已注册到 EventLoop boolean isActive(); // 是否处于活动状态 ChannelMetadata metadata(); // 元数据 ChannelPipeline pipeline(); // 获取责任链 ChannelFuture connect(SocketAddress remoteAddress); // 发起连接 ChannelFuture bind(SocketAddress localAddress); // 绑定端口 ChannelFuture close(); // 关闭通道 } ``` **Channel 状态机：** * `NEW` → `REGISTERED` → `ACTIVE` → `INACTIVE` → `CLOSED` ### 2.4 ChannelPipeline & ChannelHandler #### Pipeline（责任链） Pipeline 是 Handler 的容器，形成**单向链表结构**： ``` Head Handler → Business Handler 1 → Business Handler 2 → ... → Tail Handler ``` **消息流向：** * **Inbound（入站）**：从 Head 向 Tail 传播（自左向右） * **Outbound（出站）**：从 Tail 向 Head 传播（自右向左） #### ChannelHandler 分类 | 类型 | 适配器 | 主要方法 | 说明 | |------|--------|----------|------| | Inbound | ChannelInboundHandlerAdapter | channelRead, channelActive, exceptionCaught | 处理入站事件 | | Outbound | ChannelOutboundHandlerAdapter | write, read, flush, connect | 处理出站操作 | | Duplex | ChannelDuplexHandler | 两者都有 | 同时处理双向 | **核心生命周期方法：** ```java @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception {} // 添加时调用 @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {} // 移除时调用 @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {} // 异常捕获 ``` **Inbound 关键方法：** ```java void channelRegistered(ChannelHandlerContext ctx); // Channel 注册到 EventLoop void channelUnregistered(ChannelHandlerContext ctx); // Channel 从 EventLoop 注销 void channelActive(ChannelHandlerContext ctx); // 连接建立成功 void channelInactive(ChannelHandlerContext ctx); // 连接断开 void channelRead(ChannelHandlerContext ctx, Object msg);// 读取到数据 void channelReadComplete(ChannelHandlerContext ctx); // 读取完成 void userEventTriggered(ChannelHandlerContext ctx, Object evt); // 用户自定义事件 ``` ### 2.5 Bootstrap 引导类 #### ServerBootstrap（服务端） ```java EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); ServerBootstrap bootstrap = new ServerBootstrap() .group(bossGroup, workerGroup) // 设置线程组 .channel(NioServerSocketChannel.class) // 服务端 Channel 类型 .option(ChannelOption.SO_BACKLOG, 128) // 服务端选项 .childOption(ChannelOption.SO_KEEPALIVE, true) // 子 Channel 选项 .childHandler(new ChannelInitializer() { // 子 Channel 初始化器 @Override protected void initChannel(SocketChannel ch) { ch.pipeline().addLast(new BusinessHandler()); } }); ChannelFuture future = bootstrap.bind(8080).sync(); future.channel().closeFuture().sync(); // 等待关闭 ``` #### Bootstrap（客户端） ```java EventLoopGroup group = new NioEventLoopGroup(); Bootstrap bootstrap = new Bootstrap() .group(group) .channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) .handler(new ChannelInitializer() { @Override protected void initChannel(SocketChannel ch) { ch.pipeline().addLast(new ClientHandler()); } }); ChannelFuture future = bootstrap.connect("127.0.0.1", 8080).sync(); ``` *** ## 三、ByteBuf 缓冲区 ### 3.1 ByteBuf 核心概念 ByteBuf 是 Netty 对 JDK NIO ByteBuffer 的增强，解决了很多痛点： | 特性 | JDK ByteBuffer | Netty ByteBuf | |------|----------------|---------------| | 读/写索引 | 共用 position，需要 flip() | 独立 readerIndex/writerIndex | | 边界检查 | 严格，越界抛异常 | 可选择性跳过检查 | | 堆外内存 | 需手动管理 | 统一管理，自动释放 | | 复合缓冲 | 不支持 | CompositeByteBuf | | 零拷贝 | 不支持 | slice/duplicate/wrap | ### 3.2 内存管理 #### 引用计数机制 ```java ByteBuf buf = Unpooled.buffer(1024); System.out.println(buf.refCnt()); // 输出 1，初始引用计数为 1 buf.retained(); // 引用计数 +1 buf.release(); // 引用计数 -1，归零时释放内存 ``` **重要规则：** * **新创建的 ByteBuf 引用计数为 1** * **必须手动 release()**，否则会导致内存泄漏 * **pipeline 中传递的消息会自动释放**，但业务处理后需自行管理 * **CompositeByteBuf 的子 buffer 不会自动释放** #### 堆内 vs 堆外 | 类型 | 分配位置 | GC 影响 | 性能 | 适用场景 | |------|----------|--------|------|----------| | Heap Buffer | JVM 堆内 | 受 GC 影响 | 较低 | 小对象、频繁访问 | | Direct Buffer | JVM 堆外 | 不受 GC 影响 | 高 | 网络 IO、大文件 | ```java ByteBufAllocator allocator = ctx.alloc(); ByteBuf heapBuf = allocator.heapBuffer(1024); // 堆内缓冲 ByteBuf directBuf = allocator.directBuffer(1024); // 堆外缓冲 ByteBuf autoBuf = allocator.buffer(1024); // 自动选择 ``` ### 3.3 内存池（PooledByteBufAllocator） Netty 4.x 使用**内存池**减少内存分配和 GC 压力： ```java // 默认配置 PooledByteBufAllocator.DEFAULT; // 自定义配置 PooledByteBufAllocator allocator = new PooledByteBufAllocator( true, // preferDirect: 优先使用堆外内存 4, // numHeapArena: 堆内内存区域数 4, // numDirectArena: 堆外内存区域数 PageOrder(11), // chunkSize = 2^11 = 2KB PageOrder(9), // pageSize = 2^9 = 512B PageOrder(4), // smallCacheSize PageOrder(4), // normalCacheSize true, // useCacheForAllThreads 64, // tinyCacheSize 1 // maxCachedBufferCapacity ); ``` **内存层级结构：** ``` Chunk (16MB) → SubPage/PageSize (16KB/8KB/4KB/2KB/1KB/512B) → PoolChunk → PoolArena ``` ### 3.4 零拷贝技术 #### (1) CompositeByteBuf（聚合缓冲区） ```java CompositeByteBuf composite = Unpooled.compositeBuffer(); composite.addComponent(true, headerBuf); composite.addComponent(true, contentBuf); composite.addComponent(true, footerBuf); // 多个 buffer 逻辑上合并为一个，物理上不复制 ``` #### (2) slice/duplicate（视图缓冲） ```java ByteBuf original = Unpooled.buffer(1024); original.writeBytes(...); ByteBuf slice = original.slice(); // 共享内存，独立索引 ByteBuf duplicate = original.duplicate(); // 共享内存和索引 ``` #### (3) wrap（包装已有数组） ```java byte[] array = new byte[1024]; ByteBuf wrapped = Unpooled.wrappedBuffer(array); // 不复制数据 ``` #### (4) FileRegion & sendfile ```java // 利用操作系统内核零拷贝 RandomAccessFile file = new RandomAccessFile("video.mp4", "r"); FileRegion region = new DefaultFileRegion(file.getChannel(), 0, file.length()); ctx.writeAndFlush(region, ctx.newPromise()); ``` *** ## 四、编解码框架（Codec） ### 4.1 编解码器基类 #### ByteToMessageDecoder（字节转对象） 解决 TCP **粘包/拆包**问题： ```java public class TimeDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) { // in 是累积的缓冲区，可能包含多条消息 if (in.readableBytes() < 4) { return; // 数据不足，等待后续数据 } int value = in.readInt(); out.add(value); // 解码后的消息加入输出列表 } } ``` **关键方法：** * `decode()`: 核心解码逻辑，被反复调用直到无法继续解码 * `decodeRemovalTry()`: 处理剩余数据 * `cumulate()`: 累积数据 #### MessageToByteEncoder（对象转字节） ```java public class TimeEncoder extends MessageToByteEncoder { @Override protected void encode(ChannelHandlerContext ctx, Integer msg, ByteBuf out) { out.writeInt(msg); } } ``` ### 4.2 帧处理器（解决粘包拆包） #### LengthFieldBasedFrameDecoder（长度字段） 最常用，适用于自定义协议： ```java new LengthFieldBasedFrameDecoder( 1024 * 1024, // maxFrameLength: 最大帧长度 0, // lengthFieldOffset: 长度字段偏移量 4, // lengthFieldLength: 长度字段长度 0, // lengthAdjustment: 长度修正值 0 // initialBytesToStrip: 是否跳过长度字段 ); ``` 协议示例：`[4 字节长度][实际数据]` #### DelimiterBasedFrameDecoder（分隔符） ```java // 以\r\n作为分隔符 new DelimiterBasedFrameDecoder( 1024 * 1024, Unpooled.copiedBuffer("\r\n", CharsetUtil.UTF_8) ); ``` #### LineBasedFrameDecoder（行分割） ```java // 按行分割，自动处理 \n 或 \r\n new LineBasedFrameDecoder(1024 * 1024); ``` #### FixedLengthFrameDecoder（固定长度） ```java // 每条消息固定 12 字节 new FixedLengthFrameDecoder(12); ``` ### 4.3 内置编解码器 | 编解码器 | 功能 | |---------|------| | StringEncoder/StringDecoder | 字符串与字节转换 | | ByteArrayEncoder/Decoder | 字节数组转换 | | Base64Encoder/Base64Decoder | Base64 编码 | | JacksonJsonEncoder/JacksonJsonDecoder | JSON 序列化 | | ProtobufEncoder/ProtobufDecoder | Protocol Buffers | | MarshallingEncoder/MarshallingDecoder | Java 序列化 | *** ## 五、常用 Handler ### 5.1 IdleStateHandler（空闲检测） ```java new IdleStateHandler( 60, // readIdleTime: 读空闲时间（秒） 30, // writeIdleTime: 写空闲时间（秒） 0, // allIdleTime: 全部空闲时间（秒） TimeUnit.SECONDS ); // 配合使用 public class IdleHandler extends ChannelInboundHandlerAdapter { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) { if (evt instanceof IdleStateEvent) { IdleStateEvent event = (IdleStateEvent) evt; if (event.state() == IdleState.READER_IDLE) { ctx.close(); // 读空闲时关闭连接 } } } } ``` ### 5.2 ReadTimeoutHandler / WriteTimeoutHandler ```java // 读超时自动关闭连接 new ReadTimeoutHandler(30, TimeUnit.SECONDS); // 写超时自动关闭连接 new WriteTimeoutHandler(30, TimeUnit.SECONDS); ``` ### 5.3 DecoderExceptionEncoder（解码异常处理） ```java // 当解码失败时，将异常编码后发送给客户端 ch.pipeline().addLast(new DecoderExceptionEncoder()); ``` ### 5.4 LoggingHandler（日志记录） ```java ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG)); ``` ### 5.5 SSL/TLS 支持 ```java // 服务端 SSL 配置 SslContext sslCtx = SslContextBuilder.forServer( certChainFile, privateKeyFile ).sslProvider(SslProvider.OPENSSL).build(); ch.pipeline().addLast(sslCtx.newHandler(ch.alloc())); ``` *** ## 六、ChannelOption 配置 ### 6.1 常用 TCP 选项 | Option | 说明 | 推荐值 | |--------|------|--------| | SO_REUSEADDR | 允许复用本地地址 | true | | TCP_NODELAY | 禁用 Nagle 算法 | true（低延迟场景） | | SO_KEEPALIVE | TCP 保活 | true | | SO_BACKLOG | 监听队列长度 | 128/511/1024 | | SO_RCVBUF | 接收缓冲区大小 | 根据场景调整 | | SO_SNDBUF | 发送缓冲区大小 | 根据场景调整 | | IP_TOS | Type of Service | 默认 0 | ### 6.2 Netty 特有选项 | Option | 说明 | |--------|------| | MAX_MESSAGES_PER_READ | 每次 selector 轮询最大读取消息数 | | AUTO_READ | 自动读取数据（默认 true） | | WRITE_SPIN_COUNT | 单次 write 循环次数 | | ALLOCATOR | 内存分配器 | | MESSAGE_SIZE_ESTIMATOR | 消息大小估算器 | ### 6.3 水位线（Flow Control） ```java ch.config().setWriteBufferWaterMark( new WriteBufferWaterMark(32 * 1024, 64 * 1024) // 低水位 32KB，高水位 64KB ); ``` *** ## 七、异步编程模型 ### 7.1 ChannelFuture & Promise ```java // Future-Promise 模式 ChannelFuture future = bootstrap.bind(8080); // 同步等待（阻塞） future.sync(); // 异步回调 future.addListener((ChannelFutureListener) f -> { if (f.isSuccess()) { System.out.println("绑定成功"); } else { f.cause().printStackTrace(); } }); // 链式操作 future.addListener(FutureListener.of(f -> { if (f.isSuccess()) { // 成功处理 } })); ``` ### 7.2 Promise 操作 ```java Promise promise = ctx.newPromise(); // 设置成功 promise.setSuccess(); // 设置失败 promise.setFailure(new Exception("error")); // 取消 promise.tryCancel(false); ``` ### 7.3 组合 Future（Netty 5+） ```java future.map(result -> result * 2) // 映射结果 .flatMap(result -> anotherFuture) // 扁平化 .addListener(f -> { /* 处理 */ }); ``` *** ## 八、传输层适配 ### 8.1 NIO（Java NIO） ```java NioEventLoopGroup group = new NioEventLoopGroup(); NioServerSocketChannel serverChannel; NioSocketChannel clientChannel; ``` ### 8.2 Epoll（Linux 专用，性能最优） ```java EpollEventLoopGroup group = new EpollEventLoopGroup(); EpollServerSocketChannel serverChannel; EpollSocketChannel clientChannel; ``` ### 8.3 KQueue（BSD/macOS） ```java KQueueEventLoopGroup group = new KQueueEventLoopGroup(); KQueueServerSocketChannel serverChannel; KQueueSocketChannel clientChannel; ``` ### 8.4 OIO（阻塞 IO） ```java OioEventLoopGroup group = new OioEventLoopGroup(); OioServerSocketChannel serverChannel; OioSocketChannel clientChannel; ``` *** ## 九、高级特性 ### 9.1 HTTP/HTTP2/WebSocket #### HTTP 编解码 ```java ch.pipeline().addLast(new HttpServerCodec()); ch.pipeline().addLast(new HttpObjectAggregator(65536)); // 聚合 FullHttpRequest ``` #### WebSocket ```java ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws")); ``` #### HTTP/2 ```java ch.pipeline().addLast(new Http2MultiplexCodecBuilder() .frameLogger(new Http2FrameLogger()) .frameListener(new Http2FrameListener() {...}) .build()); ``` ### 9.2 流量控制 #### 手动控制读取 ```java ch.config().setAutoRead(false); // 禁用自动读取 // 在合适时机手动读取 ctx.read(); ``` #### 可写性检测 ```java if (!ch.isWritable()) { // 停止写入，等待可写事件 } @Override public void channelWritabilityChanged(ChannelHandlerContext ctx) { if (ctx.channel().isWritable()) { // 恢复写入 } } ``` ### 9.3 代理支持 ```java // SOCKS4/5 代理 ch.pipeline().addLast(new SocksInitiator(socksAddr)); // HTTP 代理 ch.pipeline().addLast(new HttpProxyHandler(httpProxyAddr)); ``` ### 9.4 线程亲和性（Thread Affinity） ```java // 将线程绑定到特定 CPU 核心，降低上下文切换开销 ThreadFactory factory = new AffinityThreadFactory( "worker", AffinityStrategies.DIFFERENT_CORE ); EventLoopGroup group = new NioEventLoopGroup(0, factory); ``` ### 9.5 半关闭 Socket ```java ch.config().setOption(ChannelOption.ALLOW_HALF_CLOSURE, true); ch.shutdownOutput(); // 只关闭输出流，仍可读取 ``` *** ## 十、性能优化 ### 10.1 内存优化 1. **使用堆外直接内存**：减少 GC 压力 2. **合理配置内存池**：避免频繁分配 3. **及时释放 ByteBuf**：防止内存泄漏 4. **使用 CompositeByteBuf**：减少数据拷贝 ### 10.2 线程优化 1. **BossGroup 单线程足够**：accept 操作很快 2. **WorkerGroup 线程数 = CPU 核心数 \* 2** 3. **CPU 隔离**：使用 isolcpus 参数隔离核心 4. **线程亲和性**：绑定线程到特定核心 ### 10.3 IO 优化 1. **零拷贝**：充分利用 sendfile、CompositeByteBuf 2. **批量读写**：增大 SO_RCVBUF/SO_SNDBUF 3. **禁用 Nagle 算法**：TCP_NODELAY=true 4. **调整 epoll 参数**：EPOLLET 边缘触发模式 ### 10.4 系统级优化 ```bash # Linux 内核参数调优 net.core.somaxconn = 1024 net.ipv4.tcp_max_syn_backlog = 8192 net.ipv4.tcp_tw_reuse = 1 net.ipv4.tcp_fin_timeout = 30 ``` *** ## 十一、最佳实践 ### 11.1 资源管理 ```java try { bossGroup = new NioEventLoopGroup(); workerGroup = new NioEventLoopGroup(); bootstrap = new ServerBootstrap()...; ChannelFuture future = bootstrap.bind(port).sync(); future.channel().closeFuture().sync(); } finally { // 必须优雅关闭 bossGroup.shutdownGracefully().sync(); workerGroup.shutdownGracefully().sync(); } ``` ### 11.2 异常处理 ```java // 在每个 Handler 中添加 exceptionCaught @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); // 发生异常时关闭连接 } ``` ### 11.3 ByteBuf 使用规范 ```java @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { ByteBuf buf = (ByteBuf) msg; try { // 业务处理 } finally { buf.release(); // 务必释放！ } } ``` *** ## 十二、面试八股文速记版 ### 一句话总结 > **Netty 是基于 Reactor 模式和 Java NIO 的异步事件驱动网络框架，通过 EventLoop 处理 IO 事件、ChannelPipeline 实现责任链、ByteBuf 管理零拷贝内存，提供高性能网络编程能力。** ### 核心知识点速记 | 主题 | 关键点 | |------|--------| | **Reactor 模型** | 主从 Reactor 多线程，Boss 负责 accept，Worker 负责读写 | | **EventLoop** | 一个 EventLoop 对应一个线程，Channel 与 EventLoop 一对一绑定 | | **ChannelPipeline** | 责任链模式，Inbound 从左向右，Outbound 从右向左 | | **ByteBuf** | 独立读写索引、引用计数、堆内外内存、内存池 | | **零拷贝** | CompositeByteBuf、slice、wrap、FileRegion/sendfile | | **粘包拆包** | LengthFieldBasedFrameDecoder、DelimiterBasedFrameDecoder | | **内存泄漏** | ByteBuf 必须手动 release，引用计数归零才释放 | | **线程安全** | 同一 Channel 的所有操作由同一个 EventLoop 线程处理 | | **优雅关闭** | shutdownGracefully() 等待任务完成再关闭线程 | | **性能优化** | 堆外内存、内存池、零拷贝、CPU 亲和性、内核参数调优 | ### 高频面试题 1. **Netty 为什么比 BIO/NIO 快？** * 零拷贝技术、内存池减少 GC、Reactor 模型高效、Epoll 优化 2. **EventLoop 为什么不阻塞？** * 非阻塞 IO + 异步任务队列 + 快速失败机制 3. **如何保证 Channel 线程安全？** * Channel 与 EventLoop 一对一绑定，所有操作在同一线程执行 4. **ByteBuf 为什么要引用计数？** * 精确控制内存释放时机，避免 GC 延迟和内存泄漏 5. **如何解决 TCP 粘包拆包？** * 固定长度、分隔符、长度字段（最常用）、行分割 6. **Netty 内存泄漏怎么排查？** * 检查 ByteBuf 是否 release、使用 jmap 分析 dump、开启 leak detection 7. **BossGroup 为什么可以只有一个线程？** * accept 操作极快，单线程足以处理大量连接请求 *** *本文档基于 Netty 官方文档和社区最佳实践整理，版本参考 Netty 4.x/5.x*