目录
1、AIO的核心概念
AIO (Asynchronous I/O) 属于异步I/O，其核心特征是异步与非阻塞，应用程序发起I/O操作后立即返回，无需等待内核完成数据读取；当 I/O 真正完成时，内核主动通知应用程序，再通过回调CompletableHandle或 Future 获取结果,与传统 BIO（Blocking I/O）在调用线程上同步等待、以及 NIO（Non-blocking I/O）通过轮询/多路复用器检测就绪事件的方式完全不同。
AIO 的底层依赖操作系统原生异步 I/O 能力如 Linux 的 AIO、Windows 的 IOCP在 Linux 环境中，可通过 aio_read、aio_write、aio_error、aio_return、lio_listio 等 API 实现真正的异步读写，使进程在等待 I/O 期间可以执行其他计算任务.
2、什么是异步非阻塞？
在聊AIO之前，我们先拆开“异步”和“非阻塞”这两个词。它们经常被连在一起说，但其实是两个独立的概念。
异步（Asynchronous）： 调用线程发起一个I/O操作后，立即返回，不会等待I/O完成。I/O操作由操作系统内核在后台完成，完成后内核主动通知JVM，JVM再调用线程池中的线程去执行回调。也就是说，发起I/O的线程和执行I/O的线程不是同一个，调用线程从头到尾不需要等待。
非阻塞（Non-blocking）： 调用线程发起I/O操作后，不会傻傻地卡在原地。但“不卡住”只是第一步——怎么拿到结果，才是区分同步和异步的关键。NIO的做法是：调用线程返回后，自己不断地轮询（selector.select()），问内核“好了没？好了没？”；而AIO的做法是：调用线程返回后彻底不管了，内核完成后主动把结果推给你。
所以AIO是“异步 + 非阻塞”： 调用线程发起请求后立刻返回（非阻塞），内核完成操作后主动回调（异步）。调用线程全程没有等待，也没有轮询，真正做到了“发起即忘记”（fire and forget）。
3、AIO的api体系
在 Java 中，AIO 位于 java.nio.channels 包下，主要涉及三组异步 Channel：AsynchronousSocketChannel（客户端/服务端）、AsynchronousServerSocketChannel（服务端监听）、AsynchronousFileChannel（文件读写）。其使用方式分为两类：基于 Future 和基于 CompletionHandler 回调;
4、AIO的异步在代码层体现在哪里？
AIO的核心在于异步回调。理解了它的两种异步结果获取方式——Future 和 CompletionHandler——就掌握了AIO的精髓。
不过要注意，严格来说只有 CompletionHandler 才是真正的“回调”。Future 方式虽然发起了异步操作，但获取结果时仍然需要主动调用 get() 去等待，本质上是一种“伪异步”。而 CompletionHandler 才是真正的“发起即忘记”操作完成后，系统主动调用你的回调方法，你全程不需要等待。
java// 方式一：Future 方式，调用后立即返回
Future<Integer> read = fileChannel.read(readBuffer, position);
int bytesRead = read.get();
//方式二：CompletableHandler
AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open();
server.bind(new InetSocketAddress(PORT));
System.out.println("AIO Future 模式服务端启动，端口：" + PORT);
server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
@Override
public void completed(AsynchronousSocketChannel client, Object attachment) {
server.accept(null, this); // 自链：准备接下一个
// ... 注册 read 的回调
}
@Override
public void failed(Throwable exc, Object attachment) { ... }
});
总结：
| 方式 | 核心 | 特点 |
|---|---|---|
| Future | 主动去问结果 | 简单，但会阻塞本质是“假装异步” |
| CompletionHandler | 被动等结果 | 真正的异步，回调驱动 |
5、实战读、写文件
使用Future的方式
java /**
* 使用Future方式写文件
*/
private static void writeByFuture(){
try {
File file = new File(PATH);
if(file.exists()){
boolean delete = file.delete();
if(delete){
file.createNewFile();
}
}else{
file.createNewFile();
}
String content = "这是文件中的测试内容\n";
AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(Paths.get(PATH), StandardOpenOption.WRITE);
long position = 0;
int size = 0;
byte[] bytes = content.getBytes(StandardCharsets.UTF_8);
ByteBuffer byteBuffer = ByteBuffer.allocate(bytes.length);
while (size<FILE_SIZE){
byteBuffer.put(bytes);
byteBuffer.flip();
Future<Integer> write = fileChannel.write(byteBuffer, position);
write.get();
size+=bytes.length;
position+=bytes.length;
byteBuffer.clear();
}
fileChannel.close();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 使用Future读取文件
*/
private static void readFile() {
Path path = Paths.get(WRITE_PATH);
try {
//打开文件通道，只读模式
AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(path, StandardOpenOption.READ);
ByteBuffer readBuffer = ByteBuffer.allocate(1024*12);
long position = 0;
// 方式一：Future 方式，调用后立即返回
Future<Integer> read = fileChannel.read(readBuffer, position);
while (!read.isDone()){
// 此处可执行其他不依赖读取结果的任务
System.out.println("文件读取中，线程可以做其他事情...");
}
int bytesRead = read.get();
readBuffer.flip();
byte[] data = new byte[bytesRead];
readBuffer.get(data);
System.out.println("读取结果: " + new String(data));
fileChannel.close();
} catch (IOException e) {
System.out.println("读取文件异常");
throw new RuntimeException(e);
} catch (ExecutionException e) {
throw new RuntimeException(e);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
使用Completable读写文件
java /**
* 使用Completable方式写文件
*/
private static void writeByCompleHandle() {
try {
File file = new File(WRITE_PATH);
if (file.exists()) {
boolean delete = file.delete();
if (delete) {
file.createNewFile();
}
}else{
file.createNewFile();
}
AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(Paths.get(WRITE_PATH),StandardOpenOption.WRITE);
String content = "使用completable写入文件\n";
byte[] bytes = content.getBytes(StandardCharsets.UTF_8);
int size = 0;
int position = 0;
while (size<FILE_SIZE){
ByteBuffer byteBuffer = ByteBuffer.allocate(bytes.length);
byteBuffer.put(bytes);
byteBuffer.flip();
fileChannel.write(byteBuffer, position, byteBuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
System.out.println(Thread.currentThread().getName()+"写入完成");
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
System.out.println(Thread.currentThread().getName()+"写入失败");
}
});
byteBuffer.clear();
size+=bytes.length;
position+=bytes.length;
}
fileChannel.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
/**
* attachment 里要带「读到哪了」—— 回调跑在别的线程上，看不见方法里的局部变量 position。
*/
private record Chunk(ByteBuffer buf, long position) {}
private static void readByCompletionHandler(){
Path path = Paths.get(WRITE_PATH);
try {
AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(path,StandardOpenOption.READ);
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
CountDownLatch latch = new CountDownLatch(1); // 只是别让 main 在回调之前退出
Chunk start = new Chunk(byteBuffer, 0);
fileChannel.read(start.buf(), start.position(), start, new CompletionHandler<Integer, Chunk>() {
@Override
public void completed(Integer result, Chunk attachment) {
if (result == -1) { // ← -1 就是 EOF：读完了
System.out.printf("[%s] 读完了，共 %d 字节%n",
Thread.currentThread().getName(), attachment.position());
latch.countDown();
return;
}
System.out.printf("[%s] position=%-6d 读到 %d 字节%n",
Thread.currentThread().getName(), attachment.position(), result);
attachment.buf().clear(); // 复用同一块 buffer，下一发继续往里写
long next = attachment.position() + result;
try {
// 自链：把 this（本 handler 实例）传回去，这一发读完接下一发
fileChannel.read(attachment.buf(), next, new Chunk(attachment.buf(), next), this);
} catch (Exception e) {
failed(e, attachment); // completed 里不能往外抛异常，必须自己兜
}
}
@Override
public void failed(Throwable exc, Chunk attachment) {
System.out.println("读取失败: " + exc); // 出错必须打出来，否则静默失败
latch.countDown();
}
});
latch.await(); // 等整条链跑完
fileChannel.close(); // 回调结束了才关通道
} catch (Exception e) {
throw new RuntimeException(e);
}
}
6、实战完成服务端和客户端的通信
服务端代码
javapublic class AsynchronousServiceDemo {
private static final int port = 8888;
private static Map<String,AsynchronousSocketChannel> clients = new ConcurrentHashMap<>();
public static void main(String[] args) {
try {
AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open();
serverSocketChannel.bind(new InetSocketAddress(port));
System.out.println("服务端开启,运行在"+port+"上");
//连接处理
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
serverSocketChannel.accept(byteBuffer, new CompletionHandler<AsynchronousSocketChannel, Object>() {
@Override
public void completed(AsynchronousSocketChannel client, Object attachment) {
serverSocketChannel.accept(attachment,this); //继续注册
//连接成功
try {
String key = client.getRemoteAddress().toString();
clients.put(key,client);
String message = key+"加入了群聊";
groupNotice(message,client);
} catch (IOException e) {
System.out.println("加入群聊异常");
throw new RuntimeException(e);
}
ByteBuffer readData = ByteBuffer.allocate(1024);
client.read(readData, readData, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
if(result==-1){
//处理断开事件
cleanUp(client);
return;
}
attachment.flip();
byte[] data = new byte[result];
attachment.get(data);
String info = new String(data, StandardCharsets.UTF_8);
System.out.println("客户端发送消息:"+info);
attachment.clear();
//广播消息
groupNotice(info,client);
//继续注册读取事件
client.read(attachment,attachment,this);
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
System.err.println("读取消息异常: " + exc.getMessage());
cleanUp(client);
}
});
}
@Override
public void failed(Throwable exc, Object attachment) {
//连接失败
System.err.println("接受连接失败: " + exc.getMessage());
}
});
//阻塞主线程不让其结束
Thread.currentThread().join();
} catch (IOException e) {
System.out.println("服务端异常:"+e.getMessage());
throw new RuntimeException(e);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
/**
* 群聊通知
*/
private static void groupNotice(String message,AsynchronousSocketChannel client){
if(!clients.isEmpty()){
byte[] data = message.getBytes(StandardCharsets.UTF_8);
ByteBuffer byteBuffer = ByteBuffer.allocate(data.length);
byteBuffer.put(data);
byteBuffer.flip();
for (AsynchronousSocketChannel asc : clients.values()) {
if(asc==client){
continue;
}
byteBuffer.rewind();
try {
asc.write(byteBuffer); //这里调用write或者read其实是吧ByteBuffer交给内核了 你就不能再操作了,要操作就去接收异步通知等内核操作完会通知你,和NIO不一样
/*while (byteBuffer.hasRemaining()) {
asc.write(byteBuffer);
}*/
} catch (Exception e) {
try {
System.out.println("广播失败，移除死连接:" + asc.getRemoteAddress());
clients.remove(String.valueOf(asc.getRemoteAddress()));
} catch (IOException ignore) {
}
try {
asc.close();
} catch (IOException ignore) {
}
}
}
}
}
/**
* 清除连接
*/
private static void cleanUp(AsynchronousSocketChannel channel){
String message = null;
try {
message = channel.getRemoteAddress().toString()+"断开连接";
clients.remove(channel.getRemoteAddress().toString());
} catch (IOException e) {
System.out.println("断开连接");
}
if(message!=null){
groupNotice(message,channel);
}
try {
channel.close();
} catch (IOException e) {
}
}
}
客户端代码
javapublic class AsynchronousClientDemo {
public static void main(String[] args) {
//开启一个线程去接收服务端的消息
ExecutorService executorService = Executors.newSingleThreadExecutor();
try {
AsynchronousSocketChannel channel = AsynchronousSocketChannel.open();
channel.connect(new InetSocketAddress("127.0.0.1",8888));
//接收服务器消息
executorService.submit(()->{
while (true){
ByteBuffer reader = ByteBuffer.allocate(20);
channel.read(reader, reader, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer byteBuffer) {
channel.read(byteBuffer,byteBuffer,this);
if(result==-1){
System.out.println("断开");
}
if(result>0){
byteBuffer.flip();
byte[] data = new byte[result];
byteBuffer.get(data);
System.out.println("接收到消息："+new String(data,StandardCharsets.UTF_8));
byteBuffer.clear();
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
System.out.println("接收消息异常");
}
});
}
});
//向服务器发送消息
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in));
String message = "";
System.out.print("请输入消息:");
while ((message= bufferedReader.readLine())!=null){
if ("exit".equalsIgnoreCase(message.trim())) {
break;
}
byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
ByteBuffer byteBuffer = ByteBuffer.allocate(bytes.length);
byteBuffer.put(bytes);
byteBuffer.flip();
channel.write(byteBuffer);
}
channel.close();
} catch (IOException e) {
throw new RuntimeException(e);
}finally {
executorService.shutdown();
}
}
}
7、为什么AIO要用自链的方式？
我觉得这是 AIO 最反直觉的地方：平时写循环，开始和结束都由我们的代码说了算；到了 AIO，这个权力被反转了。 拿 NIO 对照最清楚。NIO 里我手上有一个总控循环：
javawhile (true) {
selector.select(); // 等事件
if (key.isReadable()) { ... } // 由我决定这一轮干什么
}
selector 给我的是**「就绪」——"现在可以读了"。但读不读、读多少、还是先去做别的，全是我的决定**。这次不理它也行，下次 select() 它还会告诉你。NIO 的事件是提议，控制权始终在我手上。
AIO 不一样。它的回调给我的不是许可，是回执：completed() 一到，这件事已经结束了，这是结果。这个事件你不理，结果就丢了 —— 没有"下次再说"。而且它什么时候来完全由内核说了算，我没有任何办法去问一句"好了没"，只能等它叫我。
所以下一轮只能在回调里发起。原因： 发起下一轮的前提是"上一轮完成了"，而"完成"这个消息只送到回调那里。信息在哪，控制权就在哪。
一句话总结就是不是我在驱动流程，而是每一段流程结束的时候，由它来叫我接上下一段。
如果对你有用的话，可以打赏哦
打赏
本文作者:章鱼哥
本文链接:
版权声明:本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！
目录