一、NIO编程模型概述

ACCEPT事件：

服务器端启动ServerSocketChannel，绑定端口后注册ACCEPT事件，当客户端发送并被服务器接收的时候则触发ACCEPT事件，并通过handles处理事件、注册新的事件READ。
image.png
在客户发送信息以后READ事件就可以被触发。
注意：
处理事件都是在同一线程中完成的：
image.png

READ事件：

接上文，当用户SocketChannel中拥有数据时，Selector会发现拥有可读事件，READ被触发，通过handles把数据转发出去，同样，处理事件都是在同一线程中完成的。

注意：

虽然NIO的读写都是非阻塞性的但是Selector是阻塞性的，当没有事件发生的时候Selector的select会一直阻塞，直到新的事件触发。

第二个链接建立：

过程同上，形成如图：
image.png
注册了监听Client2的READ事件。

二、代码实践

服务器端

ChatServer

  1. package demo3.nio1.server;
  2. import java.io.Closeable;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.*;
  7. import java.nio.charset.Charset;
  8. import java.util.Set;
  10. public class ChatServer {
  12.     private static final String QUIT="quit";
  13.     private static final int DEFAULT_PORT=9999;
  14.     private static final int BUFFER=1024;
  16.     private int port;
  18.     private ServerSocketChannel server;
  19.     private Selector selector;
  21.     private ByteBuffer readBuffer=ByteBuffer.allocate(BUFFER);
  22.     private ByteBuffer writeBuffer=ByteBuffer.allocate(BUFFER);
  23.     //编码解码
  24.     private Charset charset=Charset.forName("UTF-8");
  26.     public ChatServer(int port){
  27.         this.port=port;
  28.     }
  29.     public ChatServer(){
  30.         this(DEFAULT_PORT);
  31.     }
  33.     private boolean readyToQuit(String str){
  34.         return QUIT.equalsIgnoreCase(str);
  35.     }
  37.     private void close(Closeable...closeables) {
  38.         try {
  39.             for (Closeable shut : closeables) {
  40.                 if (shut != null) {
  41.                     shut.close();
  42.                 }
  43.             }
  44.         } catch (IOException e) {
  45.             e.printStackTrace();
  46.         }
  47.     }
  49.     public void start(){
  50.         try {
  51.             server = ServerSocketChannel.open();
  52.             //关闭阻塞状态
  53.             server.configureBlocking(false);
  54.             //绑定监听端口
  55.             server.socket().bind(new InetSocketAddress(port));
  57.             selector = Selector.open();
  58.             //注册ACCEPT
  59.             server.register(selector, SelectionKey.OP_ACCEPT);
  60.             System.out.println("服务器[" + port + "]已启动");
  62.             while (true) {
  63.                 selector.select();
  64.                 Set<SelectionKey> selectionKeys = selector.selectedKeys();
  65.                 for (SelectionKey key : selectionKeys) {
  66.                     handles(key);
  67.                 }
  68.                 selectionKeys.clear();
  69.             }
  70.         }catch (ClosedSelectorException e){
  72.         } catch (IOException e) {
  73.             e.printStackTrace();
  74.         }finally {
  75.             //不必要关闭ServerChannel，因为关闭selector以后会把它对应的通道一起关闭
  76.             close(selector);
  77.         }
  78.     }
  80.     private void handles(SelectionKey key) throws IOException {
  81.         SocketChannel client = null;
  82.         // ACCEPT事件--和客户端建立连接
  83.         if (key.isAcceptable()) {
  84.             ServerSocketChannel channel = (ServerSocketChannel) key.channel();
  85.             client = channel.accept();
  86.             client.configureBlocking(false);
  87.             client.register(selector, SelectionKey.OP_READ);
  88.             System.out.println(getClientName(client) + "已连接到服务器");
  89.         }
  90.         // READ事件--客户端发送了消息
  91.         else if (key.isReadable()) {
  92.             client = (SocketChannel) key.channel();
  93.             String msg = recieve(client);
  94.             //客户端异常
  95.             if (msg.isEmpty()) {
  96.                 //取消继续监视这个通达
  97.                 key.cancel();
  98.                 //更新selector的状态
  99.                 selector.wakeup();
  100.             } else {
  101.                 //检查是否退出
  102.                 if (readyToQuit(msg)) {
  103.                     key.cancel();
  104.                     selector.wakeup();
  105.                     System.out.println(getClientName(client) + "断开连接");
  106.                 } else {
  107.                     System.out.println(getClientName(client)+msg);
  108.                     //转发数据
  109.                     forwardMessage(client, msg);
  110.                 }
  111.             }
  112.         }
  113.     }
  115.     private String getClientName(SocketChannel client){
  116.         return "客户端["+client.socket().getPort()+"]";
  117.     }
  119.     private void forwardMessage(SocketChannel client, String msg) throws IOException {
  120.         for (SelectionKey key:selector.keys()){
  121.             Channel connectedClient= key.channel();
  122.             if (connectedClient instanceof ServerSocketChannel){
  123.                 continue;
  124.             }
  125.             if (key.isValid()&&!client.equals(connectedClient)){
  126.                 writeBuffer.clear();
  127.                 writeBuffer.put(charset.encode(getClientName(client)+":"+msg+"\n"));
  128.                 writeBuffer.flip();
  129.                 while (writeBuffer.hasRemaining()){
  130.                     ((SocketChannel) connectedClient).write(writeBuffer);
  131.                 }
  132.             }
  133.         }
  134.     }
  136.     private String recieve(SocketChannel client) throws IOException {
  137.         readBuffer.clear();
  138.         while (client.read(readBuffer) > 0) {
  139.         }
  140.         readBuffer.flip();
  141.         return String.valueOf(charset.decode(readBuffer));
  142.     }
  143. }

Main

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

用户端

ChatClient

  1. package demo3.nio1.client;
  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.*;
  8. import java.nio.charset.Charset;
  9. import java.util.Set;
  11. public class ChatClient {
  12.     private static final String DEFAULT_IP = "localhost";
  13.     private static final int DEFAULT_PORT = 9999;
  14.     private static final String QUIT = "quit";
  15.     private static final int BUFFER = 1024;
  17.     private String host;
  18.     private int post;
  19.     private SocketChannel client;
  21.     private ByteBuffer readBuffer = ByteBuffer.allocate(BUFFER);
  22.     private ByteBuffer writeBuffer = ByteBuffer.allocate(BUFFER);
  23.     private Selector selector;
  24.     private Charset charset = Charset.forName("UTF-8");
  26.     public ChatClient(String host, int post) {
  27.         this.host = host;
  28.         this.post = post;
  29.     }
  31.     public ChatClient() {
  32.         this(DEFAULT_IP, DEFAULT_PORT);
  33.     }
  35.     public boolean readyToQuit(String str) {
  36.         return QUIT.equalsIgnoreCase(str);
  37.     }
  39.     private void close(Closeable... closeables) {
  40.         try {
  41.             for (Closeable shut : closeables) {
  42.                 if (shut != null) {
  43.                     shut.close();
  44.                 }
  45.             }
  46.         } catch (IOException e) {
  47.             e.printStackTrace();
  48.         }
  49.     }
  51.     public void start() {
  52.         try {
  53.             client = SocketChannel.open();
  54.             client.configureBlocking(false);
  56.             selector = Selector.open();
  57.             client.register(selector, SelectionKey.OP_CONNECT);
  58.             client.connect(new InetSocketAddress(host, post));
  59.             System.out.println("客户端[" + host + "]已连接到服务器");
  61.             while (true) {
  62.                 selector.select();
  63.                 Set<SelectionKey> selectionKeys = selector.selectedKeys();
  64.                 for (SelectionKey key : selectionKeys) {
  65.                     handles(key);
  66.                 }
  67.                 selectionKeys.clear();
  68.             }
  69.         }catch (ClosedSelectorException e){
  71.         } catch (IOException e) {
  72.             e.printStackTrace();
  73.         } finally {
  74.             close(selector);
  75.         }
  76.     }
  78.     private void handles(SelectionKey key) throws IOException {
  79.         SocketChannel channel = null;
  80.         //CONNECT事件--连接就绪
  81.         if (key.isConnectable()) {
  82.             channel = (SocketChannel) key.channel();
  83.             //判断是否就绪建立连接
  84.             if (channel.isConnectionPending()) {
  85.                 channel.finishConnect();
  86.             }
  87.             //处理用户的输入
  88.             new Thread(new UserInputHander(this)).start();
  89.             channel.register(selector, SelectionKey.OP_READ);
  90.         }
  91.         //READ事件--服务器转发消息
  92.         else if (key.isReadable()) {
  93.             channel = (SocketChannel) key.channel();
  94.             String msg = recieve(channel);
  95.             if (msg.isEmpty()) {
  96.                 //服务器异常
  97.                 System.out.println("客户端[" + host + "]已断开服务器");
  98.                 close(selector);
  99.             } else {
  100.                 System.out.println(msg);
  101.             }
  102.         }
  103.     }
  105.     private String recieve(SocketChannel channel) throws IOException {
  106.         readBuffer.clear();
  107.         while (client.read(readBuffer) > 0) {
  109.         }
  110.         readBuffer.flip();
  111.         return String.valueOf(charset.decode(readBuffer));
  112.     }
  114.     public void send(String msg) throws IOException {
  115.         if (msg.isEmpty()) {
  116.             return;
  117.         }
  118.         writeBuffer.clear();
  119.         writeBuffer.put(charset.encode(msg));
  120.         writeBuffer.flip();
  121.         while (writeBuffer.hasRemaining()) {
  122.             client.write(writeBuffer);
  123.         }
  124.         if (readyToQuit(msg)) {
  125.             close(selector);
  126.             System.out.println("客户端[" + host + "]已断开服务器");
  127.         }
  128.     }
  129. }

输入线程

  1. package demo3.nio1.client;
  3. import java.io.BufferedReader;
  4. import java.io.IOException;
  5. import java.io.InputStreamReader;
  7. public class UserInputHander implements Runnable {
  8.     private ChatClient chatClient;
  9.     private BufferedReader reader;
  11.     public UserInputHander(ChatClient chatClient) {
  12.         this.chatClient = chatClient;
  13.     }
  15.     @Override
  16.     public void run() {
  17.         reader = new BufferedReader(new InputStreamReader(System.in));
  19.         try {
  20.             while (true) {
  21.                 String msg = reader.readLine();
  22.                 chatClient.send(msg);
  23.                 if (chatClient.readyToQuit(msg)) {
  24.                     break;
  25.                 }
  26.             }
  27.         } catch (IOException e) {
  28.             e.printStackTrace();
  29.         } finally {
  30.             if (reader != null) {
  31.                 try {
  32.                     reader.close();
  33.                 } catch (IOException e) {
  34.                     e.printStackTrace();
  35.                 }
  36.             }
  37.         }
  38.     }
  39. }

Main

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

三、测试结果

image.png