AsynchronousFileChannel / AsynchronousSocketChannel AIO 源码精读

概述

AIO（Asynchronous IO，JDK 7 引入）是 Java 的真异步 IO：发起操作立即返回，完成时由**回调（CompletionHandler）或 Future**通知，全程不阻塞调用线程。NIO（Selector）是"多路复用 + 就绪通知"，而 AIO 是"内核完成通知"——Linux 上基于线程池模拟，Windows 上基于 IOCP 原生实现。本文基于 OpenJDK 21 源码拆解。

---

一、AsynchronousFileChannel.open() 的平台实现

// java.nio.channels.AsynchronousFileChannel
public static AsynchronousFileChannel open(Path file, Set<? extends OpenOption> options,
                                           ExecutorService executor, FileAttribute<?>... attrs) {
    // 平台默认实现选择
    return newFileChannel(file, options, attrs, executor);
}

// sun.nio.ch.AsynchronousChannelProvider（Linux）
public AsynchronousChannelGroup openAsynchronousChannelGroup(...) { ... }
public AsynchronousFileChannel newAsynchronousFileChannel(...) {
    return new UnixAsynchronousFileChannelImpl(...);
}

平台差异：
  Linux   → UnixAsynchronousFileChannelImpl
            用线程池 + pread/pwrite（提交任务到线程池执行，模拟异步）
  Windows → WindowsAsynchronousFileChannelImpl
            用 ReadFileEx + IOCP（内核级异步完成端口，真异步）

1.1 read 的 AIO 读（Linux 视角）

// sun.nio.ch.UnixAsynchronousFileChannelImpl
public <A> void read(ByteBuffer dst, long position, A attachment,
                     CompletionHandler<Integer,? super A> handler) {
    ...
    implRead(dst, position, attachment, handler);   // 核心
}

// 内部用 Future 封装完成状态
private <A> PendingFuture<Integer,A> implRead(...) {
    // ① 创建 PendingFuture（Future + 回调载体）
    // ② 提交到通道组线程池执行
    // ③ 完成后在回调线程执行 handler.completed(result, attachment)
}

Linux AIO 读流程：
  1. read() 调用 → 创建 PendingFuture 包装本次操作
  2. 提交到 AsynchronousChannelGroup 的线程池
  3. 线程池线程执行 pread(fd, buf, len, pos)
  4. 完成后设置 Future 结果 → 调用 handler.completed() 回调
  → 调用线程立即返回，不阻塞

1.2 Windows 的 IOCP 实现

// sun.nio.ch.WindowsAsynchronousFileChannelImpl
// 关键系统调用：
//   CreateIoCompletionPort(hFile, hIocp, ...)  → 把文件句柄绑定到完成端口
//   ReadFileEx(hFile, buf, len, overlapped, ...) → 发起异步读
//   GetQueuedCompletionStatus(hIocp, ...)       → 等待完成事件

Windows IOCP 异步读流程：
  1. 打开文件 → CreateIoCompletionPort 绑定 IOCP
  2. ReadFileEx 发起异步读（内核自行完成）
  3. 线程池线程 GetQueuedCompletionStatus 等待完成包
  4. 完成包到达 → 取出结果 → 调用 handler.completed()
  → 全程无用户态阻塞，内核完成通知

---

二、CompletionHandler 回调线程模型

// java.nio.channels.CompletionHandler
public interface CompletionHandler<V,A> {
    void completed(V result, A attachment);   // 成功回调
    void failed(Throwable exc, A attachment); // 失败回调
}

回调在哪个线程执行由 AsynchronousChannelGroup 决定：

回调执行线程：
  Linux：UnixAsynchronousChannelGroup 的线程池线程
  Windows：IOCP 完成线程（绑定在通道组的线程池中）
  默认：不传 group 时，用系统默认的 AsynchronousChannelGroup
        （所有异步通道共享的全局组）

注意：
  回调中不能做耗时操作（占住完成线程 → 阻塞后续完成事件分发）
  回调线程不是发起调用的线程 → 注意线程安全问题

// 用法示例
channel.read(buffer, 0L, attachment, new CompletionHandler<Integer, Object>() {
    public void completed(Integer result, Object attachment) {
        // 在完成线程执行（非调用线程）
    }
    public void failed(Throwable exc, Object attachment) { }
});

---

三、AsynchronousChannelGroup.withFixedThreadPool

public static AsynchronousChannelGroup withFixedThreadPool(int nThreads, ThreadFactory threadFactory)
        throws IOException {
    return new FixedThreadPoolGroup(nThreads, threadFactory);
}

// 内部（Linux）
class FixedThreadPoolGroup extends AsynchronousChannelGroupImpl {
    FixedThreadPoolGroup(int nThreads, ThreadFactory threadFactory) {
        // ThreadPoolExecutor：固定线程数，处理 IO 完成事件
        executor = new ThreadPoolExecutor(nThreads, nThreads,
            Long.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
    }
}

线程池组的作用：
  ① 执行 IO 操作（Linux 的 pread 等）
  ② 分发完成回调（handler.completed）
  ③ 通道之间共享 → 控制总线程数
  vs 默认组：系统创建的共享组（线程数 = 核数）

---

四、AsynchronousSocketChannel 连接

public static AsynchronousSocketChannel open() throws IOException {
    return provider().openAsynchronousSocketChannel();
}

// sun.nio.ch.UnixAsynchronousSocketChannelImpl
public <A> void connect(SocketAddress remote, A attachment,
                        CompletionHandler<Void,? super A> handler) {
    implConnect(remote, attachment, handler);
}

private <A> void implConnect(...) {
    // ① begin()：记录调用状态（供 cancel/中断检查）
    // ② connect0()：发起非阻塞 connect
    // ③ 未完成 → 注册到多路复用器（内部 Selector）等连接事件
    // ④ 就绪 → finishConnect() 确认
    // ⑤ 完成 → handler.completed(null, attachment)
}

AsynchronousSocketChannel 连接流程（Linux）：
  connect → begin() → connect0() 非阻塞发起
  未完成 → 挂到通道内部的多路复用器（用 Selector 监听 OP_CONNECT）
  连接就绪 → finishConnect() 校验 → 回调 completed()
  读/写同样走内部 Selector 就绪 → 提交线程池 → 回调

与 SocketChannel 的区别：AIO 通道把"就绪等待 + 数据读写 + 回调"都封装在内部，用户只写回调，无需自己维护 Selector 循环。

---

五、Future 风格的异步

除回调外，AIO 还提供 Future 风格接口：

public final Future<Integer> read(ByteBuffer dst, long position);   // 返回 Future

// 内部 PendingFuture
// sun.nio.ch.PendingFuture
class PendingFuture<V,A> implements Future<V> {
    private volatile V value;          // 结果
    private volatile Throwable exc;    // 异常
    private volatile boolean hasResult;

    public V get() throws ... {
        // 未完成 → 阻塞等待（LockSupport.park / 信号量）
        return value;
    }
    // 完成时：run() 设置结果 → 唤醒等待线程 → 调用 handler
    void setResult(V res) { value = res; hasResult = true; ... }
}

Future 风格 vs 回调风格：
  read().get()          → 阻塞等待结果（类似同步，但发起时已异步）
  read(handler)         → 完成回调（真正非阻塞）
  二者底层同一套 PendingFuture 完成机制：
    完成 → 设置 value → 唤醒 get() 等待者 + 触发 handler

// 两种风格对比
Future<Integer> f = channel.read(buffer, 0L);   // 立即返回
int n = f.get();                                 // 阻塞等结果

channel.read(buffer, 0L, null, handler);         // 立即返回，回调通知

---

六、实现要点

AIO 核心：
  Linux：UnixAsynchronousFileChannelImpl（线程池 + pread 模拟异步）
  Windows：WindowsAsynchronousFileChannelImpl（IOCP 真异步）
  CompletionHandler：completed/failed 回调，在通道组线程执行
  AsynchronousChannelGroup：固定线程池共享，控制总线程数
  IOCP：CreateIoCompletionPort + ReadFileEx + GetQueuedCompletionStatus
  AsynchronousSocketChannel：内部 Selector 就绪 + 线程池 + 回调
  PendingFuture：Future 与回调共用的完成状态载体

常见陷阱：
  回调线程阻塞 → 完成事件积压（用短回调或再投递）
  AIO 回调与调用线程不同 → 注意可见性与线程安全
  Linux 的"异步"是线程池模拟，不是内核异步
  get() 阻塞会占线程 → 高并发用回调风格
  Netty 时代 AIO 应用较少（自研事件循环已够用）