全栈攻城狮

Appearance

ByteBuffer / Channel NIO 核心源码精读 ​

概述 ​

NIO 的三大件：Buffer（数据载体）、Channel（传输管道）、Selector（多路复用）。其中 ByteBuffer 的堆/直接内存区分直接决定 IO 性能，SocketChannel/ServerSocketChannel 是网络通信的基础，Selector 则是事件驱动模型的开关。本文基于 OpenJDK 21 源码拆解 Buffer 与 Channel 的实现。

一、ByteBuffer 的四个核心指针 ​

java
// java.nio.Buffer
public abstract class Buffer {
    private int mark = -1;       // 标记位（可选）
    private int position = 0;    // 当前位置（下一个读写索引）
    private int limit;           // 读写上限
    private int capacity;        // 容量（创建时固定，不可变）
}

四个指针的不变式：0 ≤ mark ≤ position ≤ limit ≤ capacity

生命周期流转（以写→读为例）：
  初始：        pos=0      limit=cap      → 可写区域 [pos, limit)
  put 写入：    pos 前移                   → 记录写入了多少
  flip()：      limit=pos, pos=0          → 切换为读模式（读 [0, limit)）
  get 读取：    pos 前移
  clear()：     pos=0, limit=cap          → 切回写模式
  rewind()：    pos=0, limit 不变         → 重读
  mark()/reset()：保存/恢复位置
java
// 典型用法：读写切换
ByteBuffer buf = ByteBuffer.allocate(1024);
buf.put("hello".getBytes());     // 写
buf.flip();                      // 切换读模式
byte[] out = new byte[buf.remaining()];
buf.get(out);                    // 读

二、HeapByteBuffer 与 DirectByteBuffer ​

2.1 HeapByteBuffer.allocate(int) 堆内内存 ​

java
// java.nio.ByteBuffer
public static ByteBuffer allocate(int capacity) {
    if (capacity < 0) throw new IllegalArgumentException();
    return new HeapByteBuffer(capacity, capacity);   // 堆内数组实现
}

// java.nio.HeapByteBuffer
HeapByteBuffer(int cap, int lim) {
    super(-1, 0, lim, cap, new byte[cap], 0);        // 底层是 byte[] 数组
}
HeapByteBuffer 特点：
  底层：JVM 堆内的 byte[] 数组 → GC 管理、可被移动
  优点：创建便宜、访问快（无需跨 JNI）
  缺点：做系统调用时必须拷贝到直接内存（见 FileChannel 章节）

2.2 DirectByteBuffer.allocateDirect(int) 直接内存 ​

java
public static ByteBuffer allocateDirect(int capacity) {
    return new DirectByteBuffer(capacity);
}

// java.nio.DirectByteBuffer
DirectByteBuffer(int cap) {
    super(-1, 0, cap, cap);
    boolean pa = VM.isDirectMemoryPageAligned();
    int ps = Bits.pageSize();
    long size = Math.max(1L, (long)cap + (pa ? ps : 0));
    // ① 从直接内存预算中扣除（受 -XX:MaxDirectMemorySize 限制）
    Bits.reserveMemory(size, cap);
    long base = 0;
    try {
        base = unsafe.allocateMemory(size);          // ② Unsafe 分配堆外内存
    } catch (OutOfMemoryError x) {
        Bits.unreserveMemory(size, cap);
        throw x;
    }
    unsafe.setMemory(base, size, (byte) 0);          // ③ 清零
    if (pa && (base % ps != 0)) {
        // 页对齐处理：地址修正
        address = base + ps - (base & (ps - 1));
    } else {
        address = base;
    }
    // ④ 注册 Cleaner：GC 时回收堆外内存
    cleaner = Cleaner.create(this, new Deallocator(base, size, cap));
}
DirectByteBuffer 关键点：
  分配：unsafe.allocateMemory → 堆外内存（不受堆大小限制，受 MaxDirectMemorySize）
  限额：Bits.reserveMemory 全局记账，超限触发 Full GC 尝试回收
  回收：Cleaner（虚引用）→ GC 时 Deallocator.run() → unsafe.freeMemory
  地址：address 字段直接指向堆外内存 → 系统调用无需中转拷贝
java
// 清理逻辑（Cleaner 关联的 Deallocator）
private static class Deallocator implements Runnable {
    public void run() {
        unsafe.freeMemory(address);     // 释放堆外内存
        Bits.unreserveMemory(size, capacity);   // 归还直接内存预算
    }
}

DirectByteBuffer 不释放的风险：堆外内存不受 GC 直接管理，若忘记回收且引用长期存活，会耗尽直接内存（报 OutOfMemoryError: Direct buffer memory）。

三、SocketChannel 连接 ​

3.1 open() 与 connect() ​

java
public static SocketChannel open() throws IOException {
    return SelectorProvider.provider().openSocketChannel();   // 平台默认实现
}

// Linux 默认：sun.nio.ch.SelectorProviderImpl.openSocketChannel
// → new SocketChannelImpl(provider, true)
java
// sun.nio.ch.SocketChannelImpl
public boolean connect(SocketAddress sa) throws IOException {
    ...
    // ① 解析地址 → InetSocketAddress
    // ② 尝试非阻塞 connect0
    int n = Net.connect(fd, isa, blocking ? -1 : 0);
    ...
    // ③ 非阻塞模式：未完成 → 返回 false（等 OP_CONNECT 事件）
    // ④ 完成 → 走 finishConnect 校验
}

// Net.connect → connect0() native：
//   Linux   → connect(fd, sockaddr, len)
//   Windows → WSAConnect(fd, ...)
connect 语义：
  阻塞模式：connect0 阻塞直到建立连接或失败
  非阻塞模式：立即返回；未连接完成 → 注册 OP_CONNECT 等事件
  完成后：finishConnect 确认（检查 SO_ERROR，失败抛 ConnectException）

3.2 read / write 的数据通道 ​

java
public int read(ByteBuffer dst) throws IOException {
    // IOUtil.read(fd, dst, -1, nd) → read0 native → recv(fd, ...)
}
public int write(ByteBuffer src) throws IOException {
    // IOUtil.write(fd, src, -1, nd) → write0 native → send(fd, ...)
}

四、ServerSocketChannel.accept() ​

java
public SocketChannel accept() throws IOException {
    // sun.nio.ch.ServerSocketChannelImpl
    synchronized (stateLock) {
        ...
        SocketChannel sc = null;
        int n = 0;
        if (isOpen()) {
            // 阻塞模式循环尝试 accept
            do {
                n = Net.accept(fd, newfd, isaa);
            } while (n == IOStatus.UNAVAILABLE && isOpen());
            if (IOStatus.isAllocating(n)) { ... }
        }
        ...
        if (sc == null) throw new ClosedChannelException();
        return sc;                    // 新连接封装成 SocketChannelImpl
    }
}

// Net.accept → accept0() native：
//   Linux   → accept4(fd, sockaddr, ..., SOCK_CLOEXEC)
//   Windows → WSAAccept(fd, ...)
accept 流程：
  ① 内核 accept 系统调用取出连接队列中的连接
  ② 无连接且阻塞模式 → 阻塞等待
  ③ 拿到新 fd → 包装成 SocketChannelImpl（独立于 ServerSocketChannel）
  ④ 非阻塞模式 → 未就绪返回 null（配合 OP_ACCEPT 事件）

五、Selector 的平台实现选择 ​

java
public static Selector open() throws IOException {
    return SelectorProvider.provider().openSelector();
}

// sun.nio.ch.DefaultSelectorProvider（Linux）
public static SelectorProvider create() {
    // 启动属性配置：java.nio.channels.spi.SelectorProvider
    // 无配置 → 按平台返回
    if (usePollSelector()) return PollSelectorProvider.create();   // poll 模式
    return EPollSelectorProvider.create();                         // epoll 模式
}
Selector 平台实现：
  Linux   → EPollSelectorImpl（epoll_create / epoll_wait）
            （早期 JDK 用 PollSelectorImpl，后默认切 epoll）
  Windows → WindowsSelectorImpl（WSAPoll / WaitForMultipleObjects）
  macOS   → KQueueSelectorImpl（kqueue）

六、SelectionKey 事件与 Channels 适配 ​

6.1 4 种事件 ​

java
// java.nio.channels.SelectionKey
public static final int OP_READ    = 1 << 0;   // 读就绪（1）
public static final int OP_WRITE   = 1 << 2;   // 写就绪（4）
public static final int OP_CONNECT = 1 << 3;   // 连接就绪（8）
public static final int OP_ACCEPT  = 1 << 4;   // 接受就绪（16）

public abstract int interestOps();   // 关注的事件集合（位掩码）
public abstract int readyOps();      // 就绪的事件集合
public abstract boolean isReadable() { return readyOps() & OP_READ != 0; }
事件语义（与内核事件映射，epoll）：
  OP_READ    ← EPOLLIN（读缓冲有数据）
  OP_WRITE   ← EPOLLOUT（写缓冲可写）
  OP_CONNECT ← EPOLLOUT 辅助（连接完成检测）
  OP_ACCEPT  ← EPOLLIN（监听队列有连接）
java
// 用法示例：注册关注读事件
SelectionKey key = socketChannel.register(selector, SelectionKey.OP_READ);

6.2 Channels 流式适配 ​

Channels 把 Channel 转回传统流式 API，桥接新旧 IO：

java
// java.nio.channels.Channels
public static InputStream newInputStream(ReadableByteChannel ch) {
    return new ChannelInputStream(ch);
}

// 内部类 ChannelInputStream
static final class ChannelInputStream extends InputStream {
    public synchronized int read() throws IOException {
        // 每次读 1 字节：临时 1 字节 buffer
        return ch.read(ByteBuffer.allocate(1)) > 0 ? byteAt : -1;
    }
    public synchronized int read(byte[] bs, int off, int len) {
        ByteBuffer bb = ByteBuffer.wrap(bs, off, len);   // 包装数组
        return ch.read(bb);                              // 直接通道读
    }
}
Channels 提供的适配：
  newInputStream(ReadableByteChannel)   → InputStream
  newOutputStream(WritableByteChannel)  → OutputStream
  newChannel(InputStream)               → ReadableByteChannel
  newChannel(OutputStream)              → WritableByteChannel

七、实现要点 ​

Buffer / Channel 核心：
  ByteBuffer：position/limit/capacity/mark 四指针不变式
  HeapByteBuffer：堆内 byte[]，访问快但系统调用需中转
  DirectByteBuffer：unsafe.allocateMemory + Cleaner 回收，地址直达内核
  SocketChannel：connect0/recv/send native；非阻塞等 OP_CONNECT
  ServerSocketChannel：accept0 取出新连接包装成 SocketChannelImpl
  Selector：平台实现 EPoll/Windows/Poll/KQueue
  SelectionKey：OP_READ/OP_WRITE/OP_CONNECT/OP_ACCEPT 位掩码
  Channels：Channel ↔ 流式 API 双向适配

常见陷阱：
  flip 忘记调用 → 读模式错乱（position/limit 混乱）
  DirectByteBuffer 泄漏 → 直接内存耗尽（用 Cleaner/池化）
  每次 allocate(1) 读单字节 → 性能极差（用批量读）
  非阻塞 write 可能部分写入 → 需要循环直到写完
  selector.select 返回 0 空转 → 检查注册事件是否有效