network

Java NIO/AIO 与网络编程

掌握 Java NIO 的 Channel、Buffer 与 Selector 核心组件，理解非阻塞 I/O 与异步 I/O 的演进与选型

Leeting Yan 7 分钟阅读 3399 字

Java NIO（New I/O，非阻塞 I/O）在 JDK 1.4 引入，AIO（Asynchronous I/O）在 JDK 1.7 引入。理解这些 I/O 模型是掌握 Netty 等高性能框架的基础。

一、I/O 模型对比

1.1 五种 I/O 模型

模型阻塞点Java 支持特点
阻塞 I/O（BIO）全程阻塞JDK 1.0一个连接一个线程
非阻塞 I/O（NIO）无JDK 1.4多路复用，单线程处理多连接
I/O 多路复用select/poll/epollJDK 1.4Selector 机制
信号驱动 I/O无无原生支持Linux 信号机制
异步 I/O（AIO）完全无阻塞JDK 1.7内核完成 I/O 后回调

1.2 形象比喻

阻塞 I/O（BIO）：
  你去餐厅，点餐后站在柜台等，直到餐做好才离开（阻塞）

非阻塞 I/O（NIO）：
  你去餐厅，点餐后拿个号牌，每隔几分钟去问一次好了没（轮询）

I/O 多路复用：
  你去餐厅，点餐后坐在座位上，服务员做好后叫你的号牌（事件通知）

异步 I/O（AIO）：
  你打电话点餐，商家做好后送到你家（完全异步， callback）

二、NIO 三大核心组件

2.1 Channel（通道）

// Channel 是双向的，区别于 Stream 的单向
// 主要实现：

// 文件通道
FileChannel fileChannel = new RandomAccessFile("data.txt", "rw").getChannel();

// 网络通道
ServerSocketChannel serverChannel = ServerSocketChannel.open();
SocketChannel socketChannel = SocketChannel.open();

// UDP 通道
DatagramChannel datagramChannel = DatagramChannel.open();

2.2 Buffer（缓冲区）

// NIO 数据读写必须通过 Buffer

// 创建 Buffer
ByteBuffer buffer = ByteBuffer.allocate(1024);        // 堆内存
ByteBuffer directBuffer = ByteBuffer.allocateDirect(1024);  // 直接内存

// Buffer 状态属性
// position: 当前读写位置
// limit: 可以读写的边界
// capacity: 最大容量

// 写入数据（写模式）
buffer.put("Hello".getBytes());
buffer.putInt(123);

// 切换为读模式
buffer.flip();
// flip()：limit = position, position = 0

// 读取数据
byte[] data = new byte[buffer.remaining()];
buffer.get(data);
System.out.println(new String(data));

// 清空，重新写入
buffer.clear();
// clear()：position = 0, limit = capacity

// 或者 compact()：将未读数据移到头部
buffer.compact();

2.3 Selector（多路复用器）

// Selector 是 NIO 的核心，允许单线程监控多个 Channel

Selector selector = Selector.open();

// 注册 Channel 到 Selector
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.bind(new InetSocketAddress(8080));
serverChannel.configureBlocking(false);  // 必须非阻塞
serverChannel.register(selector, SelectionKey.OP_ACCEPT);

// 事件循环
while (true) {
    // 阻塞等待就绪事件（有超时版本）
    int readyCount = selector.select();
    
    if (readyCount == 0) continue;
    
    // 获取就绪的 SelectionKey
    Set<SelectionKey> readyKeys = selector.selectedKeys();
    Iterator<SelectionKey> iterator = readyKeys.iterator();
    
    while (iterator.hasNext()) {
        SelectionKey key = iterator.next();
        iterator.remove();  // 必须移除，否则重复处理
        
        if (key.isAcceptable()) {
            // 有新的连接请求
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            SocketChannel client = server.accept();
            client.configureBlocking(false);
            client.register(selector, SelectionKey.OP_READ);
            System.out.println("Client connected: " + client.getRemoteAddress());
        }
        
        if (key.isReadable()) {
            // 有数据可读
            SocketChannel client = (SocketChannel) key.channel();
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            int read = client.read(buffer);
            
            if (read == -1) {
                // 连接关闭
                key.cancel();
                client.close();
            } else if (read > 0) {
                buffer.flip();
                // 处理数据...
                System.out.println("Received: " + new String(buffer.array(), 0, read));
                
                // 注册写事件（响应客户端）
                key.interestOps(SelectionKey.OP_WRITE);
            }
        }
        
        if (key.isWritable()) {
            // 可以写入数据
            SocketChannel client = (SocketChannel) key.channel();
            ByteBuffer response = ByteBuffer.wrap("OK\n".getBytes());
            client.write(response);
            
            // 写完后改回读事件
            key.interestOps(SelectionKey.OP_READ);
        }
    }
}

三、完整 NIO Server 示例

public class NioEchoServer {
    
    private Selector selector;
    private ServerSocketChannel serverChannel;
    private ByteBuffer buffer = ByteBuffer.allocate(1024);
    
    public void start(int port) throws IOException {
        selector = Selector.open();
        
        serverChannel = ServerSocketChannel.open();
        serverChannel.bind(new InetSocketAddress(port));
        serverChannel.configureBlocking(false);
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        
        System.out.println("NIO Server started on port " + port);
        
        while (true) {
            selector.select();
            
            Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
            while (keys.hasNext()) {
                SelectionKey key = keys.next();
                keys.remove();
                
                if (key.isAcceptable()) {
                    handleAccept(key);
                } else if (key.isReadable()) {
                    handleRead(key);
                } else if (key.isWritable()) {
                    handleWrite(key);
                }
            }
        }
    }
    
    private void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel server = (ServerSocketChannel) key.channel();
        SocketChannel client = server.accept();
        client.configureBlocking(false);
        client.register(selector, SelectionKey.OP_READ);
        System.out.println("Accepted connection from " + client.getRemoteAddress());
    }
    
    private void handleRead(SelectionKey key) throws IOException {
        SocketChannel client = (SocketChannel) key.channel();
        buffer.clear();
        
        int read = client.read(buffer);
        if (read == -1) {
            System.out.println("Client disconnected: " + client.getRemoteAddress());
            key.cancel();
            client.close();
            return;
        }
        
        buffer.flip();
        byte[] data = new byte[buffer.remaining()];
        buffer.get(data);
        String message = new String(data);
        System.out.println("Received from " + client.getRemoteAddress() + ": " + message.trim());
        
        // 准备响应
        key.attach("Echo: " + message);
        key.interestOps(SelectionKey.OP_WRITE);
    }
    
    private void handleWrite(SelectionKey key) throws IOException {
        SocketChannel client = (SocketChannel) key.channel();
        String response = (String) key.attachment();
        
        ByteBuffer buf = ByteBuffer.wrap(response.getBytes());
        client.write(buf);
        
        // 写完后继续监听读取
        key.interestOps(SelectionKey.OP_READ);
        key.attach(null);
    }
    
    public static void main(String[] args) throws IOException {
        new NioEchoServer().start(8080);
    }
}

四、AIO（NIO.2）

4.1 异步通道

public class AioEchoServer {
    
    public static void main(String[] args) throws IOException, InterruptedException {
        AsynchronousServerSocketChannel server = 
            AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(8080));
        
        System.out.println("AIO Server started on port 8080");
        
        // 异步接受连接（回调方式）
        server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {
            @Override
            public void completed(AsynchronousSocketChannel client, Void attachment) {
                // 继续接受下一个连接
                server.accept(null, this);
                
                // 处理当前连接
                handleClient(client);
            }
            
            @Override
            public void failed(Throwable exc, Void attachment) {
                exc.printStackTrace();
            }
        });
        
        // 主线程阻塞，等待异步操作
        Thread.sleep(Long.MAX_VALUE);
    }
    
    private static void handleClient(AsynchronousSocketChannel client) {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        
        // 异步读取
        client.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer result, ByteBuffer buffer) {
                if (result == -1) {
                    try {
                        client.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    return;
                }
                
                buffer.flip();
                byte[] data = new byte[buffer.remaining()];
                buffer.get(data);
                String message = new String(data);
                System.out.println("Received: " + message.trim());
                
                // 异步写入响应
                ByteBuffer response = ByteBuffer.wrap(("Echo: " + message).getBytes());
                client.write(response, null, new CompletionHandler<Integer, Void>() {
                    @Override
                    public void completed(Integer result, Void attachment) {
                        // 继续读取
                        ByteBuffer newBuffer = ByteBuffer.allocate(1024);
                        client.read(newBuffer, newBuffer, this);
                    }
                    
                    @Override
                    public void failed(Throwable exc, Void attachment) {
                        exc.printStackTrace();
                    }
                });
            }
            
            @Override
            public void failed(Throwable exc, ByteBuffer buffer) {
                exc.printStackTrace();
            }
        });
    }
}

4.2 Future 方式

// Future 方式（阻塞获取结果，不推荐）
Future<AsynchronousSocketChannel> future = server.accept();
AsynchronousSocketChannel client = future.get();  // 阻塞

// 更好的方式：使用 CompletionHandler（回调）
// 或使用 CompletableFuture 包装

五、NIO vs AIO 对比

特性NIOAIO
模型多路复用（Reactor）真正异步（Proactor）
阻塞非阻塞（需轮询 Selector）完全非阻塞
线程应用处理 I/O内核处理 I/O，应用接收回调
实现复杂度较复杂较简单
Linux 实现epollio_uring（新）/ aio（旧，有限制）
成熟度高（Netty 基于此）中
适用场景高并发网络服务文件 I/O 为主

六、Channel 类型速查

// 文件通道（NIO）
FileChannel fc = FileChannel.open(Path.of("file.txt"), StandardOpenOption.READ);
MappedByteBuffer mapped = fc.map(FileChannel.MapMode.READ_ONLY, 0, fc.size());  // 内存映射

// Socket 通道
SocketChannel sc = SocketChannel.open(new InetSocketAddress("localhost", 80));
sc.configureBlocking(false);

// ServerSocket 通道
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.bind(new InetSocketAddress(8080));

// 文件传输（零拷贝）
FileChannel src = new FileInputStream("src.txt").getChannel();
FileChannel dst = new FileOutputStream("dst.txt").getChannel();
src.transferTo(0, src.size(), dst);  // 零拷贝传输

七、零拷贝与内存映射文件

零拷贝（Zero-Copy）是指数据在 kernel space 和 user space 之间的拷贝次数 minimized。传统 I/O 的瓶颈不在网络带宽，而在内存拷贝。

传统文件发送的 4 次拷贝

磁盘 → Kernel Buffer → User Buffer → Kernel Socket Buffer → 网卡
     ① DMA拷贝          ② CPU拷贝      ③ CPU拷贝          ④ DMA拷贝

NIO 零拷贝（transferTo）的 2 次拷贝

// FileChannel.transferTo：内核态直接转发，无需用户态中转
FileChannel src = new FileInputStream("large.zip").getChannel();
SocketChannel dst = SocketChannel.open(new InetSocketAddress("remote", 8080));

src.transferTo(0, src.size(), dst);

// 实际路径：磁盘 → Kernel PageCache → Kernel Socket Buffer → 网卡
// 拷贝次数：① DMA → ② DMA（零 CPU 拷贝）
技术拷贝次数CPU 参与适用场景
传统 I/O4 次2 次小文件、需加工
transferTo/From2 次0 次大文件传输、静态资源
mmap + write2 次1 次随机访问 + 网络发送

内存映射文件（Memory-Mapped File）

FileChannel fc = new RandomAccessFile("data.bin", "rw").getChannel();
MappedByteBuffer mapped = fc.map(
    FileChannel.MapMode.READ_WRITE,  // 或 READ_ONLY
    0,                               // 起始偏移
    fc.size()                        // 映射大小
);

// 像操作数组一样读写文件（实际由 OS 页缓存管理）
mapped.putInt(0, 42);
mapped.putLong(4, System.currentTimeMillis());

// 强制刷盘（否则依赖 OS 刷页策略）
mapped.force();

MappedByteBuffer 的优势：

  • 大文件处理无需载入完整内容到 JVM 堆（避免 GC 压力）
  • 进程间共享内存（多个 JVM 可映射同一文件）
  • 与零拷贝结合可实现极致的文件传输性能

注意限制：映射区域大小受 Integer.MAX_VALUE（约 2GB）限制，超大文件需分片映射。

八、DirectBuffer vs HeapBuffer

特性HeapByteBufferDirectByteBuffer
内存位置JVM 堆内堆外（Native Memory）
创建速度快慢（需向 OS 申请）
I/O 效率低（需额外 JNI 拷贝到原生内存）高（直接传递给内核）
GC 影响受 GC 管理不受 GC，通过 Cleaner 释放
适用场景小数据量、频繁创建销毁网络 I/O、大缓冲、长生命周期
// Heap Buffer：适合临时小数据
ByteBuffer heapBuf = ByteBuffer.allocate(1024);

// Direct Buffer：适合 Socket I/O
ByteBuffer directBuf = ByteBuffer.allocateDirect(64 * 1024);

// 堆外内存大小限制（默认 ≈ 最大堆内存）
// 修改：-XX:MaxDirectMemorySize=512M

Netty 默认使用 DirectBuffer 作为网络缓冲，通过对象池（Recycler）复用 Buffer 避免频繁分配和释放的开销。

九、epoll 底层原理（Linux）

Java NIO 在 Linux 下通过 EPollSelectorImpl 实现，本质上是对 Linux epoll 的封装。

e_poll 相比传统 select/poll 的优势：

维度select/pollepoll
描述符限制select: 1024；poll: 无硬限制但遍历慢无限制（仅受系统内存）
事件检测每次调用遍历所有 fd（O(n)）回调通知就绪 fd（O(1)）
数据拷贝每次调用拷贝 fd 集合到内核epoll_ctl 注册一次，epoll_wait 不拷贝
水平触发支持支持（默认）
边缘触发不支持支持（EPOLLET）
// epoll API 简化示意
int epfd = epoll_create1(0);
epoll_ctl(epfd, EPOLL_CTL_ADD, socket_fd, &event);  // 注册一次
epoll_wait(epfd, events, maxevents, timeout);        // 等待就绪（O(1)）

Java NIO 使用水平触发（Level-Triggered）：只要 fd 就绪，每次 select() 都会返回。边缘触发（Edge-Triggered）只在状态变化时通知一次，需一次性读完数据，Netty 的 Epoll 模式支持 ET。

十、Netty 与 NIO 的关系

Netty 不是替代 NIO，而是基于 NIO 构建的高性能网络框架：

特性Java NIONetty
线程模型需自行设计 Reactor内置主从 Reactor（Boss/Worker）
内存管理手动 Buffer 管理ByteBuf 对象池 + 引用计数
编解码手写编解码逻辑内置 LengthFieldBasedFrameDecoder 等
事件传播手动处理 SelectionKeyPipeline + ChannelHandler 链式处理
优雅关闭需自行处理内置优雅关闭（Graceful Shutdown）
性能万级并发百万级并发（经优化后）

学习路径建议：先掌握 Java NIO 三大组件 → 理解 Reactor 模式 → 学习 Netty 的线程模型和 Pipeline → 深入研究 Netty 内存管理和对象池。

十一、CPU 亲和性与线程绑定

极端性能场景下，可将 NIO 的 Selector 线程绑定到特定 CPU 核心，减少上下文切换和缓存失效：

// 通过 JNI 调用 sched_setaffinity（Linux）
// 或使用 Netty 的 EventLoopGroup 配合 isolcpus 内核参数

EventLoopGroup bossGroup = new NioEventLoopGroup(1);   // 绑定核心 0
EventLoopGroup workerGroup = new NioEventLoopGroup(4); // 绑定核心 1-4

# 启动时隔离 CPU（grub 配置）
isolcpus=1,2,3,4  # 将核心 1-4 隔离，仅供应用使用

此优化仅在高频交易、游戏服务器等微秒级延迟敏感场景必要，常规应用无需考虑。

七、总结

组件职责关键 API
ChannelI/O 通道，双向传输open(), configureBlocking(), register()
Buffer数据容器allocate(), flip(), clear(), compact()
Selector多路复用器open(), select(), selectedKeys()
SelectionKey注册关系与事件interestOps(), readyOps(), attach()

Java I/O 模型演进路线：BIO（阻塞）→ NIO（多路复用）→ AIO（异步回调）→ Netty（框架封装）。目前生产环境中，NIO + Netty 是高并发网络服务的标准组合；AIO 因 Linux 下 aio 不完善、io_uring 尚未被 JDK 原生支持，实际使用较少。

理解零拷贝、DirectBuffer、epoll 原理后，才能在实际调优中做出正确决策——比如为文件传输服务启用 transferTo，为 RPC 框架选择 HeapBuffer 减少 GC 压力，为延迟敏感系统配置 CPU 亲和性。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「network」更多文章

  1. 46. 排队论与容量估算：利特尔法则与尾延迟
  2. MPTCP 与多路径传输
  3. 流量整形与 TC/qdisc