image.png

Server端

  1. public class Server {
  2.     //定义属性
  3.     private Selector selector;
  4.     private ServerSocketChannel listenChannel;
  5.     private static final int PORT = 6667;
  7.     //构造器完成初始化工作
  8.     public Server() {
  9.         try {
  10.             //得到选择器
  11.             selector = Selector.open();
  12.             //serverSocketChannel
  13.             listenChannel = ServerSocketChannel.open();
  14.             //绑定端口
  15.             listenChannel.socket().bind(new InetSocketAddress(PORT));
  16.             //设置非阻塞
  17.             listenChannel.configureBlocking(false);
  18.             //将listenerChannel注册到selector
  19.             listenChannel.register(selector, SelectionKey.OP_ACCEPT);
  20.         } catch (Exception e) {
  21.             e.printStackTrace();
  22.         }
  23.     }
  25.     //监听
  26.     public void listen() {
  27.         try {
  28.             while (true) {
  29.                 int count = selector.select(2000);
  30.                 if (count > 0) { //表明有事件需要处理
  31.                     //遍历得到selectionKey集合
  32.                     Set<SelectionKey> selectionKeys = selector.selectedKeys();
  33.                     Iterator<SelectionKey> iterator = selectionKeys.iterator();
  34.                     while ((iterator.hasNext())) {
  35.                         //取出SelectionKey
  36.                         SelectionKey key = iterator.next();
  37.                         //监听到accept
  38.                         if (key.isAcceptable()) {
  39.                             SocketChannel socketChannel = listenChannel.accept();
  40.                             //设置非阻塞
  41.                             socketChannel.configureBlocking(false);
  42.                             //将该socketChannel注册到seletor上
  43.                             socketChannel.register(selector, SelectionKey.OP_READ);
  44.                             //提示上线
  45.                             System.out.println(socketChannel.getRemoteAddress() + "上线了");
  46.                         }
  47.                         if (key.isReadable()) {
  48.                             //读取数据
  49.                             readData(key);
  50.                         }
  51.                         //删除key，防止重复
  52.                         iterator.remove();
  53.                     }
  54.                 } else {
  55.                     //System.out.println("等待中");
  56.                 }
  57.             }
  58.         } catch (Exception e) {
  59.             e.printStackTrace();
  60.         } finally {
  62.         }
  63.     }
  65.     /**
  66.      * 读取消息
  67.      *
  68.      * @param key 传入SelectionKey
  69.      */
  70.     private void readData(SelectionKey key) {
  71.         //定义socketChannel
  72.         SocketChannel channel = null;
  73.         try {
  74.             channel = (SocketChannel) key.channel();
  75.             //创建Buffer
  76.             ByteBuffer buffer = ByteBuffer.allocate(1024);
  77.             int count = channel.read(buffer);
  78.             //根据count的值做处理
  79.             if (count > 0) {
  80.                 //读取到了数据转化成字符串
  81.                 String msg = new String(buffer.array());
  82.                 //输出消息
  83.                 System.out.println("form 客户端" + msg);
  84.                 //向其他的客户端转发消息
  85.                 sendInfoToaOtherClients(msg, channel);
  86.             }
  87.         } catch (Exception e) {
  88.             try {
  89.                 System.out.println(channel.getRemoteAddress() + "离线了");
  90.                 //取消注册
  91.                 key.cancel();
  92.                 //关闭通道
  93.                 channel.close();
  94.             } catch (IOException ioException) {
  95.                 ioException.printStackTrace();
  96.             }
  97.         }
  98.     }
  100.     /**
  101.      * 转发消息给其他用户
  102.      *
  103.      * @param msg  消息
  104.      * @param self 排除自己--传入自己的channel
  105.      */
  106.     private void sendInfoToaOtherClients(String msg, SocketChannel self) throws IOException {
  107.         System.out.println("服务器转发消息");
  108.         //遍历所有注册到selector中的socketChannel并排除自己
  109.         for (SelectionKey key : selector.keys()) {
  110.             //获取到key对应的channel
  111.             Channel channel = key.channel();
  112.             //排除自己
  113.             if (channel instanceof SocketChannel && channel != self) {
  114.                 SocketChannel dest = (SocketChannel) channel;
  115.                 //将消息存储到buffer
  116.                 ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
  117.                 //将buffer的数据写入通道
  118.                 dest.write(buffer);
  119.             }
  120.         }
  121.     }
  123.     public static void main(String[] args) {
  124.         Server server = new Server();
  125.         //监听
  126.         server.listen();
  128.     }
  129. }

Client端

  1. public class Client1 {
  2.     private final static String HOST = "127.0.0.1"; //ip
  3.     private final static int PORT = 6667; //端口
  4.     private final Selector selector;
  5.     private final SocketChannel socketChannel;
  6.     private final String username;
  8.     public Client1() throws Exception {
  9.         selector = Selector.open();
  10.         //连接
  11.         socketChannel = SocketChannel.open(new InetSocketAddress(HOST, PORT));
  12.         //设置非阻塞
  13.         socketChannel.configureBlocking(false);
  14.         //将channel注册到selector
  15.         socketChannel.register(selector, SelectionKey.OP_READ);
  16.         //得到username
  17.         username = socketChannel.getLocalAddress().toString().substring(1);
  18.         System.out.println(username + "is ok");
  20.     }
  22.     /**
  23.      * 向服务器端发送消息
  24.      *
  25.      * @param info 发送的消息
  26.      */
  27.     public void sendInfo(String info) {
  28.         info = username + "说：" + info;
  29.         try {
  30.             socketChannel.write(ByteBuffer.wrap(info.getBytes(StandardCharsets.UTF_8)));
  31.         } catch (Exception e) {
  32.             e.printStackTrace();
  33.         }
  34.     }
  36.     public void readInfo() {
  37.         try {
  38.             int ReadSelects = selector.select();
  39.             if (ReadSelects > 0) {//说明有事件发生的通道
  40.                 Set<SelectionKey> selectionKeys = selector.selectedKeys();
  41.                 Iterator<SelectionKey> iterator = selectionKeys.iterator();
  42.                 while (iterator.hasNext()) {
  43.                     SelectionKey key = iterator.next();
  44.                     if (key.isReadable()) {
  45.                         //得到相关的通道
  46.                         SocketChannel channel = (SocketChannel) key.channel();
  47.                         //得到一个buffer
  48.                         ByteBuffer buffer = ByteBuffer.allocate(1024);
  49.                         channel.read(buffer);
  50.                         //输出读到缓冲区的内容
  51.                         String msg = new String(buffer.array());
  52.                         System.out.println(msg.trim());
  53.                     }
  54.                 }
  55.                 iterator.remove(); //移除当前key
  56.             }
  57.         } catch (Exception e) {
  58.             e.printStackTrace();
  59.         }
  60.     }
  62.     public static void main(String[] args) throws Exception {
  63.         //启动客户端
  64.         Client1 client1 = new Client1();
  65.         //启动线程，每隔3s，读取从服务器端发送数据
  66.         new Thread(() -> {
  67.             while (true) {
  68.                 client1.readInfo();
  69.                 try {
  70.                     Thread.sleep(1000);
  71.                 } catch (Exception e) {
  72.                     e.printStackTrace();
  73.                 }
  74.             }
  75.         }).start();
  77.         //发送数据给服务器
  78.         Scanner scanner = new Scanner(System.in);
  79.         while (scanner.hasNextLine()) {
  80.             String s = scanner.nextLine();
  81.             client1.sendInfo(s);
  82.         }
  83.     }
  84. }