一、AIO的异步操作

AsyncSocketChannel和AsyncServerSocketChannel都是支持如图几个IO操作异步调用的。
image.png

二、如何实现异步调用

Future

简单来说通过Channel来调用那几个函数，然后返回一个Future的对象，Future在线程池中介绍过，他是一个描述未来的一个对象，通过get、isDone等等的调用查询未来的任务。
image.png

CompletionHander

通过Channel调用IO操作，然后不等待是否完成，然后把一些参数传入一个回调函数（Handler）进一步操作（实现）。
image.png
Handler拥有两种函数，一个是Completed，是在完成后调用，另一种是Failed，是在失败了后调用。

三、EchoDemo实现异步操作机制

Server：

主体函数：

  1. package demo4.aio.server;
  3. import java.io.Closeable;
  4. import java.io.IOException;
  5. import java.net.InetSocketAddress;
  6. import java.nio.ByteBuffer;
  7. import java.nio.channels.AsynchronousServerSocketChannel;
  8. import java.nio.channels.AsynchronousSocketChannel;
  9. import java.nio.channels.CompletionHandler;
  10. import java.util.HashMap;
  11. import java.util.Map;
  13. public class Server {
  14.     private String LOCALHOST="localhost";
  15.     private static final int DEFAULT_PORT=9999;
  16.     private AsynchronousServerSocketChannel serverSocketChannel;
  17.     private static final String QUIT="quit";
  19.     private boolean readyToQuit(String str){
  20.         return QUIT.equalsIgnoreCase(str);
  21.     }
  23.     private void shutDown(Closeable...closeables){
  24.         try {
  25.             for (Closeable shut : closeables) {
  26.                 if (shut != null) {
  27.                     shut.close();
  28.                 }
  29.             }
  30.         } catch (IOException e) {
  31.             e.printStackTrace();
  32.         }
  33.     }
  35.     public void start(){
  36.         try {
  37.             //绑定监听端口
  38.             serverSocketChannel=AsynchronousServerSocketChannel.open();
  39.             serverSocketChannel.bind(new InetSocketAddress(LOCALHOST,DEFAULT_PORT));
  40.             System.out.println("服务器已启动，正在监听：["+LOCALHOST+","+DEFAULT_PORT+"]");
  41.             while (true) {
  42.                 //AcceptHandler在AsynchronousChannelGroup线程池中完成,而非主线程。
  43.                 //AsynchronousChannelGroup未定义时使用默认的AsynchronousChannelGroup
  44.                 serverSocketChannel.accept(null, new AcceptHandler());
  45.                 //防止频繁调用accept函数。
  46.                 System.in.read();
  47.             }
  48.         } catch (IOException e) {
  49.             e.printStackTrace();
  50.         }finally {
  51.             shutDown(serverSocketChannel);
  52.         }
  53.     }
  55.     private class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel,Object> {
  56.         @Override
  57.         public void completed(AsynchronousSocketChannel result, Object attachment) {
  58.             //并不会造成stepOverFlow在底层实现拥有次数限制
  59.             if (serverSocketChannel.isOpen()){
  60.                 serverSocketChannel.accept(null,this);
  61.             }
  63.             AsynchronousSocketChannel clientChannel=result;
  64.             if (clientChannel!=null&&clientChannel.isOpen()) {
  65.                 ClientHander hander = new ClientHander(clientChannel);
  66.                 ByteBuffer buffer = ByteBuffer.allocate(1024);
  67.                 Map<String, Object> info = new HashMap<>();
  68.                 info.put("type", "read");
  69.                 info.put("buffer", buffer);
  70.                 clientChannel.read(buffer, info, hander);
  71.             }
  72.         }
  74.         @Override
  75.         public void failed(Throwable exc, Object attachment) {
  76.             //异常处理,略
  77.         }
  78.     }
  80.     private class ClientHander implements CompletionHandler<Integer,Object> {
  81.         private AsynchronousSocketChannel clientChannel;
  83.         public ClientHander(AsynchronousSocketChannel clientChannel) {
  84.             try {
  85.                 System.out.println("客户端["+clientChannel.getLocalAddress()+"]已连接到服务器");
  86.             } catch (IOException e) {
  87.                 e.printStackTrace();
  88.             }
  89.             this.clientChannel=clientChannel;
  90.         }
  92.         @Override
  93.         public void completed(Integer result, Object attachment) {
  94.             Map<String,Object> info= (Map<String, Object>) attachment;
  95.             String type= (String) info.get("type");
  96.             if ("read".equalsIgnoreCase(type)) {
  97.                 try {
  98.                     ByteBuffer buffer = (ByteBuffer) info.get("buffer");
  99.                     String str = new String(buffer.array());
  100.                     if (readyToQuit(str)) {
  101.                         System.out.println("客户端[" + clientChannel.getLocalAddress() + "]已断开连接");
  102.                         shutDown(clientChannel);
  103.                     } else {
  104.                         buffer.flip();
  105.                         info.put("type", "write");
  106.                         clientChannel.write(buffer, info, this);
  107.                         buffer.clear();
  108.                         System.out.println("客户端[" + clientChannel.getLocalAddress() + "]:" + str);
  109.                     }
  110.                 } catch (IOException e) {
  111.                     e.printStackTrace();
  112.                 }
  113.             }
  114.             if ("write".equalsIgnoreCase(type)){
  115.                 ByteBuffer buffer= ByteBuffer.allocate(1024);
  116.                 info.put("type","read");
  117.                 info.put("buffer",buffer);
  118.                 clientChannel.read(buffer,info,this);
  119.             }
  120.         }
  122.         @Override
  123.         public void failed(Throwable exc, Object attachment) {
  124.             //异常处理,略
  125.         }
  126.     }
  127. }

Main:

  1. package demo4.aio.server;
  3. public class ServerMain {
  4.     public static void main(String[] args) {
  5.         Server server=new Server();
  6.         server.start();
  7.     }
  8. }

Client

主题函数：

  1. package demo4.aio.client;
  3. import java.io.BufferedReader;
  4. import java.io.Closeable;
  5. import java.io.IOException;
  6. import java.io.InputStreamReader;
  7. import java.net.InetSocketAddress;
  8. import java.nio.ByteBuffer;
  9. import java.nio.channels.AsynchronousSocketChannel;
  10. import java.util.concurrent.ExecutionException;
  11. import java.util.concurrent.Future;
  13. public class Client {
  14.     private String LOCALHOST = "localhost";
  15.     private static final int DEFAULT_PORT = 9999;
  16.     private AsynchronousSocketChannel clientChannel;
  17.     private static final String QUIT="quit";
  19.     private boolean readyToQuit(String str){
  20.         return QUIT.equalsIgnoreCase(str);
  21.     }
  23.     private void shutDown(Closeable... closeables) {
  24.         try {
  25.             for (Closeable shut : closeables) {
  26.                 if (shut != null) {
  27.                     shut.close();
  28.                 }
  29.             }
  30.         } catch (IOException e) {
  31.             e.printStackTrace();
  32.         }
  33.     }
  35.     public void start() {
  36.         try {
  37.             //创建channel
  38.             clientChannel = AsynchronousSocketChannel.open();
  39.             Future<Void> future = clientChannel.connect(new InetSocketAddress(LOCALHOST, DEFAULT_PORT));
  40.             future.get();
  41.             System.out.println("客户端["+LOCALHOST+"，"+DEFAULT_PORT+"]已连接到服务器");
  43.             //等待用户输入
  44.             BufferedReader consoleReader = new BufferedReader(new InputStreamReader(System.in));
  45.             while (true) {
  46.                 String intput = consoleReader.readLine();
  47.                 byte[] inputBytes = intput.getBytes();
  48.                 ByteBuffer buffer = ByteBuffer.wrap(inputBytes);
  49.                 Future<Integer> writeRes = clientChannel.write(buffer);
  50.                 writeRes.get();
  51.                 if (readyToQuit(intput)){
  52.                     break;
  53.                 }
  54.                 System.out.println("消息发送成功");
  56.                 buffer.flip();
  57.                 Future<Integer> readRes = clientChannel.read(buffer);
  58.                 readRes.get();
  59.                 System.out.print("收到服务器消息：");
  60.                 buffer.clear();
  61.                 String res = new String(buffer.array());
  62.                 res="[ECHO]:" +res;
  63.                 System.out.println(res);
  64.             }
  65.         } catch (IOException e) {
  66.             e.printStackTrace();
  67.         } catch (InterruptedException e) {
  68.             e.printStackTrace();
  69.         } catch (ExecutionException e) {
  70.             e.printStackTrace();
  71.         }finally {
  72.             shutDown(clientChannel);
  73.         }
  74.     }
  75. }

Main：

  1. package demo4.aio.client;
  3. public class ClientMain {
  4.     public static void main(String[] args) {
  5.         Client client=new Client();
  6.         client.start();
  7.     }
  8. }

四、结果展示

image.png