发布：2023/4/7 17:33:57作者：管理员 来源：本站 浏览次数：539

文章目录

   多人在线，多人聊天（可能有TCP粘包bug）

   多人在线，多人聊天（简单解决了TCP粘包bug）

   多人在线，单人聊天

   参考博客

多人在线，多人聊天（可能有TCP粘包bug）

服务端：

package NonBlocking;

import java.io.IOException;

import java.net.InetSocketAddress;

import java.nio.ByteBuffer;

import java.nio.channels.*;

import java.nio.charset.Charset;

import java.util.HashMap;

import java.util.Iterator;

import java.util.Map;

public class ChatServer {

   private final int port = 8899;

   private final String seperator = "[|]";                        //消息分隔符

   private final Charset charset = Charset.forName("UTF-8");    //字符集

   private ByteBuffer buffer = ByteBuffer.allocate(1024);        //缓存

   private Map<String, SocketChannel> onlineUsers = new HashMap<String, SocketChannel>();//将用户对应的channel对应起来

   private Selector selector;

   private ServerSocketChannel server;

   public void startServer() throws IOException {

       //NIO server初始化固定流程：5步

       selector = Selector.open();                    //1.selector open

       server = ServerSocketChannel.open();        //2.ServerSocketChannel open

       server.bind(new InetSocketAddress(port));    //3.serverChannel绑定端口

       server.configureBlocking(false);            //4.设置NIO为非阻塞模式

       server.register(selector, SelectionKey.OP_ACCEPT);//5.将channel注册在选择器上

       //NIO server处理数据固定流程:5步

       SocketChannel client;

       SelectionKey key;

       Iterator<SelectionKey> iKeys;

       while (true) {

           selector.select();                            //1.用select()方法阻塞，一直到有可用连接加入

           iKeys = selector.selectedKeys().iterator();    //2.到了这步，说明有可用连接到底，取出所有可用连接

           while (iKeys.hasNext()) {

               key = iKeys.next();                        //3.遍历

               if (key.isAcceptable()) {                    //4.对每个连接感兴趣的事做不同的处理

                   //对于客户端连接，注册到服务端

                   client = server.accept();            //获取客户端首次连接

                   client.configureBlocking(false);

                   //不用注册写，只有当写入量大，或写需要争用时，才考虑注册写事件

                   client.register(selector, SelectionKey.OP_READ);

                   System.out.println("+++++客户端：" + client.getRemoteAddress() + "，建立连接+++++");

                   client.write(charset.encode("请输入自定义用户名："));

               }

               if (key.isReadable()) {

                   client = (SocketChannel) key.channel();//通过key取得客户端channel

                   StringBuilder msg = new StringBuilder();

                   buffer.clear();        //多次使用的缓存，用前要先清空

                   try {

                       System.out.println(buffer);

                       while (client.read(buffer) > 0) {

                           buffer.flip();    //将写模式转换为读模式

                           msg.append(charset.decode(buffer));

                           buffer.clear();

                       }

                   } catch (IOException e) {

                       //如果client.read(buffer)抛出异常，说明此客户端主动断开连接，需做下面处理

                       client.close();            //关闭channel

                       key.cancel();            //将channel对应的key置为不可用

                       onlineUsers.values().remove(client);    //将问题连接从map中删除

                       System.out.println("-----用户'" + key.attachment().toString() + "'退出连接，当前用户列表：" + onlineUsers.keySet().toString() + "-----");

                       continue;                //跳出循环

                   }

                   if (msg.length() > 0) this.processMsg(msg.toString(), client, key);    //处理消息体

               }

               iKeys.remove();                    //5.处理完一次事件后，要显式的移除

           }

       }

   }

   /**

    * 处理客户端传来的消息

    *

    * @param msg 格式：user_to|body|user_from

    * @throws IOException

    * @Key 这里主要用attach()方法，给通道定义一个表示符

    */

   private void processMsg(String msg, SocketChannel client, SelectionKey key) throws IOException {

       String[] ms = msg.split(seperator);

       if (ms.length == 1) {

           String user = ms[0];    //输入的是自定义用户名

           if (onlineUsers.containsKey(user)) {

               client.write(charset.encode("当前用户已存在，请重新输入用户名："));

           } else {

               onlineUsers.put(user, client);

               key.attach(user);    //给通道定义一个表示符

               String welCome = "\t欢迎'" + user + "'上线，当前在线人数" + this.getOnLineNum() + "人。用户列表：" + onlineUsers.keySet().toString();

               client.write(charset.encode("您的昵称通过验证 "+user));

               this.broadCast(welCome);    //给所用用户推送上线信息，包括自己

           }

       } else if (ms.length == 2) {

           String msg_body = ms[0];

           String user_from = ms[1];

           broadCast("来自'" + user_from + "'的消息：" + msg_body);

       }

   }

   //map中的有效数量已被很好的控制，可以从map中获取，也可以用下面的方法取

   private int getOnLineNum() {

       int count = 0;

       Channel channel;

       for (SelectionKey k : selector.keys()) {

           channel = k.channel();

           if (channel instanceof SocketChannel) {    //排除ServerSocketChannel

               count++;

           }

       }

       return count;

   }

   //广播上线消息

   private void broadCast(String msg) throws IOException {

       Channel channel;

       for (SelectionKey k : selector.keys()) {

           channel = k.channel();

           if (channel instanceof SocketChannel) {

               SocketChannel client = (SocketChannel) channel;

               client.write(charset.encode(msg));

           }

       }

   }

   public static void main(String[] args) {

       try {

           new ChatServer().startServer();

       } catch (IOException e) {

           e.printStackTrace();

       }

   }

}

客户端：

package NonBlocking;

import java.io.IOException;

import java.net.InetSocketAddress;

import java.nio.ByteBuffer;

import java.nio.channels.*;

import java.nio.charset.Charset;

import java.util.Iterator;

import java.util.Scanner;

public class ChatClient1 {

   private final int port = 8899;

   private final String seperator = "|";

   private final Charset charset = Charset.forName("UTF-8"); //字符集

   private ByteBuffer buffer = ByteBuffer.allocate(1024);

   private SocketChannel _self;

   private Selector selector;

   private String name = "";

   private boolean flag = true; //服务端断开，客户端的读事件不会一直发生（与服务端不一样）

   Scanner scanner = new Scanner(System.in);

   public void startClient() throws IOException{

       //客户端初始化固定流程：4步

       selector = Selector.open(); //1.打开Selector

       _self = SocketChannel.open(new InetSocketAddress(port));//2.连接服务端，这里默认本机的IP

       _self.configureBlocking(false); //3.配置此channel非阻塞

       _self.register(selector, SelectionKey.OP_READ); //4.将channel的读事件注册到选择器

       /*

        * 因为等待用户输入会导致主线程阻塞

        * 所以用主线程处理输入，新开一个线程处理读数据

        */

       new Thread(new ClientReadThread()).start(); //开一个异步线程处理读

       String input = "";

       while(flag){

           input = scanner.nextLine();

           if("".equals(input)){  

               System.out.println("不允许输入空串！");

               continue;

           }else if("".equals(name)){  //姓名如果没有初始化

//啥也不干，之后发给服务端验证姓名

           }else if(!"".equals(name)) {  //如果姓名已经初始化，那么说明现在的字符串就是想说的话

               input = input + seperator + name;

           }

           try{

               _self.write(charset.encode(input));

           }catch(Exception e){

               System.out.println(e.getMessage()+"客户端主线程退出连接！！");

           }

       }

   }

   private class ClientReadThread implements Runnable{

       @Override

       public void run(){

           Iterator<SelectionKey> ikeys;

           SelectionKey key;

           SocketChannel client;

           try {

               while(flag){

                   selector.select(); //调用此方法一直阻塞，直到有channel可用

                   ikeys = selector.selectedKeys().iterator();

                   while(ikeys.hasNext()){

                       key = ikeys.next();

                       if(key.isReadable()){ //处理读事件

                           client = (SocketChannel) key.channel();

                           //这里的输出是true，从selector的key中获取的客户端channel，是同一个

// System.out.println("client == _self:"+ (client == _self));

                           buffer.clear();

                           StringBuilder msg = new StringBuilder();

                           try{

                               while(client.read(buffer) > 0){

                                   buffer.flip(); //将写模式转换为读模式

                                   msg.append(charset.decode(buffer));

                               }

                           }catch(IOException en){

                               System.out.println(en.getMessage()+",客户端'"+key.attachment().toString()+"'读线程退出！！");

                               stopMainThread();

                           }

                           if (msg.toString().contains("您的昵称通过验证")) {

                               String[] returnStr = msg.toString().split(" ");

                               name = returnStr[1];

                               key.attach(name);

                           }

                           System.out.println(msg.toString());

                       }

                       ikeys.remove();

                   }

               }

           } catch (Exception e) {

               e.printStackTrace();

           }

       }

   }

   private void stopMainThread(){

       flag = false;

   }

   public static void main(String[] args){

       try {

           new ChatClient1().startClient();

       } catch (IOException e) {

           e.printStackTrace();

       }

   }

}

使用方法：

   仿照着 ChatClient1，再弄一个 ChatClient2。先运行服务端，再运行客户端 1、客户端 2。

   客户端里，先输入自己的昵称。

   之后，输入想说的话。

效果是这样的：

客户端1：

请输入自定义用户名：

jojo

您的昵称通过验证 jojo

欢迎'jojo'上线，当前在线人数1人。用户列表：[jojo]

阿姨压一压

来自'jojo'的消息：阿姨压一压

我真是High到不行啊

来自'jojo'的消息：我真是High到不行啊

欢迎'dio'上线，当前在线人数2人。用户列表：[jojo, dio]

来自'dio'的消息：哈哈哈哈

来自'dio'的消息：我不做人啦

客户端2：

请输入自定义用户名：

jojo

当前用户已存在，请重新输入用户名：

dio

您的昵称通过验证 dio

欢迎'dio'上线，当前在线人数2人。用户列表：[jojo, dio]

哈哈哈哈

来自'dio'的消息：哈哈哈哈

我不做人啦

来自'dio'的消息：我不做人啦

而可能的bug是tcp粘包，简单的说，是上一次发送的尾，和下一次发送的头，挨在一起了：

//服务端代码

               client.write(charset.encode("您的昵称通过验证 "+user));

               this.broadCast(welCome);    //给所用用户推送上线信息，包括自己

这两行代码由于前后发送，所以client这个channel会先后收到 昵称验证通过 和 推送上线 的消息，如果tcp在传输过程中足够快，那么客户端在一次read事件中，会把两次消息一次性读出来。而由于这个代码写的比较简陋，在tcp的传输内容上并没有建立起足够安全的内容协议（比如消息与消息用特定的分隔符、或者用前面的几个字节来标注内容的实际字节数），所以上述代码并不能两次消息分开了。

造成的问题是：客户端是通过服务端回复的 验证消息 来初始化姓名的，如果粘包情况出现，那么客户端将会把真正的姓名和上线消息合起来作为自己的姓名。

客户端1：

请输入自定义用户名：

dio

您的昵称通过验证 dio 欢迎'dio'上线，当前在线人数 1 人。用户列表：[dio]

dkajgk

来自'dio 欢迎'dio'上线，当前在线人数 1 人。用户列表：[dio]'的消息：dkajgk

哈哈哈哈

来自'dio 欢迎'dio'上线，当前在线人数 1 人。用户列表：[dio]'的消息：哈哈哈哈

  欢迎'jojo'上线，当前在线人数 2 人。用户列表：[dio, jojo]

来自'jojo'的消息：哈哈哈哈

来自'jojo'的消息：我不做人啦

客户端2：

请输入自定义用户名：

jojo

您的昵称通过验证 jojo

  欢迎'jojo'上线，当前在线人数 2 人。用户列表：[dio, jojo]

哈哈哈哈

来自'jojo'的消息：哈哈哈哈

我不做人啦

来自'jojo'的消息：我不做人啦

多人在线，多人聊天（简单解决了TCP粘包bug）

服务端主要修改了两行代码，令服务端的每个message的结尾都加上一个分割符，好让客户端即使在tcp粘包的情况下，也能分辨出两个message来：

               client.write(charset.encode("您的昵称通过验证 "+user+"|"));

               this.broadCast(welCome+"|");    //给所用用户推送上线信息，包括自己

客户端：

   private class ClientReadThread implements Runnable{

       @Override

       public void run(){

           Iterator<SelectionKey> ikeys;

           SelectionKey key;

           SocketChannel client;

           try {

               while(flag){

                   selector.select(); //调用此方法一直阻塞，直到有channel可用

                   ikeys = selector.selectedKeys().iterator();

                   while(ikeys.hasNext()){

                       key = ikeys.next();

                       if(key.isReadable()){ //处理读事件

                           client = (SocketChannel) key.channel();

                           //这里的输出是true，从selector的key中获取的客户端channel，是同一个

// System.out.println("client == _self:"+ (client == _self));

                           buffer.clear();

                           StringBuilder msg = new StringBuilder();

                           try{

                               while(client.read(buffer) > 0){

                                   buffer.flip(); //将写模式转换为读模式

                                   msg.append(charset.decode(buffer));

                               }

                           }catch(IOException en){

                               System.out.println(en.getMessage()+",客户端'"+key.attachment().toString()+"'读线程退出！！");

                               stopMainThread();

                           }

                           //这里将读取到消息用分隔符分离

                           String[] StrArray = msg.toString().split("[|]");

                           for (String message : StrArray) {

                               if (message == "") continue;

                               if (message.contains("您的昵称通过验证")) {

                                   if (message.contains("您的昵称通过验证")) {

                                       String[] nameValid = message.split(" ");

                                       name = nameValid[1];

                                       key.attach(name);

                                   }

                               }

                               System.out.println(message);

                           }

                       }

                       ikeys.remove();

                   }

               }

           } catch (Exception e) {

               e.printStackTrace();

           }

       }

   }

使用方法：

   仿照着ChatClient1，再弄一个ChatClient2。先运行服务端，再运行客户端1、客户端2。

   客户端里，先输入自己的昵称。

   之后，输入想说的话。

不过这样改还是没有解决，如果发送的数据过多，而使得另一方的read事件发生了两次，的问题。如果发生，那么bug是 一条聊天消息被拆分成两条聊天消息。

多人在线，单人聊天

服务端：

package NonBlocking;

import java.io.IOException;

import java.net.InetSocketAddress;

import java.nio.ByteBuffer;

import java.nio.channels.*;

import java.nio.charset.Charset;

import java.util.HashMap;

import java.util.Iterator;

import java.util.Map;

public class ChatServer {

   private final int port = 8899;

   private final String seperator = "[|]";                        //消息分隔符

   private final Charset charset = Charset.forName("UTF-8");    //字符集

   private ByteBuffer buffer = ByteBuffer.allocate(1024);        //缓存

   private Map<String, SocketChannel> onlineUsers = new HashMap<String, SocketChannel>();//将用户对应的channel对应起来

   private Selector selector;

   private ServerSocketChannel server;

   public void startServer() throws IOException {

       //NIO server初始化固定流程：5步

       selector = Selector.open();                    //1.selector open

       server = ServerSocketChannel.open();        //2.ServerSocketChannel open

       server.bind(new InetSocketAddress(port));    //3.serverChannel绑定端口

       server.configureBlocking(false);            //4.设置NIO为非阻塞模式

       server.register(selector, SelectionKey.OP_ACCEPT);//5.将channel注册在选择器上

       //NIO server处理数据固定流程:5步

       SocketChannel client;

       SelectionKey key;

       Iterator<SelectionKey> iKeys;

       while (true) {

           selector.select();                            //1.用select()方法阻塞，一直到有可用连接加入

           iKeys = selector.selectedKeys().iterator();    //2.到了这步，说明有可用连接到底，取出所有可用连接

           while (iKeys.hasNext()) {

               key = iKeys.next();                        //3.遍历

               if (key.isAcceptable()) {                    //4.对每个连接感兴趣的事做不同的处理

                   //对于客户端连接，注册到服务端

                   client = server.accept();            //获取客户端首次连接

                   client.configureBlocking(false);

                   //不用注册写，只有当写入量大，或写需要争用时，才考虑注册写事件

                   client.register(selector, SelectionKey.OP_READ);

                   System.out.println("+++++客户端：" + client.getRemoteAddress() + "，建立连接+++++");

                   client.write(charset.encode("请输入自定义用户名："));

               }

               if (key.isReadable()) {

                   client = (SocketChannel) key.channel();//通过key取得客户端channel

                   StringBuilder msg = new StringBuilder();

                   buffer.clear();        //多次使用的缓存，用前要先清空

                   try {

                       while (client.read(buffer) > 0) {

                           buffer.flip();    //将写模式转换为读模式

                           msg.append(charset.decode(buffer));

                       }

                   } catch (IOException e) {

                       //如果client.read(buffer)抛出异常，说明此客户端主动断开连接，需做下面处理

                       client.close();            //关闭channel

                       key.cancel();            //将channel对应的key置为不可用

                       onlineUsers.values().remove(client);    //将问题连接从map中删除

                       System.out.println("-----用户'" + key.attachment().toString() + "'退出连接，当前用户列表：" + onlineUsers.keySet().toString() + "-----");

                       continue;                //跳出循环

                   }

                   if (msg.length() > 0) this.processMsg(msg.toString(), client, key);    //处理消息体

               }

               iKeys.remove();                    //5.处理完一次事件后，要显式的移除

           }

       }

   }

   /**

    * 处理客户端传来的消息

    *

    * @param msg 格式：user_to|body|user_from

    * @throws IOException

    * @Key 这里主要用attach()方法，给通道定义一个表示符

    */

   private void processMsg(String msg, SocketChannel client, SelectionKey key) throws IOException {

       String[] ms = msg.split(seperator);

       if (ms.length == 1) {

           String user = ms[0];    //输入的是自定义用户名

           if (onlineUsers.containsKey(user)) {

               client.write(charset.encode("当前用户已存在，请重新输入用户名："));

           } else {

               onlineUsers.put(user, client);

               key.attach(user);    //给通道定义一个表示符

               //  |字符来作为消息之间的分割符

               client.write(charset.encode("您的昵称通过验证 "+user+"|"));

               String welCome = "\t欢迎'" + user + "'上线，当前在线人数" + this.getOnLineNum() + "人。用户列表：" + onlineUsers.keySet().toString();

               this.broadCast(welCome+"|");    //给所用用户推送上线信息，包括自己

           }

       } else if (ms.length == 3) {

           String user_to = ms[0];

           String msg_body = ms[1];

           String user_from = ms[2];

           SocketChannel channel_to = onlineUsers.get(user_to);

           if (channel_to == null) {

               client.write(charset.encode("用户'" + user_to + "'不存在，当前用户列表：" + onlineUsers.keySet().toString()));

           } else {

               channel_to.write(charset.encode("来自'" + user_from + "'的消息：" + msg_body));

           }

       }

   }

   //map中的有效数量已被很好的控制，可以从map中获取，也可以用下面的方法取

   private int getOnLineNum() {

       int count = 0;

       Channel channel;

       for (SelectionKey k : selector.keys()) {

           channel = k.channel();

           if (channel instanceof SocketChannel) {    //排除ServerSocketChannel

               count++;

           }

       }

       return count;

   }

   //广播上线消息

   private void broadCast(String msg) throws IOException {

       Channel channel;

       for (SelectionKey k : selector.keys()) {

           channel = k.channel();

           if (channel instanceof SocketChannel) {

               SocketChannel client = (SocketChannel) channel;

               client.write(charset.encode(msg));

           }

       }

   }

   public static void main(String[] args) {

       try {

           new ChatServer().startServer();

       } catch (IOException e) {

           e.printStackTrace();

       }

   }

}

客户端：

package NonBlocking;

import java.io.IOException;

import java.net.InetSocketAddress;

import java.nio.ByteBuffer;

import java.nio.channels.*;

import java.nio.charset.Charset;

import java.util.Iterator;

import java.util.Scanner;

public class ChatClient1 {

   private final int port = 8899;

   private final String seperator = "|";

   private final Charset charset = Charset.forName("UTF-8"); //字符集

   private ByteBuffer buffer = ByteBuffer.allocate(1024);

   private SocketChannel _self;

   private Selector selector;

   private String name = "";

   private boolean flag = true; //服务端断开，客户端的读事件不会一直发生（与服务端不一样）

   Scanner scanner = new Scanner(System.in);

   public void startClient() throws IOException{

       //客户端初始化固定流程：4步

       selector = Selector.open(); //1.打开Selector

       _self = SocketChannel.open(new InetSocketAddress(port));//2.连接服务端，这里默认本机的IP

       _self.configureBlocking(false); //3.配置此channel非阻塞

       _self.register(selector, SelectionKey.OP_READ); //4.将channel的读事件注册到选择器

       /*

        * 因为等待用户输入会导致主线程阻塞

        * 所以用主线程处理输入，新开一个线程处理读数据

        */

       new Thread(new ClientReadThread()).start(); //开一个异步线程处理读

       String input = "";

       while(flag){

           input = scanner.nextLine();

           String[] strArray;

           if("".equals(input)){

               System.out.println("不允许输入空串！");

               continue;

           // 如果姓名没有初始化，且长度为1.说明当前在设置姓名

           }else if("".equals(name) && input.split("[|]").length == 1){

               //啥也不干

           // 如果姓名已经初始化过了，且长度为2.说明这是正常的发送格式

           }else if(!"".equals(name) && input.split("[|]").length == 2) {

               input = input + seperator + name;

           }else{

               System.out.println("输入不合法，请重新输入：");

               continue;

           }

           try{

               _self.write(charset.encode(input));

           }catch(Exception e){

               System.out.println(e.getMessage()+"客户端主线程退出连接！！");

           }

       }

   }

   private class ClientReadThread implements Runnable{

       @Override

       public void run(){

           Iterator<SelectionKey> ikeys;

           SelectionKey key;

           SocketChannel client;

           try {

               while(flag){

                   selector.select(); //调用此方法一直阻塞，直到有channel可用

                   ikeys = selector.selectedKeys().iterator();

                   while(ikeys.hasNext()){

                       key = ikeys.next();

                       if(key.isReadable()){ //处理读事件

                           client = (SocketChannel) key.channel();

                           //这里的输出是true，从selector的key中获取的客户端channel，是同一个

// System.out.println("client == _self:"+ (client == _self));

                           buffer.clear();

                           StringBuilder msg = new StringBuilder();

                           try{

                               while(client.read(buffer) > 0){

                                   buffer.flip(); //将写模式转换为读模式

                                   msg.append(charset.decode(buffer));

                               }

                           }catch(IOException en){

                               System.out.println(en.getMessage()+",客户端'"+key.attachment().toString()+"'读线程退出！！");

                               stopMainThread();

                           }

                           String[] StrArray = msg.toString().split("[|]");

                           for (String message : StrArray) {

                               if (message == "") continue;

                               if (message.contains("您的昵称通过验证")) {

                                   if (message.contains("您的昵称通过验证")) {

                                       String[] nameValid = message.split(" ");

                                       name = nameValid[1];

                                       key.attach(name);

                                   }

                               }

                               System.out.println(message);

                           }

                       }

                       ikeys.remove();

                   }

               }

           } catch (Exception e) {

               e.printStackTrace();

           }

       }

   }

   private void stopMainThread(){

       flag = false;

   }

   public static void main(String[] args){

       try {

           new ChatClient1().startClient();

       } catch (IOException e) {

           e.printStackTrace();

       }

   }

}

使用方法：

   仿照着ChatClient1，再弄一个ChatClient2。先运行服务端，再运行客户端1、客户端2。

   客户端里，先输入自己的昵称。

   之后，输入对方的名字和想说的话，用|隔开。

运行效果：

客户端1：

请输入自定义用户名：

张家辉

您的昵称通过验证 张家辉

欢迎'张家辉'上线，当前在线人数1人。用户列表：[张家辉]

张家辉|我跟自己说话

来自'张家辉'的消息：我跟自己说话

古天乐|你在吗

用户'古天乐'不存在，当前用户列表：[张家辉]

欢迎'古天乐'上线，当前在线人数2人。用户列表：[张家辉, 古天乐]

古天乐|你终于来了，兄弟

来自'古天乐'的消息：咋了，兄弟

古天乐|来玩贪玩蓝月啊

客户端2：

请输入自定义用户名：

古天乐

您的昵称通过验证 古天乐

欢迎'古天乐'上线，当前在线人数2人。用户列表：[张家辉, 古天乐]

来自'张家辉'的消息：你终于来了，兄弟

张家辉|咋了，兄弟

来自'张家辉'的消息：来玩贪玩蓝月啊

参考博客

java NIO及NIO聊天室，相比其中内容，修复了用户名已存在的情况，自己客户端的名字还是会初始化为这个已存在名字的bug。原文代码为单人聊天，经过修改为多人聊天（多人聊天更简单点）。