Java中的异步非阻塞AIO模型详解
更新时间：2023年09月19日 10:20:34 作者：超大充电宝
这篇文章主要介绍了Java中的异步非阻塞AIO模型详解,AIO需要操作系统的支持,在linux内核2.6版本中加入了对真正异步IO的支持,java从jdk1.7开始支持AIO,本文提供了部分实现代码,需要的朋友可以参考下
1、AIO：异步非阻塞简介
AIO需要操作系统的支持，在linux内核2.6版本中加入了对真正异步IO的支持，java从jdk1.7开始支持AIO
核心类有AsynchronousSocketChannel 、AsynchronousServerSocketChannel、AsynchronousChannelGroup
AsynchronousChannelGroup是异步Channel的分组管理器，它可以实现资源共享。
创建AsynchronousChannelGroup时，需要传入一个ExecutorService，也就是绑定一个线程池，该线程池负责两个任务：处理IO事件和触发CompletionHandler回调接口。
2、AsynchronousServerSocketChannel:AIO中网络通信服务端Socket
accept() 方法: AsynchronousServerSocketChannel创建成功后，类似于ServerSocket，也是调用 accept() 方法来接受来自客户端的连接， 由于异步IO实际的IO操作是交给操作系统来做的，用户进程只负责通知操作系统进行IO和接受操作系统IO完成的通知。 所以异步的ServerChannel调用 accept() 方法后，当前线程不会阻塞， 程序也不知道accept()方法什么时候能够接收到客户端请求并且操作系统完成网络IO， 为解决这个问题，AIO为accept()方法提供两个版本：
Future<AsynchronousSocketChannel> accept() ：
开始接收客户端请求，如果当前线程需要进行网络IO（即获得AsynchronousSocketChannel），则应该调用该方法返回的Future对象的get()方法，但是get()方法会阻塞该线程，所以这种方式是阻塞式的异步IO。
< A > void accept (Aattachment,CompletionHandler<AsynchronousSocketChannel,? super A> handler)：
开始接受来自客户端请求，连接成功或失败都会触发CompletionHandler对象的相应方法。
其中AsynchronousSocketChannel就代表该CompletionHandler处理器在处理连接成功时的result是AsynchronousSocketChannel的实例。
而CompletionHandler接口中定义了两个方法，
completed(V result , A attachment)：当IO完成时触发该方法，该方法的第一个参数代表IO操作返回的对象，
第二个参数代表发起IO操作时传入的附加参数。
faild(Throwable exc, A attachment)：当IO失败时触发该方法，第一个参数代表IO操作失败引发的异常或错误。
3、AIO编程
服务端
public class AioServer {
private static int DEFAULT_PORT = 12345;
private static ServerHandler serverHandle;
public volatile static long clientCount = 0;
public static void start(){
start(DEFAULT_PORT);
}
public static synchronized void start(int port){
if(serverHandle!=null)
return;
serverHandle = new ServerHandler(port);
new Thread(serverHandle,"Server").start();
}
public static void main(String[] args) {
AioServer.start();
}
}public class ServerHandler implements Runnable{
private AsynchronousServerSocketChannel channel;
public ServerHandler(int port) {
try {
//创建服务端通道
channel = AsynchronousServerSocketChannel.open();
//绑定端口
channel.bind(new InetSocketAddress(port));
System.out.println("服务端已启动，端口号:"+port);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
channel.accept(this, new AcceptHandler());
// Future <AsynchronousSocketChannel> accept = channel.accept();
//该步操作是异步操作 防止当前线程直接执行结束
//方案1： while(true)+sleep
while (true) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// //方案2 CountDownLatch 作用：在完成一组正在执行的操作之前，允许当前的现场一直阻塞 此处，让现场在此阻塞，防止服务端执行完成后退出
//
// CountDownLatch count = new CountDownLatch(1);
// channel.accept(this, new AcceptHandler());
// try {
// count.await();
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
}
// CompletionHandler<V,A>
// V-IO操作的结果，这里是成功建立的连接,AsynchronousSocketChannel
// A-IO操作附件，这里传入AsynchronousServerSocketChannel便于继续接收请求建立新连接
class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, ServerHandler> {
@Override
public void completed(AsynchronousSocketChannel channel, ServerHandler serverHandler) {
//创建新的Buffer
ByteBuffer buffer = ByteBuffer.allocate(1024);
//异步读 第三个参数为接收消息回调的业务Handler
// channel.read(buffer, buffer, new ReadHandler(channel));
//继续接受其他客户端请求
serverHandler.channel.accept(null, this);
}
@Override
public void failed(Throwable exc, ServerHandler serverHandler) {
exc.printStackTrace();
}
}
class ReadHandler implements CompletionHandler<ByteBuffer, ByteBuffer> {
//用户读取或者发送消息的channel
private AsynchronousSocketChannel channel;
public ReadHandler(AsynchronousSocketChannel channel) {
this.channel = channel;
}
@Override
public void completed(ByteBuffer result, ByteBuffer attachment) {
result.flip();
byte[] msg = new byte[result.remaining()];
result.get(msg);
try {
String expression = new String(msg, "UTF-8");
System.out.println("服务器收到消息: " + expression);
// String result1 = "服务端收到消息\n";
result.clear();
//向客户端发送消息
doWrite(expression);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
//发送消息
private void doWrite(String msg) {
byte[] bytes = msg.getBytes();
ByteBuffer buffer = ByteBuffer.allocate(1024);
buffer.put(bytes);
buffer.flip();
//异步写数据
channel.write(buffer, buffer, new CompletionHandler <Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
//如果没有发送完，继续发送
if (attachment.hasRemaining()) {
channel.write(attachment, attachment, this);
} else {
//创建新的Buffer
ByteBuffer allocate = ByteBuffer.allocate(1024);
//异步读 第三个参数为接收消息回调的业务Handler
// channel.read(allocate, attachment, new ReadHandler(channel));
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
exc.printStackTrace();
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
exc.printStackTrace();
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}客户端
public class AioClient {
private static String DEFAULT_HOST = "127.0.0.1";
private static int DEFAULT_PORT = 12345;
private static ClientHandler clientHandle;
public static void start(){
start(DEFAULT_HOST,DEFAULT_PORT);
}
public static synchronized void start(String ip,int port){
if(clientHandle!=null)
return;
clientHandle = new ClientHandler(ip,port);
new Thread(clientHandle,"Client").start();
}
//向服务器发送消息
public static boolean sendMsg(String msg) throws Exception{
if(msg.equals("exit")) return false;
clientHandle.sendMsg(msg);
return true;
}
public static void main(String[] args) throws Exception{
AioClient.start();
System.out.println("请输入请求消息：");
Scanner scanner = new Scanner(System.in);
while(AioClient.sendMsg(scanner.nextLine()));
}
}public class ClientHandler implements Runnable{
private AsynchronousSocketChannel clientChannel;
private String host;
private int port;
private CountDownLatch latch;
public ClientHandler(String host, int port) {
this.host = host;
this.port = port;
try {
//创建异步的客户端通道
clientChannel = AsynchronousSocketChannel.open();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
//创建CountDownLatch等待
// latch = new CountDownLatch(1);
//发起异步连接操作，回调参数就是这个类本身，如果连接成功会回调completed方法
clientChannel.connect(new InetSocketAddress(host, port), this, new AcceptHandler());
while (true) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// try {
// latch.await();
// } catch (InterruptedException e1) {
// e1.printStackTrace();
// }
// try {
// clientChannel.close();
// } catch (IOException e) {
// e.printStackTrace();
// }
}
//向服务器发送消息
public void sendMsg(String msg){
byte[] req = msg.getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
System.out.println(">>>>>>msg:"+msg);
writeBuffer.put(req);
writeBuffer.flip();
//异步写
clientChannel.write(writeBuffer, writeBuffer,new WriteHandler(clientChannel));
}
/**
* 接收类
*/
class AcceptHandler implements CompletionHandler<Void, ClientHandler> {
public AcceptHandler() {}
@Override
public void completed(Void result, ClientHandler attachment) {
System.out.println("连接服务器成功");
}
@Override
public void failed(Throwable exc, ClientHandler attachment) {
exc.printStackTrace();
try {
attachment.clientChannel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
class WriteHandler implements CompletionHandler<Integer, ByteBuffer> {
private AsynchronousSocketChannel channel;
public WriteHandler(AsynchronousSocketChannel channel) {
this.channel = channel;
}
@Override
public void completed(Integer result, ByteBuffer attachment) {
//完成全部数据的写入
if (attachment.hasRemaining()) {
//数据没有写完，继续写
System.out.println("WriteHandler.hasRemaining>>>>>");
clientChannel.write(attachment, attachment, this);
} else {
//读取数据
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
clientChannel.read(readBuffer, readBuffer, new ReadHandler(clientChannel));
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
exc.printStackTrace();
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {
private AsynchronousSocketChannel clientChannel;
public ReadHandler(AsynchronousSocketChannel clientChannel) {
this.clientChannel = clientChannel;
}
@Override
public void completed(Integer result,ByteBuffer buffer) {
buffer.flip();
byte[] bytes = new byte[buffer.remaining()];
buffer.get(bytes);
String body;
try {
body = new String(bytes,"UTF-8");
System.out.println("客户端收到结果:"+ body);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
@Override
public void failed(Throwable exc,ByteBuffer attachment) {
System.err.println("数据读取失败...");
try {
clientChannel.close();
} catch (IOException e) {
}
}
}
}到此这篇关于Java中的异步非阻塞AIO模型详解的文章就介绍到这了,更多相关Java的AIO模型内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
Spring Boot是一个开源Java-based框架,用于创建独立的、生产级别的Spring应用,它旨在简化Spring应用的初始搭建及开发过程,通过提供各种自动配置的starter包,Spring Boot使得项目配置变得简单快速,感兴趣的朋友一起看看吧2024-02-02
解决Springboot整合shiro时静态资源被拦截的问题
这篇文章主要介绍了解决Springboot整合shiro时静态资源被拦截的问题，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2021-01-01
今天小编就为大家分享一篇关于Java序列化与反序列化的实例分析讲解，小编觉得内容挺不错的，现在分享给大家，具有很好的参考价值，需要的朋友一起跟随小编来看看吧2018-12-12
这篇文章主要介绍了springboot集成redis启动报错问题的解决方式,从错误信息上看缺少pool2相关包,查询资料发现当redis客户端选择Lettuce时候需要增加：commons-pool22023-11-11
添加引用,重启服务,需要的朋友可以参考下
为什么jdk中把 String 类设计成final,主要是为了“ 效率 ”和“安全性”的缘故,若 String 允许被继承, 由于它的高度被使用率, 可能会降低程序的性能，所以String被定义成final,需要了解的朋友可以参考下2013-01-01
这篇文章主要介绍了在Java8与Java7中HashMap源码实现的对比,内容包括HashMap 的原理简单介绍、结合源码在Java7中是如何解决hash冲突的以及优缺点，结合源码以及在Java8中如何解决hash冲突，balance tree相关源码介绍，需要的朋友可以参考借鉴。2017-01-01
springmvc开启异步请求报错Java code using the Ser
这篇文章主要为大家介绍了springmvc开启异步请求报错Java code using the Servlet API or解决分析,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪2024-02-02
在 Spring 容器中，两个 Bean 之间除了通过 <ref> 建立依赖关系外，还存在着一些特殊关系。这篇文章给大家想想介绍了Spring Bean 之间的特殊关系，非常不错，具有参考借鉴价值，需要的朋友参考下2018-05-05
Springmvc RequestMapping请求实现方法解析
这篇文章主要介绍了Springmvc RequestMapping请求实现方法解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2020-09-09
这篇文章主要介绍了Java 设计模式中的策略模式详情，文章围绕主题展开详细的内容介绍，具有一定的参考价值，需要的小伙伴可以参考一下2022-09-09
最新评论