通过选择器 Selector 开发高性能聊天室

转x33g5p2x  于2022-05-23 转载在 其他  
字(7.2k)|赞(0)|评价(0)|浏览(276)

一 点睛

选择器（Selector）的核心作用：可以在一个选择器上注册多个通道，并且可以通过选择器切换使用这些通道。

如果不使用选择器，服务端的 I/O 代码，会给每个客户端创建一个新线程，也就是用 N 个线程去处理 N 个客户端请求。因此，如果有 1 万个客户请求，就会创建 1 万个线程，这显然是不合理地。而 NIO 处理这种问题的思路是，用一个线程处理全部请求，并通过 Selector 切换处理不同的请求通道。

二 实战

1 需求

使用 NIO 实现一个聊天室功能。要求如下：服务端启动后可以接收多个客户端连接，每个客户端都可以向服务端发送消息；服务端接受到消息后，会在控制台打印此客户端的信息，并且将此消息发给全部的客户端。

2 服务端

服务端只创建一个处理请求的线程

  1. package nio;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.net.ServerSocket;
  6. import java.nio.ByteBuffer;
  7. import java.nio.channels.SelectionKey;
  8. import java.nio.channels.Selector;
  9. import java.nio.channels.ServerSocketChannel;
  10. import java.nio.channels.SocketChannel;
  11. import java.nio.charset.Charset;
  12. import java.util.HashMap;
  13. import java.util.Iterator;
  14. import java.util.Map;
  15. import java.util.Set;
  17. /**
  18.  * @className: ChatServer
  19.  * @description: 服务器
  20.  * @date: 2022/5/23
  21.  * @author: cakin
  22.  */
  23. public class ChatServer {
  24.     /*
  25.         clientsMap:保存所有的客户端
  26.         key：客户端的名字
  27.         value:客户端连接服务端的Channel
  28.      */
  29.     private static Map<String, SocketChannel> clientsMap = new HashMap();
  31.     public static void main(String[] args) throws IOException {
  32.         int[] ports = new int[]{7777, 8888, 9999};
  33.         Selector selector = Selector.open();
  35.         for (int port : ports) {
  36.             ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
  37.             serverSocketChannel.configureBlocking(false);
  38.             ServerSocket serverSocket = serverSocketChannel.socket();
  39.             // 将聊天服务绑定到 7777、8888和9999 三个端口上
  40.             serverSocket.bind(new InetSocketAddress(port));
  41.             System.out.println("服务端启动成功，端口" + port);
  42.             // 在服务端的选择器上，注册一个通道，并标识该通道所感兴趣的事件是：接收客户端连接（接收就绪）
  43.             serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
  45.         }
  47.         while (true) {
  48.             // 一直阻塞，直到选择器上存在已经就绪的通道（包含感兴趣的事件）
  49.             selector.select();
  50.             // selectionKeys包含了所有通道与选择器之间的关系（接收连接、读、写）
  51.             Set<SelectionKey> selectionKeys = selector.selectedKeys();
  52.             Iterator<SelectionKey> keyIterator = selectionKeys.iterator();
  53.             // 如果selector中有多个就绪通道（接收就绪、读就绪、写就绪等），则遍历这些通道
  54.             while (keyIterator.hasNext()) {
  55.                 SelectionKey selectedKey = keyIterator.next();
  56.                 String receive = null;
  57.                 // 与客户端交互的通道
  58.                 SocketChannel clientChannel;
  59.                 try {
  60.                     // 接收就绪（已经可以接收客户端的连接了）
  61.                     if (selectedKey.isAcceptable()) {
  62.                         ServerSocketChannel server = (ServerSocketChannel) selectedKey.channel();
  63.                         clientChannel = server.accept();
  64.                         // 切换到非阻塞模式
  65.                         clientChannel.configureBlocking(false);
  66.                         // 再在服务端的选择器上，注册第二个通道，并标识该通道所感兴趣的事件是：接收客户端发来的消息（读就绪）
  67.                         clientChannel.register(selector, SelectionKey.OP_READ);
  68.                         // 用“key四位随机数”的形式模拟客户端的key值
  69.                         String key = "key" + (int) (Math.random() * 9000 + 1000);
  70.                         // 将该建立完毕连接的 通道 保存到clientsMap中
  71.                         clientsMap.put(key, clientChannel);
  72.                         // 读就绪（已经可以读取客户端发来的信息了）
  73.                     } else if (selectedKey.isReadable()) {
  74.                         clientChannel = (SocketChannel) selectedKey.channel();
  75.                         ByteBuffer readBuffer = ByteBuffer.allocate(1024);
  76.                         int result = -1;
  77.                         try {
  78.                             // 将服务端读取到的客户端消息，放入 readBuffer 中
  79.                             result = clientChannel.read(readBuffer);
  80.                             // 如果终止客户端，则 read() 会抛出 IOException 异常，可以依次判断是否有客户端退出。
  81.                         } catch (IOException e) {
  82.                             // 获取退出连接的 client 对应的 key
  83.                             String clientKey = getClientKey(clientChannel);
  84.                             System.out.println("客户端" + clientKey + "退出聊天室");
  85.                             clientsMap.remove(clientKey);
  86.                             clientChannel.close();
  87.                             selectedKey.cancel();
  88.                             continue;
  89.                         }
  90.                         if (result > 0) {
  91.                             readBuffer.flip();
  92.                             Charset charset = Charset.forName("utf-8");
  93.                             receive = String.valueOf(charset.decode(readBuffer).array());
  94.                             // 将读取到的客户端消息，打印在服务端的控制台
  95.                             System.out.println(clientChannel + ":" + receive);
  96.                             // 处理客户端第一次发来的连接测试信息
  97.                             if ("connecting".equals(receive)) {
  98.                                 receive = "新客户端加入聊天!";
  99.                             }
  100.                             // 将读取到的客户消息保存在 attachment 中，用于后续向所有客户端转发此消息
  101.                             selectedKey.attach(receive);
  102.                             // 将通道所感兴趣的事件标识为：向客户端发送消息（写就绪）
  103.                             selectedKey.interestOps(SelectionKey.OP_WRITE);
  104.                         }
  105.                         // 写就绪
  106.                     } else if (selectedKey.isWritable()) {
  107.                         clientChannel = (SocketChannel) selectedKey.channel();
  108.                         // 获取发送消息从client对应的key
  109.                         String sendKey = getClientKey(clientChannel);
  110.                         // 将接收到的消息，拼接成“发送消息的客户端Key:消息”的形式，再广播给所有client
  111.                         for (Map.Entry<String, SocketChannel> entry : clientsMap.entrySet()) {
  112.                             SocketChannel eachClient = entry.getValue();
  113.                             ByteBuffer broadcastMsg = ByteBuffer.allocate(1024);
  114.                             broadcastMsg.put((sendKey + ":" + selectedKey.attachment()).getBytes());
  115.                             broadcastMsg.flip();
  116.                             eachClient.write(broadcastMsg);
  117.                         }
  118.                         selectedKey.interestOps(SelectionKey.OP_READ);
  119.                     }
  120.                 } catch (Exception e) {
  121.                     e.printStackTrace();
  122.                 }
  123.             }
  124.             selectionKeys.clear();
  125.         }
  126.     }
  128.     public static String getClientKey(SocketChannel clientChannel) {
  129.         String sendKey = null;
  130.         //很多client在发下消息，通过for找到是哪个client在发消息，找到该client的key
  131.         for (Map.Entry<String, SocketChannel> entry : clientsMap.entrySet()) {
  132.             if (clientChannel == entry.getValue()) {
  133.                 //找到发送消息的client所对应的key
  134.                 sendKey = entry.getKey();
  135.                 break;
  136.             }
  137.         }
  138.         return sendKey;
  139.     }
  140. }

3 客户端

  1. package nio;
  3. import java.io.BufferedReader;
  4. import java.io.IOException;
  5. import java.io.InputStreamReader;
  6. import java.net.InetSocketAddress;
  7. import java.nio.ByteBuffer;
  8. import java.nio.channels.SelectionKey;
  9. import java.nio.channels.Selector;
  10. import java.nio.channels.SocketChannel;
  11. import java.util.Iterator;
  12. import java.util.Set;
  14. /**
  15.  * @className: ChatClient
  16.  * @description: 客户端
  17.  * @date: 2022/5/23
  18.  * @author: cakin
  19.  */
  20. public class ChatClient {
  21.     public static void main(String[] args) {
  22.         try {
  23.             SocketChannel socketChannel = SocketChannel.open();
  24.             // 切换到非阻塞模式
  25.             socketChannel.configureBlocking(false);
  26.             Selector selector = Selector.open();
  27.             // 在客户端的选择器上，注册一个通道，并标识该通道所感兴趣的事件是：向服务端发送连接（连接就绪）。对应于服务端的OP_ACCEPT事件
  28.             socketChannel.register(selector, SelectionKey.OP_CONNECT);
  29.             // 随机连接到服务端提供的一个端口上
  30.             int[] ports = {7777, 8888, 9999};
  31.             int port = ports[(int) (Math.random() * 3)];
  32.             socketChannel.connect(new InetSocketAddress("127.0.0.1", port));
  33.             while (true) {
  34.                 selector.select();
  35.                 // selectionKeys 包含了所有通道与选择器之间的关系（请求连接、读、写）
  36.                 Set<SelectionKey> selectionKeys = selector.selectedKeys();
  37.                 Iterator<SelectionKey> keyIterator = selectionKeys.iterator();
  38.                 while (keyIterator.hasNext()) {
  39.                     SelectionKey selectedKey = keyIterator.next();
  40.                     // 判断是否连接成功
  41.                     if (selectedKey.isConnectable()) {
  42.                         ByteBuffer sendBuffer = ByteBuffer.allocate(1024);
  43.                         // 创建一个用于和服务端交互的 Channel
  44.                         SocketChannel client = (SocketChannel) selectedKey.channel();
  45.                         // 如果状态是：正在连接中...
  46.                         if (client.isConnectionPending()) {
  47.                             boolean isConnected = client.finishConnect();
  48.                             if (isConnected) {
  49.                                 System.out.println("连接成功！访问的端口是：" + port);
  50.                                 // 向服务端发送一条测试消息
  51.                                 sendBuffer.put("connecting".getBytes());
  52.                                 sendBuffer.flip();
  53.                                 client.write(sendBuffer);
  54.                             }
  56.                             // 在“聊天室”中，对于客户端而言，可以随时向服务端发送消息（写操作），因此，需要建立一个单独写线程
  57.                             new Thread(() -> {
  58.                                 while (true) {
  59.                                     try {
  60.                                         sendBuffer.clear();
  61.                                         // 接收用户从控制台输入的内容，并发送给服务端
  62.                                         InputStreamReader reader = new InputStreamReader(System.in);
  63.                                         BufferedReader bReader = new BufferedReader(reader);
  64.                                         String message = bReader.readLine();
  66.                                         sendBuffer.put(message.getBytes());
  67.                                         sendBuffer.flip();
  68.                                         client.write(sendBuffer);
  69.                                     } catch (Exception e) {
  70.                                         e.printStackTrace();
  71.                                     }
  72.                                 }
  73.                             }).start();
  74.                         }
  75.                         // 标记通道感兴趣的事件是：读取服务端消息（读就绪）
  76.                         client.register(selector, SelectionKey.OP_READ);
  77.                         // 客户端读取服务端的反馈消息
  78.                     } else if (selectedKey.isReadable()) {
  79.                         SocketChannel client = (SocketChannel) selectedKey.channel();
  80.                         ByteBuffer readBuffer = ByteBuffer.allocate(1024);
  81.                         // 将服务端的反馈消息放入 readBuffer中
  82.                         int len = client.read(readBuffer);
  83.                         if (len > 0) {
  84.                             String receive = new String(readBuffer.array(), 0, len);
  85.                             System.out.println(receive);
  86.                         }
  87.                     }
  88.                 }
  89.                 selectionKeys.clear();
  90.             }
  91.         } catch (IOException e) {
  92.             e.printStackTrace();
  93.         }
  94.     }
  95. }

三 测试

依次启动服务端和2个客户端，并发送消息

1 服务端打印

服务端启动成功，端口7777

服务端启动成功，端口8888

服务端启动成功，端口9999

java.nio.channels.SocketChannel[connected local=/127.0.0.1:7777 remote=/127.0.0.1:52781]:connecting

java.nio.channels.SocketChannel[connected local=/127.0.0.1:7777 remote=/127.0.0.1:52892]:connecting

java.nio.channels.SocketChannel[connected local=/127.0.0.1:7777 remote=/127.0.0.1:52892]:你好    

java.nio.channels.SocketChannel[connected local=/127.0.0.1:7777 remote=/127.0.0.1:52781]:你来自哪里？            

2 第1个客户端打印

连接成功！访问的端口是：7777

key9664:新客户端加入聊天!

key4978:新客户端加入聊天!

key4978:你好    

你来自哪里？

key9664:你来自哪里？            

3 第2个客户端打印

连接成功！访问的端口是：7777

key4978:新客户端加入聊天!

你好

key4978:你好    

key9664:你来自哪里？            

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/chengqiuming/article/details/124932371

内容来源于网络，如有侵权，请联系作者删除！

Java

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com