Java NIO 与 Netty

Java NIO

NIO vs BIO

特性	BIO（Blocking IO）	NIO（Non-blocking IO）
模型	同步阻塞	同步非阻塞
线程模型	一连接一线程	少量线程处理多连接
API	Stream 流	Channel + Buffer + Selector
数据	字节流单向	通道双向，缓冲区操作
并发	线程开销大	事件驱动，高并发

核心组件

         ┌────────────────┐
         │    Selector     │  ← 事件选择器
         └────────┬───────┘
                  │ 注册
    ┌─────────────┼─────────────┐
    │             │             │
    ▼             ▼             ▼
 Channel1      Channel2      Channel3  ← 通道（双向）
    │             │             │
    ▼             ▼             ▼
  Buffer        Buffer        Buffer   ← 缓冲区

Buffer（缓冲区）

java

// 分配
ByteBuffer buf = ByteBuffer.allocate(1024);       // 堆内
ByteBuffer directBuf = ByteBuffer.allocateDirect(1024);  // 直接内存（堆外）

// 写入
buf.put("hello".getBytes());

// 切换读模式
buf.flip();

// 读取
while (buf.hasRemaining()) {
    System.out.print((char) buf.get());
}

// 重置
buf.rewind();    // 重新读
buf.clear();     // 清空（写入模式）
buf.compact();   // 压缩（未读数据保留）

Buffer 核心属性：

属性	说明
`capacity`	缓冲区容量
`position`	当前读写位置
`limit`	读写边界
`mark`	标记位置

flip() 前后变化：

写模式： position=5    limit=10    capacity=10
flip() 后：
读模式： position=0    limit=5     capacity=10

直接内存 vs 堆内存：

特性	堆内 Buffer	堆外 DirectBuffer
分配/回收	快	慢
IO 效率	需中间拷贝	零拷贝
适用	小数据量、频繁分配	大数据量、长生命周期

Channel（通道）

java

// 文件通道
try (FileChannel fc = FileChannel.open(Paths.get("input.txt"), StandardOpenOption.READ)) {
    ByteBuffer buf = ByteBuffer.allocate(1024);
    fc.read(buf);
}

// Socket 通道（非阻塞模式）
SocketChannel sc = SocketChannel.open();
sc.configureBlocking(false);  // 非阻塞
sc.connect(new InetSocketAddress("localhost", 8080));

MappedByteBuffer（文件内存映射）：

java

try (FileChannel fc = FileChannel.open(Paths.get("large.dat"), StandardOpenOption.READ)) {
    MappedByteBuffer mbb = fc.map(FileChannel.MapMode.READ_ONLY, 0, fc.size());
    // 直接操作内存，无需 read()
}

Selector（选择器）

java

// 创建 Selector
Selector selector = Selector.open();

// 注册通道
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);
ssc.socket().bind(new InetSocketAddress(8080));
ssc.register(selector, SelectionKey.OP_ACCEPT);  // OP_ACCEPT | OP_READ | OP_WRITE | OP_CONNECT

// 事件循环
while (true) {
    int ready = selector.select(1000);  // 阻塞等待事件
    if (ready == 0) continue;
    
    Set<SelectionKey> keys = selector.selectedKeys();
    Iterator<SelectionKey> it = keys.iterator();
    
    while (it.hasNext()) {
        SelectionKey key = it.next();
        it.remove();
        
        if (key.isAcceptable()) {
            // 接收连接
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            SocketChannel client = server.accept();
            client.configureBlocking(false);
            client.register(selector, SelectionKey.OP_READ);
        }
        if (key.isReadable()) {
            // 读取数据
            SocketChannel client = (SocketChannel) key.channel();
            ByteBuffer buf = ByteBuffer.allocate(1024);
            client.read(buf);
        }
    }
}

Netty

概述

Netty 是高性能 Java 网络框架，封装了 NIO 的复杂性，广泛应用于 RPC、游戏服务器、实时推送等场景。

核心优势：

高性能：零拷贝、Reactor 模型、内存池
易用：简洁的 API、强大的编解码器
稳定：断线重连、心跳、拥塞控制

核心组件

java

// 服务端
EventLoopGroup boss = new NioEventLoopGroup(1);       // 接收连接
EventLoopGroup worker = new NioEventLoopGroup();       // 处理 IO

try {
    ServerBootstrap b = new ServerBootstrap();
    b.group(boss, worker)
     .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 StringDecoder())
               .addLast(new StringEncoder())
               .addLast(new ServerHandler());
         }
     });

    ChannelFuture f = b.bind(8080).sync();
    f.channel().closeFuture().sync();
} finally {
    boss.shutdownGracefully();
    worker.shutdownGracefully();
}

处理器：

java

public class ServerHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        System.out.println("收到: " + msg);
        ctx.writeAndFlush("服务端回复: " + msg);
    }
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("客户端连接: " + ctx.channel().remoteAddress());
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

Netty 线程模型

                          ┌──────────────────────┐
                          │   EventLoopGroup      │
                          │   (bossGroup, 1 线程)  │
                          │   只做 accept          │
                          └──────────┬───────────┘
                                     │
          ┌──────────────────────────┼──────────────────────┐
          │          │          │          │          │
          ▼          ▼          ▼          ▼          ▼
   ┌───────────────────────────────────────────────────────┐
   │                EventLoopGroup (worker)                │
   │   EventLoop 1      EventLoop 2      EventLoop N      │
   │   Socket          Socket            Socket           │
   │   Channel           Channel           Channel        │
   └───────────────────────────────────────────────────────┘

一个 EventLoop 绑定一个固定线程和多个 Channel
一个 Channel 的所有 IO 操作由同一个 EventLoop 处理（线程安全）

粘包拆包与编解码器

粘包原因：TCP 是流协议，没有消息边界

Netty 内置的解码器：

解码器	说明	适用
`LineBasedFrameDecoder`	按换行符 \n 分割	文本协议
`DelimiterBasedFrameDecoder`	按自定义分隔符	简单协议
`FixedLengthFrameDecoder`	固定长度	定长协议
`LengthFieldBasedFrameDecoder`	长度字段在数据中	最常用

LengthFieldBasedFrameDecoder 示例：

java

ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(
    1024,       // maxFrameLength 最大帧长
    0,          // lengthFieldOffset 长度字段偏移
    4,          // lengthFieldLength 长度字段字节数
    0,          // lengthAdjustment 补偿
    4           // initialBytesToStrip 跳过的字节数
));

自定义协议实战

协议格式：

魔数(4B) | 版本(1B) | 序列化(1B) | 指令(2B) | 长度(4B) | 数据(NB)

编码器：

java

public class MyEncoder extends MessageToByteEncoder<MyMessage> {
    @Override
    protected void encode(ChannelHandlerContext ctx, MyMessage msg, ByteBuf out) {
        out.writeInt(0xCAFEBABE);      // 魔数
        out.writeByte(1);               // 版本
        out.writeByte(msg.serializeType());
        out.writeShort(msg.command());
        out.writeInt(msg.data().length);
        out.writeBytes(msg.data());
    }
}

解码器：

java

public class MyDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        if (in.readableBytes() < 12) return;           // 不够头部
        in.markReaderIndex();
        if (in.readInt() != 0xCAFEBABE) {               // 魔数校验
            ctx.close();
            return;
        }
        byte version = in.readByte();
        byte serialize = in.readByte();
        short command = in.readShort();
        int length = in.readInt();
        if (in.readableBytes() < length) {
            in.resetReaderIndex();                       // 不够，重置
            return;
        }
        byte[] data = new byte[length];
        in.readBytes(data);
        out.add(new MyMessage(version, serialize, command, data));
    }
}

客户端示例

java

EventLoopGroup group = new NioEventLoopGroup();
try {
    Bootstrap b = new Bootstrap();
    b.group(group)
     .channel(NioSocketChannel.class)
     .handler(new ChannelInitializer<SocketChannel>() {
         @Override
         protected void initChannel(SocketChannel ch) {
             ch.pipeline().addLast(new StringEncoder(), new ClientHandler());
         }
     });

    Channel ch = b.connect("localhost", 8080).sync().channel();
    ch.writeAndFlush("Hello Netty");
    ch.closeFuture().sync();
} finally {
    group.shutdownGracefully();
}

零拷贝

机制	说明
CompositeByteBuf	合并多个 Buffer 避免拷贝
Wrap/FastThreadLocal	减少线程局部变量开销
FileRegion	`FileChannel.transferTo()` 零拷贝传输文件
DirectBuffer	堆外内存，避免堆内→堆外拷贝

java

// FileRegion 零拷贝
FileRegion region = new DefaultFileRegion(
    new FileInputStream("large.zip").getChannel(), 0, fileSize);
ctx.writeAndFlush(region);

Netty 在 RPC 框架中的应用

服务端 Netty ← [序列化/反序列化] ← [自定义协议] → 客户端 Netty
                           ↕
                     [服务路由/负载均衡]

Dubbo：使用 Netty 作为默认通信层
Spring Cloud Gateway：基于 Netty 的 Reactor 框架
gRPC：底层 HTTP/2 基于 Netty

最佳实践

合理配置线程数：boss=1（默认），worker=CPU 核数×2
内存池化：-Dio.netty.allocator.type=pooled
Pipeline 顺序：解码器 → 业务处理器 → 编码器
业务处理不要阻塞 EventLoop：长耗时操作放到业务线程池
配置 WriteBufferWaterMark：控制写缓冲区水位线
添加 IdleStateHandler：心跳检测
Channel 复用：channel 可重用，避免频繁创建

Java NIO 与 Netty ​

Java NIO ​

NIO vs BIO ​

核心组件 ​

Buffer（缓冲区） ​

Channel（通道） ​

Selector（选择器） ​

Netty ​

概述 ​

核心组件 ​

Netty 线程模型 ​

粘包拆包与编解码器 ​

自定义协议实战 ​

客户端示例 ​

零拷贝 ​

Netty 在 RPC 框架中的应用 ​

最佳实践 ​

Java NIO 与 Netty

Java NIO

NIO vs BIO

核心组件

Buffer（缓冲区）

Channel（通道）

Selector（选择器）

Netty

概述

核心组件

Netty 线程模型

粘包拆包与编解码器

自定义协议实战

客户端示例

零拷贝

Netty 在 RPC 框架中的应用

最佳实践