AIO 是彻底的异步通信，NIO 是同步非阻塞通信，AIO是真正的非阻塞模型，但相比NIO并没有明显的性能提高，因此NIO目前依旧是主流
image.png
AIO操作流程

解释：

假设有这么一个场景，有一排水壶（客户）在烧水。AIO的做法是，每个水壶上装一个开关，当水开了以后会提醒对应的线程去处理。NIO的做法是，叫一个线程不停的循环观察每一个水壶，根据每个水壶当前的状态去处理。BIO的做法是，叫一个线程停留在一个水壶那，直到这个水壶烧开，才去处理下一个水壶

案例演示：

做一个类似回音壁的AIO案例，服务端收到客户端消息后直接将消息再返回给客户端

流程分析：

1、服务器端创建 AsynchronousServerSocketChannel （异步服务器端通道）绑定端口和地址，默认属于AsynchronousChannelGroup 通道组（可以被多个异步通道进行资源共享的群组）
2、服务端创建 服务端CompletionHandler 和 客户端ClientHandler 并使用 accept 方法异步等待客户端连接，当有客户端加入时将会触发 服务端CompletionHandler 的 completed 方法，在该方法内通知服务端接收连接请求
3、客户端创建 AsynchronousSocketChannel（异步客户端通道），绑定端口和地址，建立与服务端的连接后默认也属于AsynchronousChannelGroup
4、客户端输入信息后将数据写入 AsynchronousSocketChannel，服务端拿到客户端发送的数据后传给客户端ClientHandler ，根据额外参数判断并返回给客户端 AsynchronousSocketChannel
5、每一个新加入到服务端的 AsynchronousSocketChannel 实际都会创建一个 Handler 用来处理与其对应的读写IO事件
![C2)I@Z%5}Z(HMEP)@87T7T.png

服务端代码：

  1. package study;
  3. import java.io.*;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.AsynchronousServerSocketChannel;
  7. import java.nio.channels.AsynchronousSocketChannel;
  8. import java.nio.channels.CompletionHandler;
  9. import java.util.HashMap;
  10. import java.util.Map;
  13. /**
  14.  * 服务端
  15.  */
  16. public class Server {
  18.     final String LOCALHOST = "localhost";
  19.     final int DEFAULT_PORT = 8888;
  20.     AsynchronousServerSocketChannel serverChannel;
  23.     /* 关闭资源流 */
  24.     private void close(Closeable closable) {
  25.         if (closable != null) {
  26.             try {
  27.                 closable.close();
  28.                 System.out.println("关闭" + closable);
  29.             } catch (IOException e) {
  30.                 e.printStackTrace();
  31.             }
  32.         }
  33.     }
  36.     public void start() {
  37.         try {
  38.             serverChannel = AsynchronousServerSocketChannel.open();                  //创建AIO通道，默认使用AsynchronousChannelGroup
  39.             serverChannel.bind(new InetSocketAddress(LOCALHOST, DEFAULT_PORT));      //绑定监听端口
  40.             System.out.println("启动服务器，监听端口:" + DEFAULT_PORT);
  42.             /* accept 判断后立马返回，直到客户端发送消息后 AcceptHandler 方法才会被系统调用返回数据
  43.              * 通过accept函数返回的Feture对象 或者 有回调函数CompletionHandler作为参数的accept方法与客户端建立连接 ，这里使用自定义的 CompletionHandler 作为参数的方式进行异步
  44.              * 参数一: 在回调的时候添加额外的数据，非必须，可以理解为邮件发送的附件
  45.              * 参数二: 回调函数的 CompletionHandler
  46.              */
  47.             while (true) {
  48.                 serverChannel.accept(null, new AcceptHandler());        //异步等待新客户端连接
  49.                 System.in.read();
  50.             }
  51.         } catch (IOException e) {
  52.             e.printStackTrace();
  53.         } finally {
  54.             close(serverChannel);
  55.         }
  56.     }
  59.     /* 参数一: IO操作的返回结果，由于是服务端调用，返回的应该是客户端对象，所以是 AsynchronousSocketChannel，
  60.      * 参数二：accept方法的参数一
  61.      */
  62.     private class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object> {
  64.         /* 异步调用的函数有数据返回时调用该方法
  65.         *       AsynchronousSocketChannel : 用于服务器和客户端之间收发信息
  66.         * */
  67.         @Override
  68.         public void completed(AsynchronousSocketChannel result, Object attachment) {
  69.             /* 判断服务端channel是否处于正常开放状态 */
  70.             if (serverChannel.isOpen()) {
  71.                 serverChannel.accept(null, this);   //通知服务端接收客户端连接
  72.             }
  74.             AsynchronousSocketChannel clientChannel = result;        //获取异步的客户端通道
  75.             /* 判断客户端Channle是否处于开放状态 */
  76.             if (clientChannel != null && clientChannel.isOpen()) {
  77.                 ClientHandler handler = new ClientHandler(clientChannel);
  79.                 ByteBuffer buffer = ByteBuffer.allocate(1024);
  80.                 /* 使用额外参数让 ClientHandler 判断是读操作还是写操作 */
  81.                 Map<String, Object> info = new HashMap<>();
  82.                 info.put("type", "read");
  83.                 info.put("buffer", buffer);
  84.                 clientChannel.read(buffer, info, handler);   //从客户端通道读取输入信息返回给ClientHandler: 参数一 > 缓存区,参数二 > 添加到客户端Handler回调函数的额外参数,参数三 > 客户端Handler
  86.             }
  87.         }
  90.         /* 异步调用的操作有错误时调用该方法 */
  91.         @Override
  92.         public void failed(Throwable exc, Object attachment) {
  94.         }
  96.     }
  99.     /* 客户端Handler,处理 clientChannel 异步调用读&写产生的结果
  100.      *  参数一: IO操作的函数返回的数据类型,因为是字节数所以是Integer
  101.      *  参数二：传给handler的额外参数
  102.      */
  103.     private class ClientHandler implements CompletionHandler<Integer, Object>{
  105.         private AsynchronousSocketChannel clientChannel;
  107.         public ClientHandler(AsynchronousSocketChannel channel) {
  108.             this.clientChannel = channel;
  109.         }
  111.         /* 异步调用的函数有数据返回时调用该方法 */
  112.         @Override
  113.         public void completed(Integer result, Object attachment) {
  114.             Map<String, Object> info = (Map<String, Object>) attachment;         //获取服务端的额外信息
  115.             String type = (String) info.get("type");
  116.             /* 接收客户端数据再返回给客户端 */
  117.             if ("read".equals(type)) {
  118.                 ByteBuffer buffer = (ByteBuffer) info.get("buffer");             //从客户端Buffer读取数据
  119.                 buffer.flip();
  120.                 info.put("type", "write");
  121.                 clientChannel.write(buffer, info, this);
  122.                 buffer.clear();
  123.             }
  124.             /* 监听客户端发送的数据 */
  125.             if ("write".equals(type)) {
  126.                 ByteBuffer buffer = ByteBuffer.allocate(1024);                  //从客户端Buffer读取数据
  127.                 info.put("type", "read");
  128.                 info.put("buffer", buffer);
  129.                 clientChannel.read(buffer, info, this);
  130.             }
  131.         }
  133.         @Override
  134.         public void failed(Throwable exc, Object attachment) {
  135.             // 处理错误
  136.         }
  138.     }
  141.     public static void main(String[] args) {
  142.         Server server = new Server();
  143.         server.start();
  144.     }
  147. }

客户端代码：

  1. package study;
  3. import java.io.*;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.AsynchronousSocketChannel;
  7. import java.util.concurrent.ExecutionException;
  8. import java.util.concurrent.Future;
  10. public class Client {
  12.     final String LOCALHOST = "localhost";
  13.     final int DEFAULT_PORT = 8888;
  14.     AsynchronousSocketChannel clientChannel;
  17.     private void close(Closeable closable) {
  18.         if (closable != null) {
  19.             try {
  20.                 closable.close();
  21.                 System.out.println("关闭" + closable);
  22.             } catch (IOException e) {
  23.                 e.printStackTrace();
  24.             }
  25.         }
  26.     }
  29.     public void start() {
  30.         try {
  31.             clientChannel = AsynchronousSocketChannel.open();                                                   //创建channel
  32.             Future<Void> future = clientChannel.connect(new InetSocketAddress(LOCALHOST, DEFAULT_PORT));        //绑定端口
  33.             future.get();       //进行异步的服务端连接
  35.             BufferedReader consoleReader = new BufferedReader(new InputStreamReader(System.in));                //等待用户的输入
  36.             while (true) {
  37.                 String input = consoleReader.readLine();
  38.                 byte[] inputBytes = input.getBytes();
  39.                 ByteBuffer buffer = ByteBuffer.wrap(inputBytes);        //wrap写入数据后会自动flip
  41.                 Future<Integer> writeResult = clientChannel.write(buffer);
  42.                 writeResult.get();          //有返回值则认为用户数据成功写入到通道并发送给服务器
  43.                 buffer.flip();
  44.                 Future<Integer> readResult = clientChannel.read(buffer);
  45.                 readResult.get();           //有返回值则认为获取到服务端返回的数据并写入到Buffer
  47.                 String echo = new String(buffer.array());
  48.                 buffer.clear();
  49.                 System.out.println(echo);
  50.             }
  52.         } catch (IOException e) {
  53.             e.printStackTrace();
  54.         } catch (InterruptedException e) {
  55.             e.printStackTrace();
  56.         } catch (ExecutionException e) {
  57.             e.printStackTrace();
  58.         } finally {
  59.             close(clientChannel);
  60.         }
  61.     }
  64.     public static void main(String[] args) {
  65.         Client client = new Client();
  66.         client.start();
  67.     }
  70. }

测试：

启动客户端和服务端，客户端输入信息后服务端返回相同数据
image.png
测试结果

AIO聊天室：

服务端代码：

  1. package chatroom;
  4. import java.io.Closeable;
  5. import java.io.IOException;
  6. import java.net.InetSocketAddress;
  7. import java.nio.ByteBuffer;
  8. import java.nio.CharBuffer;
  9. import java.nio.channels.AsynchronousChannelGroup;
  10. import java.nio.channels.AsynchronousServerSocketChannel;
  11. import java.nio.channels.AsynchronousSocketChannel;
  12. import java.nio.channels.CompletionHandler;
  13. import java.nio.charset.Charset;
  14. import java.util.ArrayList;
  15. import java.util.List;
  16. import java.util.concurrent.ArrayBlockingQueue;
  17. import java.util.concurrent.ThreadPoolExecutor;
  18. import java.util.concurrent.TimeUnit;
  21. /* AIO聊天室服务端 */
  22. public class ChatServer {
  24.     private static final String LOCALHOST = "localhost";
  25.     private static final int DEFAULT_PORT = 8888;
  26.     private static final String QUIT = "quit";
  27.     private static final int BUFFER = 1024;
  29.     private AsynchronousChannelGroup channelGroup;                  //线程组
  30.     private AsynchronousServerSocketChannel serverChannel;          //服务端通道
  31.     private Charset charset = Charset. forName("UTF-8");
  32.     private int port;
  33.     private List<ClientHandler> connectedClients;                   //在线客户列表
  35.     public ChatServer(){
  36.         this(DEFAULT_PORT);
  37.     }
  39.     /* 用户自定义监听端口 */
  40.     public ChatServer(int port){
  41.         this.port = port;
  42.         this.connectedClients = new ArrayList<>();
  43.     }
  46.     /* 退出操作 */
  47.     private void close(Closeable closable) {
  48.         if (closable != null) {
  49.             try {
  50.                 closable.close();
  51.             } catch (IOException e) {
  52.                 e.printStackTrace();
  53.             }
  54.         }
  55.     }
  58.     private void start(){
  60.         ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5,8,
  61.                 200, TimeUnit.MILLISECONDS,
  62.                 new ArrayBlockingQueue<>(10));
  63.         try {
  64.             channelGroup = AsynchronousChannelGroup.withThreadPool(threadPool);     //创建一个使用自定义线程池的
  65.             serverChannel = AsynchronousServerSocketChannel.open(channelGroup);     //使用自定义的channelGroup
  66.             serverChannel.bind(new InetSocketAddress(LOCALHOST,port));              //绑定监听地址和端口
  67.             System.out.println("启动服务器,监听端口: "+port);
  69.             while (true){
  70.                 serverChannel.accept(null, new AcceptHandler());         //等待客户端连接
  71.                 System.in.read();
  72.             }
  74.         } catch (IOException e) {
  75.             e.printStackTrace();
  76.         }finally{
  77.             close(serverChannel);
  78.         }
  80.     }
  83.     public static void main(String[] args) {
  84.         ChatServer chatServer = new ChatServer(6767);
  85.         chatServer.start();
  86.     }
  89.     /* 服务端 Handler */
  90.     private class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel,Object>{
  92.         @Override
  93.         public void completed(AsynchronousSocketChannel clientChannel, Object attachment) {
  94.             /* 判断服务端channel是否处于正常开放状态 */
  95.             if (serverChannel.isOpen()) {
  96.                 serverChannel.accept(null, this);   //通知服务端接收客户端连接
  97.             }
  99.             /* 判断客户端Channle是否处于开放状态 */
  100.             if (clientChannel != null && clientChannel.isOpen()) {
  101.                 ClientHandler handler = new ClientHandler(clientChannel);
  102.                 ByteBuffer buffer = ByteBuffer.allocate(BUFFER);
  103.                 addClient(handler);         //将新用户加入到在线用户列表
  104.                 clientChannel.read(buffer, buffer, handler);   //从客户端通道读取输入信息返回给ClientHandler: 参数一 > 缓存区,参数二 > 添加到客户端Handler回调函数的额外参数,参数三 > 客户端Handler
  105.             }
  106.         }
  108.         @Override
  109.         public void failed(Throwable exc, Object attachment) {
  110.             System.out.println("连接失败: "+exc);
  111.         }
  113.     }
  116.     /* 将新用户加入到在线用户列表 */
  117.     private synchronized void addClient(ClientHandler handler) {
  118.         connectedClients.add(handler);
  119.         System.out.println(getClientName(handler.clientChannel)+"已连接到服务器");
  120.     }
  123.     /* 用户下线，从在线用户列表移除 */
  124.     private synchronized void removeClient(ClientHandler handler) {
  125.         connectedClients.remove(handler);
  126.         System.out.println(getClientName(handler.clientChannel)+"已断开连接");
  127.         close(handler.clientChannel);
  128.     }
  131.     /*转换为字符串*/
  132.     private String receive(ByteBuffer buffer){
  133.         CharBuffer decode = charset.decode(buffer);
  134.         return String.valueOf(decode);
  135.     }
  138.     /* 客户端Handler,处理 clientChannel 异步调用读&写产生的结果
  139.      *  参数一: IO操作的数据类型 > 字节数
  140.      *  参数二: 传给handler的额外参数
  141.      */
  142.     private class ClientHandler implements CompletionHandler<Integer,Object>{
  144.         private AsynchronousSocketChannel clientChannel;
  146.         public ClientHandler(AsynchronousSocketChannel channel){
  147.             this.clientChannel = channel;
  148.         }
  150.         @Override
  151.         public void completed(Integer result, Object attachment) {
  152.             ByteBuffer buffer = (ByteBuffer) attachment;        //read函数调用才有的对象,write是没有额外对象的
  153.             if(null!=buffer){
  154.                 if(result<=0){
  155.                     removeClient(this);//客户端异常,将客户移除出在线客户列表
  156.                 }else{
  157.                     buffer.flip();
  158.                     String fwdMsg = receive(buffer);
  159.                     System.out.println(getClientName(clientChannel)+":"+fwdMsg);
  160.                     forwardMessage(clientChannel,fwdMsg);   //转发信息
  161.                     buffer.clear();
  163.                     /** 检查用户是否退出 */
  164.                     if(QUIT.equals(fwdMsg)){
  165.                         removeClient(this);
  166.                     }else{
  167.                         clientChannel.read(buffer,buffer,this);     //继续接收用户后续消息
  168.                     }
  170.                 }
  171.             }
  172.         }
  175.         @Override
  176.         public void failed(Throwable exc, Object attachment) {
  177.             System.out.println("读写失败: "+exc);
  178.         }
  180.     }
  183.     /* 将消息转发给在线用户列表中的其他用户 */
  184.     private synchronized void forwardMessage(AsynchronousSocketChannel clientChannel, String fwdMsg) {
  185.         for (ClientHandler handler : connectedClients) {
  186.            if(!clientChannel.equals(handler.clientChannel)){
  187.                try {
  188.                    ByteBuffer buffer = charset.encode(getClientName(handler.clientChannel) + ":" + fwdMsg);
  189.                    handler.clientChannel.write(buffer,null,handler);
  190.                } catch (Exception e) {
  191.                    System.out.println(handler.clientChannel+"消息转发错误!");
  192.                    e.printStackTrace();
  193.                }
  194.            }
  195.         }
  196.     }
  199.     /* 获取对应客户端对象,返回客户端ID */
  200.     private String getClientName(AsynchronousSocketChannel clientChannel) {
  201.         int clientPort = -1;
  202.         try {
  203.             InetSocketAddress address = (InetSocketAddress) clientChannel.getRemoteAddress();
  204.             clientPort = address.getPort();              //获取客户端端口
  205.         } catch (IOException e) {
  206.             e.printStackTrace();
  207.         }
  208.         return "客户端["+clientPort+"]";
  209.     }
  212. }

客户端代码：

  1. package chatroom;
  4. import java.io.Closeable;
  5. import java.io.IOException;
  6. import java.net.InetSocketAddress;
  7. import java.nio.ByteBuffer;
  8. import java.nio.channels.AsynchronousSocketChannel;
  9. import java.nio.charset.Charset;
  10. import java.util.concurrent.ExecutionException;
  11. import java.util.concurrent.Future;
  14. /* AIO聊天室客户端 */
  15. public class ChatClient {
  17.     private static final String LOCALHOST = "localhost";
  18.     private static final int DEFAULT_PORT = 8888;
  19.     private static final int BUFFER = 1024;
  21.     private String host;            //用户定义地址
  22.     private int port;               //用户定义端口
  23.     private AsynchronousSocketChannel clientChannel;
  24.     private Charset chaset = Charset.forName("UTF-8");
  27.     public ChatClient(){
  28.         this(LOCALHOST,DEFAULT_PORT);
  29.     }
  31.     public ChatClient(String host,int port){
  32.         this.port = port;
  33.         this.host = host;
  34.     }
  37.     /* 退出操作 */
  38.     private void close(Closeable closable) {
  39.         if (closable != null) {
  40.             try {
  41.                 closable.close();
  42.             } catch (IOException e) {
  43.                 e.printStackTrace();
  44.             }
  45.         }
  46.     }
  49.     private void start(){
  50.         try {
  51.             clientChannel = AsynchronousSocketChannel.open();              //创建Channel
  52.             Future<Void> future = clientChannel.connect(new InetSocketAddress(host,port));       //绑定端口
  53.             future.get();       //进行异步的服务端连接
  55.             /* 启用新线程处理用户输入 */
  56.             new Thread(new UserInputHander(this)).start();
  58.             ByteBuffer buffer = ByteBuffer.allocate(BUFFER);        //创建缓存
  59.             while(true){
  60.                 Future<Integer> readResult = clientChannel.read(buffer);
  61.                 int result = readResult.get();                 //等待返回读取结果
  62.                 if(result<=0){
  63.                     System.out.println("已断开与服务器的连接");
  64.                     close(clientChannel);
  65.                     System.exit(1);     //中止当前虚拟机的运行进行强制性的程序退出
  66.                 }else{
  67.                     buffer.flip();
  68.                     String msg = String.valueOf(chaset.decode(buffer));
  69.                     buffer.clear();
  70.                     System.out.println("服务器返回数据: "+msg);
  71.                 }
  72.             }
  74.         } catch (Exception e) {
  75.             e.printStackTrace();
  76.         }
  77.     }
  80.     public void send(String msg){
  81.         /* 误输入空格 */
  82.         if(msg.isEmpty()){
  83.             return;
  84.         }
  85.         /* 有意义的数据 */
  86.         ByteBuffer buffer = chaset.encode(msg);
  87.         Future<Integer> writeResult = clientChannel.write(buffer);
  88.         try {
  89.             writeResult.get();
  90.         } catch (Exception e) {
  91.             System.out.println("发送消息失败!");
  92.             e.printStackTrace();
  93.         }
  94.     }
  97.     public static void main(String[] args) {
  98.         ChatClient chatClient = new ChatClient(LOCALHOST,6767);
  99.         chatClient.start();
  100.     }
  103. }

客户端消息发送方法：

  1. package chatroom;
  3. import java.io.BufferedReader;
  4. import java.io.IOException;
  5. import java.io.InputStreamReader;
  8. /* 客户端消息发送 */
  9. public class UserInputHander implements Runnable {
  11.     private ChatClient chatClient;
  13.     public UserInputHander(ChatClient chatClient){
  14.         this.chatClient = chatClient;
  15.     }
  17.     @Override
  18.     public void run() {
  19.         try {
  20.             BufferedReader consoleRerader = new BufferedReader(new InputStreamReader(System.in));
  21.             while (true){
  22.                 String input = consoleRerader.readLine();
  23.                 //向服务器发送消息
  24.                 chatClient.send(input);
  25.                 //检查用户是否准备退出
  26.                 if("quit".equals(input)){
  27.                     break;
  28.                 }
  29.             }
  30.         } catch (IOException e) {
  31.             e.printStackTrace();
  32.         }
  33.     }
  36. }

测试：

启动 ChatServer 和多个 ChatClient 实例进行测试，在客户端窗口中发送消息进行测试
image.png