AIO编程

Dear 丶 2021-09-20 17:54 346阅读 0赞

AIO是异步IO的缩写，虽然NIO在网络操作中，提供了非阻塞的方法，但是NIO的IO行为还是同步的。对于NIO来说，我们的业务线程是在IO操作准备好时得到通知，接着就由这个线程自行进行IO操作，IO操作本身还是同步的。NIO 2.0引入了新的异步通道的概念, 并提供了异步文件通道和异步套接字通道的实现。异步通道提供以下两种方式获取获取操作结果：

*  通过 java.util.concurrent.Future类来表示异步操作的结果
 *  在执行异步操作的时候传入一个java.nio.channels.CompletionHandler接口的实现类作为操作完成的回调 。

NIO2.0的异步套接字通道是真正的异步非阻塞I/O, 对应于UNIX网络编程中的事件驱动 I/O ( AIO)。它不需要通过多路复用器 ( Selector) 对注册的通道进行轮询操作即可实现异步读写 , 从而简化NIO的编程模型 。

在AIO socket编程中，服务端通道是AsynchronousServerSocketChannel，这个类提供了一个open()静态工厂，一个bind()方法用于绑定服务端IP地址（还有端口号），另外还提供了accept()用于接收用户连接请求。在客户端使用的通道是AsynchronousSocketChannel,这个通道处理提供open静态工厂方法外，还提供了read和write方法。

在AIO编程中，发出一个事件（accept read write等）之后要指定事件处理类（回调函数），AIO中的事件处理类是CompletionHandler<V,A>，这个接口定义了如下两个方法，分别在异步操作成功和失败时被回调。

*  void completed(V result, A attachment);
 *  void failed(Throwable exc, A attachment);

说了这么多，感觉很像Ajax中的回调方法，只不过换成了类。直接看网友提供的代码吧：

服务端：

public class AIOEchoServer implements Runnable{
        public final static int PORT = 8001;
        public final static String IP = "127.0.0.1";
    
    
        private AsynchronousServerSocketChannel server = null;
    
        public AIOEchoServer() {
            try {
                ExecutorService executor = Executors.newFixedThreadPool(20);
                AsynchronousChannelGroup asyncChannelGroup = AsynchronousChannelGroup.withThreadPool(executor);
                server = AsynchronousServerSocketChannel.open(asyncChannelGroup).bind(new InetSocketAddress(IP, PORT));
                //同样是利用工厂方法产生一个通道，异步通道 AsynchronousServerSocketChannel
    //            server = AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(IP,PORT));
                //通过setOption配置Socket
                server.setOption(StandardSocketOptions.SO_REUSEADDR, true);
                server.setOption(StandardSocketOptions.SO_RCVBUF, 16 * 1024);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        //使用这个通道(server)来进行客户端的接收和处理
        @Override
        public void run() {
            System.out.println("Server listen on " + PORT);
    
            //注册事件和事件完成后的处理器，这个CompletionHandler就是事件完成后的处理器
            server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
    
                final ByteBuffer buffer = ByteBuffer.allocate(1024);
    
                @Override
                public void completed(AsynchronousSocketChannel result, Object attachment) {
    
                    System.out.println(Thread.currentThread().getName());
                    Future<Integer> writeResult = null;
    
                    try {
                        buffer.clear();
                        //把socket中的数据读取到buffer中
                        result.read(buffer).get(100, TimeUnit.SECONDS);
                        System.out.println("In server: " + new String(buffer.array()));
                        buffer.flip();
                        //将数据写回客户端
                        writeResult = result.write(buffer);
                        buffer.flip();
                    } catch (InterruptedException | ExecutionException | TimeoutException e) {
                        e.printStackTrace();
                    } finally {
                        try {
                            System.out.println(writeResult.get());
                            //关闭处理完的socket，并重新调用accept等待新的连接
                            result.close();
                            server.accept(null, this);
                        } catch (InterruptedException | ExecutionException e) {
                            e.printStackTrace();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
    
                }
    
                @Override
                public void failed(Throwable exc, Object attachment) {
                    System.out.println("failed:" + exc);
                }
    
            });
        }
      public static void main(String[] args) {
            //因为AIO不会阻塞调用进程，因此必须在主进程阻塞，才能保持进程存活。
            new Thread(new AIOEchoServer()).start();
            while (true) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
    }

客户端：

public class AIOClient implements Runnable {
        private AsynchronousSocketChannel channel;
    
        public AIOClient() {
            try {
                channel = AsynchronousSocketChannel.open();
            } catch (IOException e) {
                e.printStackTrace();
            }
    
        }
    
        @Override
        public void run() {
            InetSocketAddress serverAddress = new InetSocketAddress("127.0.0.1", 8001);
            channel.connect(serverAddress, null, new CompletionHandler<Void, Object>() {
    
                @Override
                public void completed(Void result, Object attachment) {
                    channel.write(ByteBuffer.wrap("Hello server".getBytes()));
                    final ByteBuffer buffer = ByteBuffer.allocate(1024);
                    channel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
    
                        @Override
                        public void completed(Integer result, ByteBuffer attachment) {
                            buffer.flip();
                            System.out.println(new String(buffer.array()));
                            try {
                                channel.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
    
                        @Override
                        public void failed(Throwable exc, ByteBuffer attachment) {
                        }
    
                    });
                }
    
                @Override
                public void failed(Throwable exc, Object attachment) {
                }
    
            });
        }
    
        public static void main(String[] args) throws IOException {
    
            new Thread(new AIOClient()).start();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
    
    }

当然，以上代码是示例代码，目标是为了让您了解JAVA AIO框架的基本使用。所以它还有很多改造的空间，例如，在生产环境下，我们需要记录这个通道上“用户的登录信息”。那么这个需求可以使用JAVA AIO中的“附件”功能进行实现。我们在示例中使用字符串传输内容，但是在正式生产环境下，您会这样用吗？显然是不会的，我们可能会使用json格式等等，所以需要我们做的事情还有很多

参考：

[https://blog.csdn.net/x\_i\_y\_u\_e/article/details/52223406][https_blog.csdn.net_x_i_y_u_e_article_details_52223406]

[https://blog.csdn.net/caomiao2006/article/details/47819255][https_blog.csdn.net_caomiao2006_article_details_47819255]

[https://blog.csdn.net/u010963948/article/details/78507615][https_blog.csdn.net_u010963948_article_details_78507615]

[https://blog.csdn.net/z69183787/article/details/52953974][https_blog.csdn.net_z69183787_article_details_52953974]

[https://blog.csdn.net/chenyi8888/article/details/8086614][https_blog.csdn.net_chenyi8888_article_details_8086614]

[https_blog.csdn.net_x_i_y_u_e_article_details_52223406]: https://blog.csdn.net/x_i_y_u_e/article/details/52223406
[https_blog.csdn.net_caomiao2006_article_details_47819255]: https://blog.csdn.net/caomiao2006/article/details/47819255
[https_blog.csdn.net_u010963948_article_details_78507615]: https://blog.csdn.net/u010963948/article/details/78507615
[https_blog.csdn.net_z69183787_article_details_52953974]: https://blog.csdn.net/z69183787/article/details/52953974
[https_blog.csdn.net_chenyi8888_article_details_8086614]: https://blog.csdn.net/chenyi8888/article/details/8086614