Java网络编程-Socket编程初涉三（伪异步I/O模型的简易多人聊天室）

向右看齐 2023-07-03 07:07 2阅读 0赞

推荐：[Java网络编程汇总][Java]

## Java网络编程-Socket编程初涉三（伪异步I/O模型的简易多人聊天室） ##

**什么是伪异步I/O模型？**

[BIO模型与伪异步I/O模型][BIO_I_O]

现在应该了解了什么是`伪异步I/O模型`了吧，看起来它和`BIO模型`其实差别不大，只不过`伪异步I/O模型`是用线程池来管理线程去和客户端进行数据交互，而`BIO模型`是每次客户端的连接请求成功后，都创建新的线程去与客户端进行数据交互，虽然差别不大，但`伪异步I/O模型`的优势还是很明显的，尤其当客户端连接请求并发数比较大时。

两个模型差别不大，也就意味着需要改动代码的地方很少，如果需要代码解释，请看下面这篇博客。  
[Java网络编程-Socket编程初涉二（基于BIO模型的简易多人聊天室）][Java_-Socket_BIO]

**改进**

只需要改动服务器端的代码，将每次客户端连接成功后都创建新线程来与客户端进行数据交互，改进成由线程池来管理线程去与客户端进行数据交互。

ChatServer类（`有改动`）。

package bio.chatroom.server;
    
    import java.io.BufferedWriter;
    import java.io.IOException;
    import java.io.OutputStreamWriter;
    import java.io.Writer;
    import java.net.ServerSocket;
    import java.net.Socket;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class ChatServer { 
    
        private final int DEFAULT_PORT = 8888;
        private final String QUIT = "quit";
    
        private ExecutorService executorService;
        private ServerSocket serverSocket;
    
        // 把客户端的port当作客户端的id
        private Map<Integer , Writer> connectedClients;
    
        public ChatServer(){ 
            executorService = Executors.newFixedThreadPool(10);
            connectedClients = new HashMap<>();
        }
    
        public synchronized void addClient(Socket socket) throws IOException { 
            if(socket != null){ 
                int port = socket.getPort();
                BufferedWriter writer = new BufferedWriter(
                        new OutputStreamWriter(socket.getOutputStream())
                );
                connectedClients.put(port , writer);
                System.out.println("客户端["+port+"]已连接到服务器");
            }
        }
    
        public synchronized void removeClient(Socket socket) throws IOException { 
            if(socket != null){ 
                int port = socket.getPort();
                if(connectedClients.containsKey(port)){ 
                    connectedClients.get(port).close();
                    connectedClients.remove(port);
                    System.out.println("客户端["+port+"]已断开连接");
                }
            }
        }
    
        public synchronized void forwardMessage(Socket socket , String fwdMsg) throws IOException { 
            // 发送消息的端口
            int sendMessagePort = socket.getPort();
            for(Integer port : connectedClients.keySet()){ 
                if(!port.equals(sendMessagePort)){ 
                    Writer writer = connectedClients.get(port);
                    writer.write(fwdMsg);
                    writer.flush();
                }
            }
        }
    
        public boolean readyToQuit(String msg){ 
            return QUIT.equalsIgnoreCase(msg);
        }
    
        public synchronized void close(){ 
            if(serverSocket != null){ 
                try { 
                    serverSocket.close();
                    System.out.println("关闭了ServerSocket");
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        public void start(){ 
            try { 
                // 创建ServerSocket，绑定和监听端口
                serverSocket = new ServerSocket(DEFAULT_PORT);
                System.out.println("启动服务器，监听端口"+DEFAULT_PORT+"...");
    
                while(true){ 
                    // 等待客户端连接
                    Socket socket = serverSocket.accept();
                    // 创建ChatHandler线程
    // new Thread(new ChatHandler(this , socket)).start();
                    executorService.execute(new ChatHandler(this , socket));
                }
            } catch (IOException e) { 
                e.printStackTrace();
            }finally { 
                close();
            }
        }
    
        public static void main(String[] args) { 
            ChatServer server = new ChatServer();
            server.start();
        }
    }

ChatHandler类（`无改动`）。

package bio.chatroom.server;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.net.Socket;
    
    public class ChatHandler implements Runnable{ 
    
        private ChatServer server;
        private Socket socket;
    
        public ChatHandler(ChatServer server , Socket socket){ 
            this.server = server;
            this.socket = socket;
        }
    
        @Override
        public void run() { 
            try { 
                // 存储新上线用户
                server.addClient(socket);
    
                // 读取用户发送的消息
                BufferedReader reader = new BufferedReader(
                        new InputStreamReader(socket.getInputStream())
                );
    
                String msg = null;
                while((msg = reader.readLine()) != null){ 
                    String fwdMsg = "客户端["+socket.getPort()+"]："+msg+"\n";
                    System.out.print(fwdMsg);
    
                    // 将消息转发给聊天室里在线的其他用户
                    server.forwardMessage(socket , fwdMsg);
    
                    // 检查用户是否准备退出
                    if(server.readyToQuit(msg)){ 
                        break;
                    }
                }
            } catch (IOException e) { 
                e.printStackTrace();
            }finally { 
                try { 
                    // 从服务器移除退出的用户
                    server.removeClient(socket);
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

ChatClient类（`无改动`）。

package bio.chatroom.client;
    
    import java.io.*;
    import java.net.Socket;
    
    public class ChatClient { 
    
        private final String DEFAULT_SERVER_HOST = "127.0.0.1";
        private final int DEFAULT_PORT = 8888;
        private final String QUIT = "quit";
    
        private Socket socket;
        private BufferedReader reader;
        private BufferedWriter writer;
    
        // 发送消息给服务器
        public void send(String msg) throws IOException { 
            if(!socket.isOutputShutdown()){ 
                writer.write(msg+"\n");
                writer.flush();
            }
        }
    
        // 接收服务器的消息
        public String receive() throws IOException { 
            String msg = null;
            if(!socket.isInputShutdown()){ 
                msg = reader.readLine();
            }
            return msg;
        }
    
        // 检查用户是否准备退出
        public boolean readyToQuit(String msg){ 
            return QUIT.equalsIgnoreCase(msg);
        }
    
        public void close(){ 
            if(writer != null){ 
                try { 
                    writer.close();
                    System.out.println("关闭socket");
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        public void start(){ 
            try { 
                // 创建socket
                socket = new Socket(DEFAULT_SERVER_HOST , DEFAULT_PORT);
    
                // 创建IO流
                reader = new BufferedReader(
                        new InputStreamReader(socket.getInputStream())
                );
                writer = new BufferedWriter(
                        new OutputStreamWriter(socket.getOutputStream())
                );
    
                // 处理用户的输入
                new Thread(new UserInputHandler(this)).start();
    
                // 读取服务器转发的消息
                String msg = null;
                while((msg = receive()) != null){ 
                    System.out.println(msg);
                }
            } catch (IOException e) { 
                e.printStackTrace();
            } finally{ 
                close();
            }
        }
    
        public static void main(String[] args) { 
            ChatClient client = new ChatClient();
            client.start();
        }
    }

UserInputHandler类（`无改动`）。

package bio.chatroom.client;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    
    public class UserInputHandler implements Runnable{ 
    
        private ChatClient client;
    
        public UserInputHandler(ChatClient client){ 
            this.client = client;
        }
    
        @Override
        public void run() { 
            try { 
                // 等待用户输入消息
                BufferedReader consoleReader = new BufferedReader(
                        new InputStreamReader(System.in)
                );
    
                while(true){ 
                    String input = consoleReader.readLine();
    
                    // 向服务器发送消息
                    client.send(input);
    
                    //检查用户是否准备退出
                    if(client.readyToQuit(input)){ 
                        break;
                    }
                }
            } catch (IOException e){ 
                e.printStackTrace();
            }
        }
    }

**改动的地方**  
增加了一个属性。

private ExecutorService executorService;

构造器也需要改，需要初始化这个`executorService`。

public ChatServer(){ 
            executorService = Executors.newFixedThreadPool(10);
            connectedClients = new HashMap<>();
        }

当客户端连接成功后，创建新线程的地方也需要改。

// 创建ChatHandler线程
    // new Thread(new ChatHandler(this , socket)).start();
                    executorService.execute(new ChatHandler(this , socket));

就需要改这三个地方。

由改进的代码可以知道，当大量客户端向服务器发出连接请求后，服务器虽然可以与这些客户端进行连接，但最多只有`10个客户端同时在线`，而其他的客户端会处于等待状态，也就是服务器最多同时存在`10个线程来与客户端进行数据交互`。

关于线程池的使用，这里也只是涉及皮毛，也不打算多讲，相信大家都看得懂。

这里便完成了`伪异步I/O模型`的简易多人聊天室，大家可以自己实现一下。

**测试**

为了方便测试，将线程池允许容纳的线程数设置为`2`。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 3]

测试是没问题的。

如果有说错的地方，请大家不吝赐教（记得留言哦~~~~）。

[Java]: https://kaven.blog.csdn.net/article/details/104140118
[BIO_I_O]: https://blog.csdn.net/qq_37960603/article/details/104153857
[Java_-Socket_BIO]: https://blog.csdn.net/qq_37960603/article/details/104157914
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200203210307274.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200203210321341.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200203210352484.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200203210411965.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9rYXZlbi5ibG9nLmNzZG4ubmV0,size_16,color_FFFFFF,t_70