AIO编程

Posted 2021-02-01 junjiang3

AIO简介

我们知道NIO是同步非阻塞，服务器实现模式为一个请求一个线程，即客户端发送的连接请求都会注册到多路复用器上，多路复用器轮询到连接有I/O请求时才启动一个线程进行处理。
而AIO则是则是异步非阻塞的，并且提供了异步文件通道和异步套接字通道的实现。主要通过两种方式获取操作的结果：

• 通过Future类来表示异步操作的结果
• 在执行异步操作的时候传入一个java.nio.channels
  AIO的异步套接字通道是真正的异步非阻塞I/O，对应于UNIX网络编程模型中的事件驱动I/O，他不需要通过多路复用器对注册的通道进行轮询操作即可实现异步读写，从而简化了NIO编程模型。

AIO创建的TimeServer

public class TimeServer {

    public static void main(String[] args){
        int port = 8080;
        if (args != null && args.length >0){
            try{
                port = Integer.parseInt(args[0]);
            }catch (NumberFormatException e){

            }
        }
        AsyncTimeServerHandler timeServerHandler = new AsyncTimeServerHandler(port);
        new Thread(timeServerHandler,"AIO-AsyncTimeServerHandler-001").start();
    }
}

首先创建异步的时间服务处理器，然后启动线程将异步时间服务Handler拉起

public class AsyncTimeServerHandler implements Runnable{

    private int port;
    CountDownLatch countDownLatch;
    AsynchronousServerSocketChannel channel;

    public AsyncTimeServerHandler(int port) {
        this.port = port;
        try {
            channel = AsynchronousServerSocketChannel.open();
            channel.bind(new InetSocketAddress(port));
        }catch (IOException e){
            e.printStackTrace();
        }

    }

    public void run() {
        countDownLatch = new CountDownLatch(1);
        doAccept();
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void doAccept(){
        channel.accept(this,new AcceptCompletionHandler());
    }
}

在构造方法中，我们创建了一个一步的Channel，然后调用bind方法绑定了监听的端口。
在run方法中我们初始化了一个CountDownLatch对象，是为了在完成一组正在执行的操作之前，线程一直阻塞在那儿
在doAccept方法中接收客户端的连接，我们可以传递一个handler示例接受accept操作成功的通知消息，其代码如下：

public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel,AsyncTimeServerHandler>{

    @Override
    public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {
        attachment.channel.accept(attachment,this);
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        result.read(buffer,buffer,new ReadCompltetionHandler(result));
    }

    @Override
    public void failed(Throwable exc, AsyncTimeServerHandler attachment) {
        exc.printStackTrace();
        attachment.countDownLatch.countDown();
    }
}

其中，我们在complete方法中继续调用了accept方法，是为了有新的客户端接入成功，因为一个AsynchronousServerSocketChannel可以接受成千上万个客户端
而链路建立成功以后，服务端可以接受客户端的请求消息了，通过read方法进行异步读操作，其中传入了一个Handler，接受通知回调业务。其代码如下

public class ReadCompltetionHandler implements CompletionHandler<Integer,ByteBuffer>{

    private AsynchronousSocketChannel channel;

    public ReadCompltetionHandler(AsynchronousSocketChannel channel) {
        if (channel == null)
            this.channel = channel;
    }

    @Override
    public void completed(Integer result, ByteBuffer attachment) {
        attachment.flip();
        byte[] body = new byte[attachment.remaining()];
        attachment.get(body);
        try {
            String req = new String(body,"UTF-8");
            String curentTime = "QUERY TIME ORDER".equalsIgnoreCase(req)?new Date(
                    System.currentTimeMillis()
            ).toString():"BAD ORDER";
            doWrite(curentTime);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
    }

    private void doWrite(String currentTime){
        if (!StringUtil.isNullOrEmpty(currentTime)){
            byte[] bytes = currentTime.getBytes();
            ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
            buffer.put(bytes);
            buffer.flip();
            channel.write(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
                @Override
                public void completed(Integer result, ByteBuffer attachment) {
                    if (buffer.hasRemaining()){
                        channel.write(buffer,buffer,this);
                    }
                }

                @Override
                public void failed(Throwable exc, ByteBuffer attachment) {
                    try {
                        channel.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

    @Override
    public void failed(Throwable exc, ByteBuffer attachment) {
        try {
            this.channel.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

其中具体的处理逻辑和NIO的TimeServer相同，不做详细分析了

AIO的TimeClient

public class TimeClient {

    public static void main(String[] args){
        int port = 8080;
        if (args != null && args.length >0){
            try{
                port = Integer.parseInt(args[0]);
            }catch (NumberFormatException e){

            }
        }

        AsyncTimeClientHandler timeClientHandler = new AsyncTimeClientHandler("127.0.0.1",port);
        new Thread(timeClientHandler,"AIO-AsyncTimeClientHandler-001").start();
    }
}

在其中我们通过一个I/O线程创建一步时间服务器客户端Handler，具体代码如下：

public class AsyncTimeClientHandler implements CompletionHandler<Void,AsyncTimeClientHandler>,Runnable{

    private AsynchronousSocketChannel client;
    private String host;
    private int port;
    private CountDownLatch countDownLatch;

    public AsyncTimeClientHandler(String host, int port) {
        this.host = host;
        this.port = port;
        try {
            client = AsynchronousSocketChannel.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void completed(Void result, AsyncTimeClientHandler attachment) {
        byte[] req = "QUERY TIME ORDER".getBytes();
        ByteBuffer buffer = ByteBuffer.allocate(req.length);
        buffer.put(req);
        buffer.flip();
        client.write(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer result, ByteBuffer attachment) {
                if (attachment.hasRemaining()){
                    client.write(attachment,attachment,this);
                }else{
                    ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                    client.read(readBuffer, readBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                        @Override
                        public void completed(Integer result, ByteBuffer attachment) {
                            byte[] bytes = new byte[attachment.remaining()];
                            attachment.get(bytes);
                            try {
                                String body = new String(bytes,"UTF-8");
                                System.out.print(body);
                            } catch (UnsupportedEncodingException e) {
                                e.printStackTrace();
                            }
                        }

                        @Override
                        public void failed(Throwable exc, ByteBuffer attachment) {
                            try {
                                client.close();
                                countDownLatch.countDown();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    });
                }
            }

            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                try {
                    client.close();
                    countDownLatch.countDown();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    @Override
    public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
        exc.printStackTrace();
        try {
            client.close();
            countDownLatch.countDown();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        countDownLatch = new CountDownLatch(1);
        client.connect(new InetSocketAddress(host,port),this,this);
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            client.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}