AIO编程-摩杜云开发者社区

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格式等等，所以需要我们做的事情还有很多