Java网络编程系列之基于BIO的多人聊天室设计与实现

转x33g5p2x  于2021-12-07 转载在 Java  
字(8.4k)|赞(0)|评价(0)|浏览(615)

BIO模型

传统的BIO模型（同步阻塞IO模型）+线程池（多线程）模式：适合活动连接次数不是特别高。该模式是1:1，即每次连接每个线程。

处理步骤：客户端发送请求，接收器Acceptor每接收一个请求，就创建一个新线程，处理完成之后，再通过输出流返回到客户端，然后销毁线程。

缺陷：一个客户端请求，就对应一个线程，客户端的请求和服务端的线程就成为1:1的比例，当请求过多的时候，线程越来越多，最后导致JVM的内存被大量的占用，堆栈溢出，发生异常。

多人聊天室功能概述

多人聊天室设计UML建模之时序图

  • 1.服务器端: 通过ServerSocket对象，绑定端口，调用accept函数，等待客户端连接
  • 2.服务器端维护一个map集合，通过每个客户端的端口号，来唯一识别每一个客户端对象
  • 2.当有客户端连接成功后，通过ChatHandler创建一个新的线程用以处理当前客户端的连接
  • 4.ChatHandler负责将当前连接成功的客户端放入当前在线用户集合中，然后保持与当前客户端的线程连接，直到当前客户端主动退出连接
  • 5.客户端ChatClient通过服务器ip和端口与之建立连接，然后等待接收服务器发送过来的消息
  • 6.同时客户端创建一个单独的线程UserInputHandler，负责发送消息，当客户端这边输入quit指令的时候，表示客户端要退出连接

服务端代码实现

ChatServer

  1. import java.io.BufferedWriter;
  2. import java.io.IOException;
  3. import java.io.OutputStreamWriter;
  4. import java.io.Writer;
  5. import java.net.ServerSocket;
  6. import java.net.Socket;
  7. import java.util.Map;
  8. import java.util.concurrent.ConcurrentHashMap;
  10. public class ChatServer
  11. {
  12.     private  final int SERVER_PORT=8080;
  13.     private final String QUIT="quit";
  15.     private ServerSocket serverSocket;
  16.     //key是当前客户端对应的端口号,value是服务器与当前客户端之前的关联输出字符流
  17.     private Map<Integer, Writer> connectedClients;
  19.     public ChatServer()
  20.     {
  21.         //初始化map集合
  22.         connectedClients=new ConcurrentHashMap<>();
  23.     }
  25.     //新增客户端
  26.     public synchronized void addClient(Socket socket) throws IOException {
  27.         if(socket!=null)
  28.         {
  29.             //添加进集合
  30.             connectedClients.put(socket.getPort(),new BufferedWriter(new OutputStreamWriter(socket.getOutputStream())));
  31.             System.out.println("当前客户端["+socket.getPort()+"]已成功连接到服务器");
  32.         }
  33.     }
  35.     //移除客户端
  36.     public synchronized void removeClient(Socket socket) throws IOException {
  37.         if(socket!=null)
  38.         {
  39.             //从集合中移除，并关闭相关流
  40.             if(connectedClients.containsKey(socket.getPort()))
  41.             {
  42.                 connectedClients.get(socket.getPort()).close();
  43.                 connectedClients.remove(socket.getPort());
  44.                 System.out.println("客户端[" + socket.getPort() + "]已断开连接");
  45.             }
  46.         }
  47.     }
  49.     //群发消息给其他客户端
  50.     public synchronized void forwardMessage(Socket socket,String msg) throws IOException {
  51.         //遍历集合，挨个转发消息--但是不发给自己
  52. // connectedClients.forEach(
  53. // (port,client)->
  54. // {
  55. // if(!port.equals(socket.getPort()))
  56. // {
  57. // System.out.println("当前客户端["+port+"]");
  58. // //这里最好还是把异常外抛出去，因为这里使用lambda写法，因此没法抛出去
  59. // try
  60. // {
  61. // client.write(msg+"\n");
  62. // client.flush();
  63. // }
  64. // catch (IOException e)
  65. // {
  66. // e.printStackTrace();
  67. // }
  68. // }
  69. // }
  70. // );
  71.         for (Integer id : connectedClients.keySet()) {
  72.             if (!id.equals(socket.getPort())) {
  73.                 Writer writer = connectedClients.get(id);
  74.                 //这个地方必须要加\n,否则readline读取不到换行符，会阻塞住
  75.                 writer.write(msg+"\n");
  76.                 writer.flush();
  77.             }
  78.         }
  79.     }
  81.     //客户端是否准备好退出连接
  82.     public boolean clientReadyToQuit(String msg)
  83.     {
  84.           return QUIT.equals(msg);
  85.     }
  87.     //关闭服务器端口
  88.     public synchronized void close() throws IOException {
  89.             serverSocket.close();
  90.             System.out.println("关闭socket服务器");
  91.     }
  93.     //启动服务器
  94.     public void start()
  95.     {
  96.         try
  97.         {
  98.             //绑定服务器与对应的端口
  99.             ServerSocket serverSocket = new ServerSocket(SERVER_PORT);
  100.             System.out.println("服务器启动,对应的端口为: "+SERVER_PORT);
  101.             //等待客户端连接
  102.             while(true)
  103.             {
  104.                 //如果没有客户端连接，这边就会阻塞住
  105.                 System.out.println("等待客户端连接...");
  106.                 Socket socket = serverSocket.accept();
  107.                 System.out.println("客户端连接中...");
  108.                 //有客户端连接后，创建ChatHandler线程
  109.                new Thread(new ChatHandler(this,socket)).start();
  110.             }
  111.         } catch (IOException e) {
  112.             e.printStackTrace();
  113.         }
  114.     }
  116.     //启动服务
  117.     public static void main(String[] args) {
  118.         ChatServer chatServer=new ChatServer();
  119.         chatServer.start();
  120.     }
  121. }

ChatHandler

  1. import java.io.BufferedReader;
  2. import java.io.IOException;
  3. import java.io.InputStreamReader;
  4. import java.net.Socket;
  6. public class ChatHandler implements Runnable
  7. {
  8.     //保存服务器对象和对应的客户端socket连接对象
  9.     private ChatServer chatServer;
  10.     private  Socket socket;
  12.    ChatHandler(ChatServer chatServer,Socket socket)
  13.    {
  14.        this.chatServer=chatServer;
  15.        this.socket=socket;
  16.    }
  18.     @Override
  19.     public void run()
  20.     {
  21.         try
  22.         {
  23.             //保存新用户连接进在线用户集合
  24.             chatServer.addClient(socket);
  26.             //读取用户发送的信息
  27.             BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
  28.             //判断当前用户是否发送了消息
  29.             String msg=null;
  30.             //bufferedReader.readLine()是阻塞式的,直到收到客户端发送过来的一条信息
  31.             //读取直到信息是以换行符或者回车结尾,然后返回换行符之前的所有信息
  32.             //当客户端断开连接后,此时readLine函数会返回null值
  33.             while((msg=bufferedReader.readLine())!=null)
  34.             {
  35.                 //检查当前用户是否准备退出
  36.                 if(chatServer.clientReadyToQuit(msg))
  37.                 {
  38.                     break;
  39.                 }
  40.                 String clientMsg = "客户端[" + socket.getPort() + "]的消息: " + msg;
  41.                 System.out.println(clientMsg);
  42.                 //转发消息
  43.                 chatServer.forwardMessage(socket,clientMsg);
  44.             }
  45.         }
  46.         catch (IOException e)
  47.         {
  48.             e.printStackTrace();
  49.         }finally
  50.         {
  51.             try {
  52.                 chatServer.removeClient(socket);
  53.             } catch (IOException e) {
  54.                 e.printStackTrace();
  55.             }
  56.         }
  57.     }
  58. }

'

客户端代码实现

ChatClient

  1. //客户端
  2. public class ChatClient
  3. {
  4.     private final String DEFAULT_SERVER_HOST = "127.0.0.1";
  5.     private final int DEFAULT_SERVER_PORT = 8080;
  6.     private final String QUIT = "quit";
  8.     private Socket socket;
  9.     private BufferedReader reader;
  10.     private BufferedWriter writer;
  12.     // 发送消息给服务器
  13.     public void send(String msg) throws IOException
  14.     {
  15.         //当前当前客户端与服务器端之间的输出流是否关闭
  16.         if (!socket.isOutputShutdown())
  17.         {
  18.             writer.write(msg + "\n");
  19.             writer.flush();
  20.         }
  21.     }
  23.     // 从服务器接收消息
  24.     public String receive() throws IOException {
  25.         String msg = null;
  26.         //判断当前客户端与服务器端之间的输入流是否关闭
  27.         if (!socket.isInputShutdown()) {
  28.             msg = reader.readLine();
  29.         }
  30.         return msg;
  31.     }
  33.     // 检查用户是否准备退出
  34.     public boolean readyToQuit(String msg) {
  35.         return QUIT.equals(msg);
  36.     }
  38.     public void close() {
  39.         if (writer != null) {
  40.             try {
  41.                 System.out.println("关闭socket");
  42.                 writer.close();
  43.             } catch (IOException e) {
  44.                 e.printStackTrace();
  45.             }
  46.         }
  47.     }
  49.     public void start()
  50.     {
  51.         try {
  52.             // 创建socket
  53.             socket = new Socket(DEFAULT_SERVER_HOST, DEFAULT_SERVER_PORT);
  55.             // 创建IO流
  56.             reader = new BufferedReader(
  57.                     new InputStreamReader(socket.getInputStream())
  58.             );
  59.             writer = new BufferedWriter(
  60.                     new OutputStreamWriter(socket.getOutputStream())
  61.             );
  63.             // 处理用户的输入
  64.             new Thread(new UserInputHandler(this)).start();
  66.             // 读取服务器转发的消息
  67.             String msg = null;
  68.             while ((msg = receive()) != null) {
  69.                 System.out.println(msg);
  70.             }
  71.         } catch (IOException e) {
  72.             e.printStackTrace();
  73.         } finally {
  74.             close();
  75.         }
  76.     }
  78.     public static void main(String[] args) {
  79.         ChatClient chatClient = new ChatClient();
  80.         chatClient.start();
  81.     }
  82. }

UserInputHandler

  1. public class UserInputHandler implements Runnable {
  3.     private ChatClient chatClient;
  5.     public UserInputHandler(ChatClient chatClient) {
  6.         this.chatClient = chatClient;
  7.     }
  9.     @Override
  10.     public void run() {
  11.         try {
  12.             // 等待用户输入消息
  13.             BufferedReader consoleReader =
  14.                     new BufferedReader(new InputStreamReader(System.in));
  15.             while (true) {
  16.                 String input = consoleReader.readLine();
  18.                 // 向服务器发送消息
  19.                 chatClient.send(input);
  21.                 // 检查用户是否准备退出
  22.                 if (chatClient.readyToQuit(input)) {
  23.                     break;
  24.                 }
  25.             }
  26.         } catch (IOException e) {
  27.             e.printStackTrace();
  28.         }
  29.     }
  30. }

伪异步IO编程模型简析

上面版本的聊天室缺陷:

  • 服务端会为每一个客户端都创建一个新的线程用来通信，这样比较耗费资源，好的做法是创建一个线程池用来使用

复习java提供的线程池操作:

代码实现

  1. public class ChatServer
  2. {
  3.     private  final int SERVER_PORT=8080;
  4.     private final String QUIT="quit";
  6.     private ServerSocket serverSocket;
  7.     //key是当前客户端对应的端口号,value是服务器与当前客户端之前的关联输出字符流
  8.     private Map<Integer, Writer> connectedClients;
  9.     //维护一个线程池对象
  10.     private ExecutorService executorService;
  12.     public ChatServer()
  13.     {
  14.         //初始化map集合
  15.         connectedClients=new ConcurrentHashMap<>();
  16.         //线程池对象初始化---线程池里面固定线程数量
  17.         executorService=Executors.newFixedThreadPool(10);
  18.     }
  20.     //新增客户端
  21.     public synchronized void addClient(Socket socket) throws IOException {
  22.         if(socket!=null)
  23.         {
  24.             //添加进集合
  25.             connectedClients.put(socket.getPort(),new BufferedWriter(new OutputStreamWriter(socket.getOutputStream())));
  26.             System.out.println("当前客户端["+socket.getPort()+"]已成功连接到服务器");
  27.         }
  28.     }
  30.     //移除客户端
  31.     public synchronized void removeClient(Socket socket) throws IOException {
  32.         if(socket!=null)
  33.         {
  34.             //从集合中移除，并关闭相关流
  35.             if(connectedClients.containsKey(socket.getPort()))
  36.             {
  37.                 connectedClients.get(socket.getPort()).close();
  38.                 connectedClients.remove(socket.getPort());
  39.                 System.out.println("客户端[" + socket.getPort() + "]已断开连接");
  40.             }
  41.         }
  42.     }
  44.     //群发消息给其他客户端
  45.     public synchronized void forwardMessage(Socket socket,String msg) throws IOException {
  46.         //遍历集合，挨个转发消息--但是不发给自己
  47. // connectedClients.forEach(
  48. // (port,client)->
  49. // {
  50. // if(!port.equals(socket.getPort()))
  51. // {
  52. // System.out.println("当前客户端["+port+"]");
  53. // //这里最好还是把异常外抛出去，因为这里使用lambda写法，因此没法抛出去
  54. // try
  55. // {
  56. // client.write(msg+"\n");
  57. // client.flush();
  58. // }
  59. // catch (IOException e)
  60. // {
  61. // e.printStackTrace();
  62. // }
  63. // }
  64. // }
  65. // );
  66.         for (Integer id : connectedClients.keySet()) {
  67.             if (!id.equals(socket.getPort())) {
  68.                 Writer writer = connectedClients.get(id);
  69.                 //这个地方必须要加\n,否则readline读取不到换行符，会阻塞住
  70.                 writer.write(msg+"\n");
  71.                 writer.flush();
  72.             }
  73.         }
  74.     }
  76.     //客户端是否准备好退出连接
  77.     public boolean clientReadyToQuit(String msg)
  78.     {
  79.           return QUIT.equals(msg);
  80.     }
  82.     //关闭服务器端口
  83.     public synchronized void close() throws IOException {
  84.             serverSocket.close();
  85.             System.out.println("关闭socket服务器");
  86.     }
  88.     //启动服务器
  89.     public void start()
  90.     {
  91.         try
  92.         {
  93.             //绑定服务器与对应的端口
  94.             ServerSocket serverSocket = new ServerSocket(SERVER_PORT);
  95.             System.out.println("服务器启动,对应的端口为: "+SERVER_PORT);
  96.             //等待客户端连接
  97.             while(true)
  98.             {
  99.                 //如果没有客户端连接，这边就会阻塞住
  100.                 System.out.println("等待客户端连接...");
  101.                 Socket socket = serverSocket.accept();
  102.                 System.out.println("客户端连接中...");
  103.                 //有客户端连接后，创建ChatHandler线程
  104.                //new Thread(new ChatHandler(this,socket)).start();
  105.                //使用线程池管理客户端连接---这里excute会调用线程的start方法
  106.                 executorService.execute(new ChatHandler(this,socket));
  107.             }
  108.         } catch (IOException e) {
  109.             e.printStackTrace();
  110.         }
  111.     }
  113.     //启动服务
  114.     public static void main(String[] args) {
  115.         ChatServer chatServer=new ChatServer();
  116.         chatServer.start();
  117.     }
  118. }

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

原文链接 : https://blog.csdn.net/m0_53157173/article/details/121756354

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

Javanio网络开发语言

相关文章

    查看更多

    热门标签

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

    最新文章

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