Java 用NIO实现一个聊天室（多人聊天、单人聊天）

朱雀 2023-07-16 03:53 11阅读 0赞

### 文章目录 ###

*  多人在线，多人聊天（可能有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();
            }
        }
    }

客户端：

使用方法：

*  仿照着 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聊天室][java NIO_NIO]，相比其中内容，修复了用户名已存在的情况，自己客户端的名字还是会初始化为这个已存在名字的bug。原文代码为单人聊天，经过修改为多人聊天（多人聊天更简单点）。

[java NIO_NIO]: https://blog.csdn.net/lh87522/article/details/46327601