【Java】Java AIO使用

时间 2019-11-16

标签 Java java aio 使用栏目 Java 繁體版

原文原文链接

欢迎关注公众号: nullobject 。
文章首发在我的博客 https://www.nullobject.cn，公众号 nullobject同步更新。

这篇文章主要介绍Java AIO网络编程。

1. AIO是什么

本文所说的AIO特指Java环境下的AIO。AIO是java中IO模型的一种，做为NIO的改进和加强随JDK1.7版本更新被集成在JDK的nio包中，所以AIO也被称做是NIO2.0。区别于传统的BIO(Blocking IO,同步阻塞式模型,JDK1.4以前就存在于JDK中，NIO于JDK1.4版本发布更新)的阻塞式读写，AIO提供了从创建链接到读、写的全异步操做。AIO可用于异步的文件读写和网络通讯。本文将介绍如何使用AIO实现一个简单的网络通讯以及AIO的一些比较关键的API。java

2. 简单的使用

首先以Server端为例，须要建立一个AsynchronousServerSocketChannel示例并绑定监听端口，接着开始监听客户端链接：编程

public class SimpleAIOServer {

    public static void main(String[] args) {
        try {
            final int port = 5555;
            //首先打开一个ServerSocket通道并获取AsynchronousServerSocketChannel实例：
            AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open();
            //绑定须要监听的端口到serverSocketChannel:  
            serverSocketChannel.bind(new InetSocketAddress(port));
            //实现一个CompletionHandler回调接口handler，
            //以后须要在handler的实现中处理链接请求和监听下一个链接、数据收发，以及通讯异常。
            CompletionHandler<AsynchronousSocketChannel, Object> handler = new CompletionHandler<AsynchronousSocketChannel,
                    Object>() {
                @Override
                public void completed(final AsynchronousSocketChannel result, final Object attachment) {
                    // 继续监听下一个链接请求  
                    serverSocketChannel.accept(attachment, this);
                    try {
                        System.out.println("接受了一个链接：" + result.getRemoteAddress()
                                                              .toString());
                        // 给客户端发送数据并等待发送完成
                        result.write(ByteBuffer.wrap("From Server:Hello i am server".getBytes()))
                              .get();
                        ByteBuffer readBuffer = ByteBuffer.allocate(128);
                        // 阻塞等待客户端接收数据
                        result.read(readBuffer)
                              .get();
                        System.out.println(new String(readBuffer.array()));

                    } catch (IOException | InterruptedException | ExecutionException e) {
                        e.printStackTrace();
                    }
                }

                @Override
                public void failed(final Throwable exc, final Object attachment) {
                    System.out.println("出错了：" + exc.getMessage());
                }
            };
            serverSocketChannel.accept(null, handler);
            // 因为serverSocketChannel.accept(null, handler);是一个异步方法，调用会直接返回，
            // 为了让子线程可以有时间处理监听客户端的链接会话，
            // 这里经过让主线程休眠一段时间(固然实际开发通常不会这么作)以确保应用程序不会当即退出。
            TimeUnit.MINUTES.sleep(Integer.MAX_VALUE);
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

其中result即表示当前接受的客户端的链接会话，与客户端的通讯都须要经过该链接会话进行。安全

Client端：服务器

public class SimpleAIOClient {

    public static void main(String[] args) {
        try {
            // 打开一个SocketChannel通道并获取AsynchronousSocketChannel实例
            AsynchronousSocketChannel client = AsynchronousSocketChannel.open();
            // 链接到服务器并处理链接结果
            client.connect(new InetSocketAddress("127.0.0.1", 5555), null, new CompletionHandler<Void, Void>() {
                @Override
                public void completed(final Void result, final Void attachment) {
                    System.out.println("成功链接到服务器!");
                    try {
                        // 给服务器发送信息并等待发送完成
                        client.write(ByteBuffer.wrap("From client:Hello i am client".getBytes()))
                              .get();
                        ByteBuffer readBuffer = ByteBuffer.allocate(128);
                        // 阻塞等待接收服务端数据
                        client.read(readBuffer)
                              .get();
                        System.out.println(new String(readBuffer.array()));
                    } catch (InterruptedException | ExecutionException e) {
                        e.printStackTrace();
                    }
                }

                @Override
                public void failed(final Throwable exc, final Void attachment) {
                    exc.printStackTrace();
                }
            });
            TimeUnit.MINUTES.sleep(Integer.MAX_VALUE);
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

3. AIO主要API详解

从第2节例子能够看到，实现一个最简单的AIO socket通讯server、client，主要须要这些相关的类和接口：网络

AsynchronousServerSocketChannel
服务端Socket通道类，负责服务端Socket的建立和监听；异步
AsynchronousSocketChannel
客户端Socket通道类，负责客户端消息读写；socket
CompletionHandler<A,V>
消息处理回调接口，是一个负责消费异步IO操做结果的消息处理器；ide
ByteBuffer
负责承载通讯过程当中须要读、写的消息。this

此外，还有可选的用于异步通道资源共享的AsynchronousChannelGroup类，接下来将一一介绍这些类的主要接口及使用。线程

3.1.1 AsynchronousServerSocketChannel

AsynchronousServerSocketChannel是一个 流式监听套接字的异步通道。

AsynchronousServerSocketChannel的使用须要通过三个步骤：建立/打开通道、绑定地址和端口和监听客户端链接请求。

1、建立/打开通道：简单地，能够经过调用AsynchronousServerSocketChannel的静态方法open()来建立AsynchronousServerSocketChannel实例：

try {
  AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open();
} catch (IOException e) {
  e.printStackTrace();
}

当打开通道失败时，会抛出一个IOException异常。AsynchronousServerSocketChannel提供了设置通道分组(AsynchronousChannelGroup)的功能，以实现组内通道资源共享。能够调用open(AsynchronousChannelGroup)重载方法建立指定分组的通道：

try {
  ExecutorService pool = Executors.newCachedThreadPool();
  AsynchronousChannelGroup group = AsynchronousChannelGroup.withCachedThreadPool(pool, 10);
  AsynchronousServerSocketChannel serverSocketChannel = AsynchronousServerSocketChannel.open(group);
} catch (IOException e) {
  e.printStackTrace();
}

AsynchronousChannelGroup封装了处理由绑定到组的异步通道所触发的I/O操做完成所需的机制。每一个AsynchronousChannelGroup关联了一个被用于提交处理I/O事件和分发消费在组内通道上执行的异步操做结果的completion-handlers的线程池。除了处理I/O事件，该线程池还有可能处理其余一些用于支持完成异步I/O操做的任务。从上面例子能够看到，经过指定AsynchronousChannelGroup的方式打开AsynchronousServerSocketChannel，能够定制server channel执行的线程池。有关AsynchronousChannelGroup的详细介绍能够查看官方文档注释。若是不指定AsynchronousChannelGroup，则AsynchronousServerSocketChannel会归类到一个默认的分组中。

2、绑定地址和端口：经过调用AsynchronousServerSocketChannel.bind(SocketAddress)方法来绑定监听地址和端口：

// 构建一个InetSocketAddress实例以指定监听的地址和端口，若是须要指定ip，则调用InetSocketAddress(ip,port)构造方法建立便可
serverSocketChannel.bind(new InetSocketAddress(port));

3、监听和接收客户端链接请求：

监听客户端链接请求，主要经过调用AsynchronousServerSocketChannel.accept()方法完成。accept()有两个重载方法：

public abstract <A> void accept(A,CompletionHandler<AsynchronousSocketChannel,? super A>);
public abstract Future<AsynchronousSocketChannel> accept();

这两个重载方法的行为方式彻底相同，事实上，AIO的不少异步API都封装了诸如此类的重载方法：提供CompletionHandle回调参数或者返回一个Future<T>类型变量。用过Feture接口的都知道，能够调用Feture.get()方法阻塞等待调用结果。以第一个重载方法为例，当接受一个新的客户端链接，或者accept操做发生异常时，会经过CompletionHandler将结果返回给用户处理：

serverSocketChannel
.accept(serverSocketChannel, new CompletionHandler<AsynchronousSocketChannel,
        AsynchronousServerSocketChannel>() {
          @Override
          public void completed(final AsynchronousSocketChannel result,
                                final AsynchronousServerSocketChannel attachment) {
            // 接收到新的客户端链接时回调
            // result即和该客户端的链接会话
            // 此时能够经过result与客户端进行交互
          }

          @Override
          public void failed(final Throwable exc, final AsynchronousServerSocketChannel attachment) {
            // accept失败时回调
          }
        });

须要注意的是，AsynchronousServerSocketChannel是线程安全的，但在任什么时候候同一时间内只能容许有一个accept操做。所以，必须得等待前一个accept操做完成以后才能启动下一个accept：

serverSocketChannel
.accept(serverSocketChannel, new CompletionHandler<AsynchronousSocketChannel,
        AsynchronousServerSocketChannel>() {
          @Override
          public void completed(final AsynchronousSocketChannel result,
                                final AsynchronousServerSocketChannel attachment) {
            // 接收到新的客户端链接，此时本次accept已经完成
            // 继续监听下一个客户端链接到来
            serverSocketChannel.accept(serverSocketChannel,this);
            // result即和该客户端的链接会话
            // 此时能够经过result与客户端进行交互
          }
          ...
        });

此外，还能够经过如下方法获取和设置AsynchronousServerSocketChannel的socket选项：

// 设置socket选项
serverSocketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE,true);
// 获取socket选项设置
boolean keepAlive = serverSocketChannel.getOption(StandardSocketOptions.SO_KEEPALIVE);

其中StandardSocketOptions类封装了经常使用的socket设置选项。

获取本地地址：

InetSocketAddress address = (InetSocketAddress) serverSocketChannel.getLocalAddress();

3.1.2 AsynchronousSocketChannel

AsynchronousSocketChannel是一个 流式链接套接字的异步通道。

AsynchronousSocketChannel表示服务端与客户端之间的链接通道。客户端能够经过调用AsynchronousSocketChannel静态方法open()建立，而服务端则经过调用AsynchronousServerSocketChannel.accept()方法后由AIO内部在合适的时候建立。下面以客户端实现为例，介绍AsynchronousSocketChannel。

1、建立AsynchronousSocketChannel并链接到服务端：须要经过open()建立和打开一个AsynchronousSocketChannel实例，再调用其connect()方法链接到服务端，接着才能够与服务端交互：

// 打开一个socket通道
AsynchronousSocketChannel socketChannel = AsynchronousSocketChannel.open();
// 阻塞等待链接成功
socketChannel.connect(new InetSocketAddress(ip,port)).get();
// 链接成功，接下来能够进行read、write操做

同AsynchronousServerSocketChannel，AsynchronousSocketChannel也提供了open(AsynchronousChannelGroup)方法用于指定通道分组和定制线程池。socketChannel.connect()也提供了CompletionHandler回调和Future返回值两个重载方法，上面例子使用带Future返回值的重载，并调用get()方法阻塞等待链接创建完成。

2、发送消息：

能够构建一个ByteBuffer对象并调用socketChannel.write(ByteBuffer)方法异步发送消息，并经过CompletionHandler回调接收处理发送结果：

ByteBuffer writeBuf = ByteBuffer.wrap("From socketChannel:Hello i am socketChannel".getBytes());
socketChannel.write(writeBuf, null, new CompletionHandler<Integer, Object>() {
  @Override
  public void completed(final Integer result, final Object attachment) {
    // 发送完成，result：总共写入的字节数
  }

  @Override
  public void failed(final Throwable exc, final Object attachment) {
    // 发送失败
  }
});

3、读取消息：

构建一个指定接收长度的ByteBuffer用于接收数据，调用socketChannel.read()方法读取消息并经过CompletionHandler处理读取结果：

ByteBuffer readBuffer = ByteBuffer.allocate(128);
socketChannel.read(readBuffer, null, new CompletionHandler<Integer, Object>() {
  @Override
  public void completed(final Integer result, final Object attachment) {
    // 读取完成,result:实际读取的字节数。若是通道中没有数据可读则result=-1。
  }

  @Override
  public void failed(final Throwable exc, final Object attachment) {
    // 读取失败
  }
});

此外，AsynchronousSocketChannel也封装了设置/获取socket选项的方法：

// 设置socket选项
socketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE,true);
// 获取socket选项设置
boolean keepAlive = socketChannel.getOption(StandardSocketOptions.SO_KEEPALIVE);

3.1.3 CompletionHandler

CompletionHandler是一个用于 消费异步I/O操做结果的处理器。

AIO中定义的异步通道容许指定一个CompletionHandler处理器消费一个异步操做的结果。从上文中也能够看到，AIO中大部分的异步I/O操做接口都封装了一个带CompletionHandler类型参数的重载方法，使用CompletionHandler能够很方便地处理AIO中的异步I/O操做结果。CompletionHandler是一个具备两个泛型类型参数的接口,声明了两个接口方法：

public interface CompletionHandler<V,A> {
    void completed(V result, A attachment);
    void failed(Throwable exc, A attachment);
}

其中，泛型V表示I/O操做的结果类型，经过该类型参数消费I/O操做的结果；泛型A为附加到I/O操做中的对象类型，能够经过该类型参数将须要的变量传入到CompletionHandler实现中使用。所以，AIO中大部分的异步I/O操做都有一个相似这样的重载方法：

<V,A> void ioOperate(params,A attachment,CompletionHandler<V,A> handler);

例如，AsynchronousServerSocketChannel.accept()方法：

public abstract <A> void accept(A attachment,CompletionHandler<AsynchronousSocketChannel,? super A> handler);

AsynchronousSocketChannel.write()方法等:

public final <A> void write(ByteBuffer src,A attachment,CompletionHandler<Integer,? super A> handler)

当I/O操做成功完成时，会回调到completed方法，failed方法则在I/O操做失败时被回调。须要注意的是：在CompletionHandler的实现中应立即使处理操做结果，以免一直占用调用线程而不能分发其余的CompletionHandler处理器。

4 The End :)

1. java中的AIO
2. Java BIO NIO AIO
3. Java NIO BIO AIO
4. java --- BIO NIO AIO
5. JAVA NIO AIO
6. java NIO AIO
7. [Java]socket Aio demo
8. java aio 编程
9. Java——BIO、NIO、AIO
10. Java BIO、NIO和AIO。
更多相关文章...
• SQLite - Java - SQLite教程
• 在Java程序中使用Redis - Redis教程
• Java Agent入门实战（一）-Instrumentation介绍与使用
• Java Agent入门实战（三）-JVM Attach原理与使用