AIO编程

布满荆棘的人生 2022-05-14 01:29 183阅读 0赞

NIO2.0引入了新的异步通道的概念，并提供了异步文件通道和异步套接字通道的实现。异步通道提供2种方式获取操作结果。

1.通过java.util.concurrent.Future来获取异步操作的结果；  
2.在异步操作的时候传入一个java.nio.channels。  
CompletionHandler作为异步完成的回调。

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

AIO版本的时间服务器代码  
服务端：

public class TimeServer { 
        public static void main(String[] args) {
            TimeServer ts = new TimeServer();
            AsyncTimeServerHandler atsh = ts.new AsyncTimeServerHandler(9999);
            // 可以不使用线程
            new Thread(atsh,"AsyncTimeServerHandler").start();
        }
        class AsyncTimeServerHandler implements Runnable {
            private AsynchronousServerSocketChannel asynchronousServerSocketChannel;
            private CountDownLatch latch = null;
            public AsyncTimeServerHandler(int port) {
                try {
                    // 创建一个异步的服务端通道
                    asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open();
                    // 绑定监听端口
                    asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
                    System.out.println("TimeServer is start at:" + port);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            @Override
            public void run() {
                // 使用CountDownLatch来让服务端在操作完成才退出。
                latch = new CountDownLatch(1);
                doAccept();
                try {
                    latch.await(); // 阻塞
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            private void doAccept() {
                // 接收客户端的连接，使用CompletionHandler来接收accept操作成功的通知消息
                asynchronousServerSocketChannel.accept(this,new AccpetCompleteHandler());
            }
        }
        class AccpetCompleteHandler implements CompletionHandler<AsynchronousSocketChannel,AsyncTimeServerHandler> {
            @Override
            public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {
                // 继续接收其他的客户端连接
                attachment.asynchronousServerSocketChannel.accept(attachment,this);
                // 客户端连接成功后，使用ReadCompletionHandler读取客户端发送的信息，将信息读取到byteBuffer。
                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                result.read(byteBuffer,byteBuffer,new ReadCompleteHandler(result));
            }
            @Override
            public void failed(Throwable exc, AsyncTimeServerHandler attachment) {
                exc.printStackTrace();
                attachment.latch.countDown();
            }
        }
        class ReadCompleteHandler implements CompletionHandler<Integer,ByteBuffer> {
            private AsynchronousSocketChannel channel;
            public ReadCompleteHandler(AsynchronousSocketChannel channel) {
                if (this.channel == null) {
                    this.channel = channel;
                }
            }
            @Override
            public void completed(Integer result, ByteBuffer attachment) {
                // 为后续的从缓冲区读取数据做准备
                attachment.flip();
                // 根据缓冲区的可读字节数创建数组
                byte[] bytes = new byte[attachment.remaining()];
                // 将缓冲区的可读数据读取到bytes数组
                attachment.get(bytes);
                try {
                    String body = new String(bytes,"UTF-8");
                    System.out.println("接收到客户端请求：" + body);
                    // 如果客户端发送的是Server Time，则将服务器当前时间发送给客户端。
                    if ("Server Time".equalsIgnoreCase(body)) {
                        doWrite((new Date()).toString());
                    }
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
            }
            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                exc.printStackTrace();
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            private void doWrite(String response) {
                byte[] bytes = response.getBytes();
                // 根据响应内容的大小构建缓冲区
                ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
                // 将相应内容复制到缓冲区
                buffer.put(bytes);
                // 为后续从缓冲区读取数据准备
                buffer.flip();
                // 异步write方法将缓冲区数据写出
                channel.write(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
                    @Override
                    public void completed(Integer result, ByteBuffer attachment) {
                        // 如果缓冲区还有数据，继续发送
                        if (attachment.hasRemaining()) {
                            channel.write(attachment,attachment,this);
                        }
                    }
                    @Override
                    public void failed(Throwable exc, ByteBuffer attachment) {
                        try {
                            channel.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        }
    }

客户端：

public class TimeClient { 
        public static void main(String[] args) {
            TimeClient tc = new TimeClient();
            AsyncTimeClientHandler tch = tc.new AsyncTimeClientHandler("127.0.0.1",9999);
            new Thread(tch,"TimeClientHandler").start();
        }
        class AsyncTimeClientHandler implements CompletionHandler<Void,AsyncTimeClientHandler>,Runnable {
            private String host;
            private int port;
            private CountDownLatch latch ;
            private AsynchronousSocketChannel client;
            public AsyncTimeClientHandler(String host,int port) {
                this.host = host;
                this.port = port;
                try {
                    client = AsynchronousSocketChannel.open();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            @Override
            public void run() {
                // 使用CountDownLatch，防止异步操作还未完成就退出了。
                latch = new CountDownLatch(1);
                // 异步连接服务端
                client.connect(new InetSocketAddress(host,port),this,this);
                try {
                    latch.await(); // 阻塞
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                try {
                    client.close(); // 操作完成后，关闭通道
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            @Override
            public void completed(Void result, AsyncTimeClientHandler attachment) {
                // 要向服务器发送的数据
                byte[] data = "Server Time".getBytes();
                final ByteBuffer buffer = ByteBuffer.allocate(data.length);
                buffer.put(data);
                buffer.flip();
                // 异步写消息给服务端
                client.write(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
                    @Override
                    public void completed(Integer result, ByteBuffer attachment) {
                        // 如果有消息还没有发送完毕，则继续发送
                        if (buffer.hasRemaining()) {
                            client.write(attachment,attachment,this);
                        }
                        else {
                            // 消息发送完成后，异步读取服务器响应
                            // 预分配空间为1K
                            ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                            client.read(readBuffer, readBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                                @Override
                                public void completed(Integer result, ByteBuffer attachment) {
                                    // 为后续从缓冲区读取数据做准备
                                    attachment.flip();
                                    // 根据缓冲区可读字节的长度构建字节数组
                                    byte[] data = new byte[attachment.remaining()];
                                    // 将缓冲区数据复制到data数组
                                    attachment.get(data);
                                    try {
                                        String body = new String(data,"UTF-8");
                                        System.out.println("接收到服务端消息：" + body);
                                        // 这里模拟只请求服务器1次，完成后退出。
                                        latch.countDown();
                                    } catch (UnsupportedEncodingException e) {
                                        e.printStackTrace();
                                    }
                                }
                                @Override
                                public void failed(Throwable exc, ByteBuffer attachment) {
                                    try {
                                        client.close();
                                    } catch (IOException e) {
                                        e.printStackTrace();
                                    }
                                    latch.countDown();
                                }
                            });
                        }
                    }
                    @Override
                    public void failed(Throwable exc, ByteBuffer attachment) {
                        try {
                            client.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        latch.countDown();
                    }
                });
            }
            @Override
            public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
                try {
                    client.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                latch.countDown();
            }
        }
    }

JDK底层通过线程池ThreadPoolExecutor来执行回调通知。AsynchronousServerSocketChannel和AsynchronousSocketChannel，他们都由JDK底层的线程池回调并驱动读写操作。

参考《Netty权威指南》