一、AIO编程模型梳理

回忆上一篇文章的AIO模型，这里做一些大概梳理。
首先在服务器端创建一部服务器通道，绑定监听端口
这里使用的AsynchronousServerSocketChannel 他其实属于一个通道群（AsynchronousChannelGroups），而这个通道群代表着一组可以被多个一部通道共享的资源群组。通过这个通道群来调用handler 。
当我们不做额外设定时候，系统会使用默认的通道群。
异步如何实现？
创建一个handler ，然后通过行为调用handler 来处理行为。
具体物理逻辑见上一篇文章

二、服务器创建

chatServer:

  1. package demo5.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.AsynchronousChannelGroup;
  8. import java.nio.channels.AsynchronousServerSocketChannel;
  9. import java.nio.channels.AsynchronousSocketChannel;
  10. import java.nio.channels.CompletionHandler;
  11. import java.nio.charset.Charset;
  12. import java.nio.charset.StandardCharsets;
  13. import java.util.ArrayList;
  14. import java.util.List;
  15. import java.util.concurrent.ExecutorService;
  16. import java.util.concurrent.Executors;
  18. public class ChatServer {
  19.     private static final String LOCALHOST = "localhost";
  20.     private static final int DEFAULT_PORT = 9999;
  21.     private static final String QUIT = "quit";
  22.     private static final int BUFFER = 1024;
  23.     private static final int THREADPOOL_SIZE = 8;
  25.     private AsynchronousServerSocketChannel serverSocketChannel;
  26.     private AsynchronousChannelGroup channelGroup;
  27.     private Charset charset = StandardCharsets.UTF_8;
  28.     private List<ClientHandler> connectedClients;
  29.     private int port;
  30.     private String host;
  33.     private void shutdown(Closeable... closeables) {
  34.         try {
  35.             for (Closeable shut : closeables) {
  36.                 if (shut != null) {
  37.                     shut.close();
  38.                 }
  39.             }
  40.         } catch (IOException e) {
  41.             e.printStackTrace();
  42.         }
  43.     }
  45.     private boolean readyToQuit(String str) {
  46.         return QUIT.equalsIgnoreCase(str);
  47.     }
  49.     public ChatServer(int port, String host) {
  50.         this.host = host;
  51.         this.port = port;
  52.         connectedClients = new ArrayList<>();
  53.     }
  55.     public ChatServer() {
  56.         this(DEFAULT_PORT, LOCALHOST);
  57.     }
  59.     public void start() {
  60.         try {
  61.             ExecutorService pool = Executors.newFixedThreadPool(THREADPOOL_SIZE);
  62.             channelGroup = AsynchronousChannelGroup.withThreadPool(pool);
  63.             serverSocketChannel = AsynchronousServerSocketChannel.open(channelGroup);
  64.             serverSocketChannel.bind(new InetSocketAddress(host, port));
  65.             System.out.println("服务器已启动，监听端口[" + serverSocketChannel.getLocalAddress() + "]");
  67.             while (true) {
  68.                 serverSocketChannel.accept(null, new AcceptHandler());
  69.                 System.in.read();
  70.             }
  71.         } catch (IOException e) {
  72.             e.printStackTrace();
  73.         } finally {
  74.             shutdown(serverSocketChannel);
  75.         }
  76.     }
  78.     private class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object> {
  80.         @Override
  81.         public void completed(AsynchronousSocketChannel clientChannel, Object attachment) {
  82.             if (serverSocketChannel.isOpen()) {
  83.                 serverSocketChannel.accept(null, this);
  84.             }
  85.             if (clientChannel != null && clientChannel.isOpen()) {
  86.                 ClientHandler handler = new ClientHandler(clientChannel);
  87.                 ByteBuffer buffer = ByteBuffer.allocate(BUFFER);
  88.                 //将用户添加到在线列表
  89.                 addClient(handler);
  90.                 clientChannel.read(buffer, buffer, handler);
  91.             }
  92.         }
  94.         @Override
  95.         public void failed(Throwable exc, Object attachment) {
  96.             System.out.println("用户连接失败" + exc);
  97.         }
  98.     }
  100.     private synchronized void addClient(ClientHandler handler) {
  101.         connectedClients.add(handler);
  102.         System.out.println(getClientName(handler.clientChannel) + "已成功连接到服务器");
  103.     }
  105.     private synchronized void removeClient(ClientHandler clientHandler) {
  106.         connectedClients.remove(clientHandler);
  107.         System.err.println(getClientName(clientHandler.clientChannel) + "已断开服务器连接");
  108.         shutdown(clientHandler.clientChannel);
  109.     }
  111.     private String getClientName(AsynchronousSocketChannel clientChannel) {
  112.         try {
  113.             InetSocketAddress address= (InetSocketAddress) clientChannel.getRemoteAddress();
  114.             String str="客户端["+address.getPort()+"]:";
  115.             return str;
  116.         } catch (IOException e) {
  117.             e.printStackTrace();
  118.         }
  119.         return "获取客户端失败";
  120.     }
  123.     private class ClientHandler implements CompletionHandler<Integer, ByteBuffer> {
  124.         private AsynchronousSocketChannel clientChannel;
  126.         public ClientHandler(AsynchronousSocketChannel clientChannel) {
  127.             this.clientChannel = clientChannel;
  128.         }
  130.         @Override
  131.         public void completed(Integer result, ByteBuffer attachment) {
  132.             if (attachment != null) {
  133.                 if (result <= 0) {
  134.                     //客户端异常
  135.                     //移除在线列表
  136.                     removeClient(this);
  137.                 } else {
  138.                     attachment.flip();
  139.                     String msg = receive(attachment);
  140.                     System.out.println(getClientName(clientChannel) + msg);
  141.                     forwardMessage(clientChannel, msg);
  142.                     attachment.clear();
  144.                     //检查用户是否决定退出
  145.                     if (readyToQuit(msg)) {
  146.                         removeClient(this);
  147.                     } else {
  148.                         clientChannel.read(attachment, attachment, this);
  149.                     }
  150.                 }
  151.             }
  152.         }
  154.         @Override
  155.         public void failed(Throwable exc, ByteBuffer attachment) {
  156.             System.out.println("用户读写失败" + exc);
  157.         }
  158.     }
  160.     private synchronized void forwardMessage(AsynchronousSocketChannel clientChannel, String msg) {
  162.         for (ClientHandler handler : connectedClients) {
  163.             try {
  164.                 if (!clientChannel.equals(handler.clientChannel)) {
  165.                     String str = getClientName(handler.clientChannel);
  166.                     ByteBuffer buffer = null;
  167.                     if (msg == null) {
  168.                         return;
  169.                     }
  170.                     if (readyToQuit(msg)) {
  171.                         buffer = charset.encode(str + "已断开连接");
  172.                         handler.clientChannel.write(buffer, null, handler);
  173.                     } else {
  174.                         buffer = charset.encode(str + msg);
  175.                         handler.clientChannel.write(buffer, null, handler);
  176.                     }
  177.                 }
  178.             } catch (Exception e) {
  179.                 e.printStackTrace();
  180.             }
  181.         }
  182.     }
  184.     private String receive(ByteBuffer attachment) {
  185.         return String.valueOf(charset.decode(attachment)).trim();
  186.     }
  188. }

ServerMain:

  1. package demo5.aio.server;
  3. /**
  4.  * @ClassName:
  5.  * @Description:
  6.  * @author: hszjj
  7.  * @date: 2019/11/23 19:43
  8.  */
  9. public class ServerMain {
  10.     public static void main(String[] args) {
  11.         ChatServer server=new ChatServer();
  12.         server.start();
  13.     }
  14. }

三、客户端创建

chatClient：

  1. package demo5.aio.client;
  3. import java.io.BufferedReader;
  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;
  13. public class ChatClient {
  14.     private static final String LOCALHOST = "localhost";
  15.     private static final int DEFAULT_PORT = 9999;
  16.     private static final String QUIT = "quit";
  17.     private static final int BUFFER = 1024;
  19.     private Charset charset = Charset.forName("UTF-8");
  20.     private int port;
  21.     private String host;
  22.     private AsynchronousSocketChannel clientChannel;
  24.     public ChatClient(String host, int port) {
  25.         this.host = host;
  26.         this.port = port;
  27.     }
  29.     public ChatClient() {
  30.         this(LOCALHOST, DEFAULT_PORT);
  31.     }
  33.     private void shutdown(Closeable... closeables) {
  34.         try {
  35.             for (Closeable shut : closeables) {
  36.                 if (shut != null) {
  37.                     shut.close();
  38.                 }
  39.             }
  40.         } catch (IOException e) {
  41.             e.printStackTrace();
  42.         }
  43.     }
  45.     public boolean readyToQuit(String str) {
  46.         return QUIT.equalsIgnoreCase(str);
  47.     }
  49.     public void start() {
  50.         try {
  51.             clientChannel = AsynchronousSocketChannel.open();
  52.             Future<Void> future = clientChannel.connect(new InetSocketAddress(host, port));
  53.             System.out.println("已成功连接到服务器");
  54.             future.get();
  56.             //处理用户输入
  57.             new Thread(new UserInputHander(this)).start();
  59.             ByteBuffer buffer = ByteBuffer.allocate(BUFFER);
  60.             while (true) {
  61.                 Future<Integer> readResult = clientChannel.read(buffer);
  62.                 int result = readResult.get();
  63.                 if (result <= 0) {
  64.                     //服务器异常
  65.                     System.err.println("服务器断开");
  66.                     shutdown(clientChannel);
  67.                     System.exit(-1);
  68.                 } else {
  69.                     buffer.flip();
  70.                     String msg = String.valueOf(charset.decode(buffer));
  71.                     buffer.clear();
  72.                     System.out.println(msg);
  73.                 }
  74.             }
  75.         } catch (IOException e) {
  76.             e.printStackTrace();
  77.         } catch (InterruptedException e) {
  78.             e.printStackTrace();
  79.         } catch (ExecutionException e) {
  80.             e.printStackTrace();
  81.         }
  82.     }
  84.     public void send(String msg) {
  85.         try {
  86.             if (msg.isEmpty()) {
  87.                 return;
  88.             } else {
  89.                 ByteBuffer buffer = charset.encode(msg);
  90.                 Future<Integer> future = clientChannel.write(buffer);
  91.                 future.get();
  92.             }
  93.         } catch (InterruptedException | ExecutionException e) {
  94.             e.printStackTrace();
  95.             System.err.println("消息发送失败");
  96.         }
  97.     }
  98. }

UserInputHander:

  1. package demo5.aio.client;
  3. import demo3.nio1.client.Demo3ChatClient;
  5. import java.io.BufferedReader;
  6. import java.io.IOException;
  7. import java.io.InputStreamReader;
  9. public class UserInputHander implements Runnable {
  10.     private ChatClient chatClient;
  11.     private BufferedReader reader;
  13.     public UserInputHander(ChatClient chatClient) {
  14.         this.chatClient = chatClient;
  15.     }
  17.     @Override
  18.     public void run() {
  19.         reader = new BufferedReader(new InputStreamReader(System.in));
  21.         try {
  22.             while (true) {
  23.                 String msg = reader.readLine();
  24.                 chatClient.send(msg);
  25.                 if (chatClient.readyToQuit(msg)) {
  26.                     break;
  27.                 }
  28.             }
  29.         } catch (IOException e) {
  30.             e.printStackTrace();
  31.         } finally {
  32.             if (reader != null) {
  33.                 try {
  34.                     System.out.println("已断开服务器连接");
  35.                     reader.close();
  36.                 } catch (IOException e) {
  37.                     e.printStackTrace();
  38.                 }
  39.             }
  40.         }
  41.     }
  42. }

ClientMain:

  1. package demo5.aio.client;
  3. /**
  4.  * @ClassName:
  5.  * @Description:
  6.  * @author: hszjj
  7.  * @date: 2019/11/23 20:16
  8.  */
  9. public class ClientMain {
  10.     public static void main(String[] args) {
  11.         ChatClient client=new ChatClient();
  12.         client.start();
  13.     }
  14. }

四、效果演示

image.png