程序员社区

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

2022-03-10 分类：未分类 阅读(325)

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

  • BIO模型
  • 多人聊天室功能概述
    • 多人聊天室设计UML建模之时序图
    • 服务端代码实现
      • ChatServer
      • ChatHandler
    • 客户端代码实现
      • ChatClient
      • UserInputHandler
    • 伪异步IO编程模型简析

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

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.Map;
import java.util.concurrent.ConcurrentHashMap;

public class ChatServer
{
    private  final int SERVER_PORT=8080;
    private final String QUIT="quit";

    private ServerSocket serverSocket;
    //key是当前客户端对应的端口号,value是服务器与当前客户端之前的关联输出字符流
    private Map<Integer, Writer> connectedClients;

    public ChatServer()
    {
        //初始化map集合
        connectedClients=new ConcurrentHashMap<>();
    }

    //新增客户端
    public synchronized void addClient(Socket socket) throws IOException {
        if(socket!=null)
        {
            //添加进集合
            connectedClients.put(socket.getPort(),new BufferedWriter(new OutputStreamWriter(socket.getOutputStream())));
            System.out.println("当前客户端["+socket.getPort()+"]已成功连接到服务器");
        }
    }

    //移除客户端
    public synchronized void removeClient(Socket socket) throws IOException {
        if(socket!=null)
        {
            //从集合中移除，并关闭相关流
            if(connectedClients.containsKey(socket.getPort()))
            {
                connectedClients.get(socket.getPort()).close();
                connectedClients.remove(socket.getPort());
                System.out.println("客户端[" + socket.getPort() + "]已断开连接");
            }
        }
    }

    //群发消息给其他客户端
    public synchronized void forwardMessage(Socket socket,String msg) throws IOException {
        //遍历集合，挨个转发消息--但是不发给自己
//        connectedClients.forEach(
//                (port,client)->
//                {
//                          if(!port.equals(socket.getPort()))
//                          {
//                              System.out.println("当前客户端["+port+"]");
//                              //这里最好还是把异常外抛出去，因为这里使用lambda写法，因此没法抛出去
//                              try
//                              {
//                                  client.write(msg+"\n");
//                                  client.flush();
//                              }
//                              catch (IOException e)
//                              {
//                                  e.printStackTrace();
//                              }
//                          }
//                }
//        );
        for (Integer id : connectedClients.keySet()) {
            if (!id.equals(socket.getPort())) {
                Writer writer = connectedClients.get(id);
                //这个地方必须要加\n,否则readline读取不到换行符，会阻塞住
                writer.write(msg+"\n");
                writer.flush();
            }
        }
    }

    //客户端是否准备好退出连接
    public boolean clientReadyToQuit(String msg)
    {
          return QUIT.equals(msg);
    }

    //关闭服务器端口
    public synchronized void close() throws IOException {
            serverSocket.close();
            System.out.println("关闭socket服务器");
    }

    //启动服务器
    public void start()
    {
        try
        {
            //绑定服务器与对应的端口
            ServerSocket serverSocket = new ServerSocket(SERVER_PORT);
            System.out.println("服务器启动,对应的端口为: "+SERVER_PORT);
            //等待客户端连接
            while(true)
            {
                //如果没有客户端连接，这边就会阻塞住
                System.out.println("等待客户端连接...");
                Socket socket = serverSocket.accept();
                System.out.println("客户端连接中...");
                //有客户端连接后，创建ChatHandler线程
               new Thread(new ChatHandler(this,socket)).start();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //启动服务
    public static void main(String[] args) {
        ChatServer chatServer=new ChatServer();
        chatServer.start();
    }
}

ChatHandler

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.Socket;

public class ChatHandler implements Runnable
{
    //保存服务器对象和对应的客户端socket连接对象
    private ChatServer chatServer;
    private  Socket socket;

   ChatHandler(ChatServer chatServer,Socket socket)
   {
       this.chatServer=chatServer;
       this.socket=socket;
   }


    @Override
    public void run()
    {
        try
        {
            //保存新用户连接进在线用户集合
            chatServer.addClient(socket);

            //读取用户发送的信息
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            //判断当前用户是否发送了消息
            String msg=null;
            //bufferedReader.readLine()是阻塞式的,直到收到客户端发送过来的一条信息
            //读取直到信息是以换行符或者回车结尾,然后返回换行符之前的所有信息
            //当客户端断开连接后,此时readLine函数会返回null值
            while((msg=bufferedReader.readLine())!=null)
            {
                //检查当前用户是否准备退出
                if(chatServer.clientReadyToQuit(msg))
                {
                    break;
                }
                String clientMsg = "客户端[" + socket.getPort() + "]的消息: " + msg;
                System.out.println(clientMsg);
                //转发消息
                chatServer.forwardMessage(socket,clientMsg);
            }
        }
        catch (IOException e)
        {
            e.printStackTrace();
        }finally
        {
            try {
                chatServer.removeClient(socket);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

'

客户端代码实现

ChatClient

//客户端
public class ChatClient
{
    private final String DEFAULT_SERVER_HOST = "127.0.0.1";
    private final int DEFAULT_SERVER_PORT = 8080;
    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.equals(msg);
    }


    public void close() {
        if (writer != null) {
            try {
                System.out.println("关闭socket");
                writer.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public void start()
    {
        try {
            // 创建socket
            socket = new Socket(DEFAULT_SERVER_HOST, DEFAULT_SERVER_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 chatClient = new ChatClient();
        chatClient.start();
    }
}

UserInputHandler

public class UserInputHandler implements Runnable {

    private ChatClient chatClient;

    public UserInputHandler(ChatClient chatClient) {
        this.chatClient = chatClient;
    }

    @Override
    public void run() {
        try {
            // 等待用户输入消息
            BufferedReader consoleReader =
                    new BufferedReader(new InputStreamReader(System.in));
            while (true) {
                String input = consoleReader.readLine();

                // 向服务器发送消息
                chatClient.send(input);

                // 检查用户是否准备退出
                if (chatClient.readyToQuit(input)) {
                    break;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

伪异步IO编程模型简析

上面版本的聊天室缺陷:

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

在这里插入图片描述
复习java提供的线程池操作:

在这里插入图片描述
在这里插入图片描述

代码实现

在这里插入图片描述
在这里插入图片描述

public class ChatServer
{
    private  final int SERVER_PORT=8080;
    private final String QUIT="quit";

    private ServerSocket serverSocket;
    //key是当前客户端对应的端口号,value是服务器与当前客户端之前的关联输出字符流
    private Map<Integer, Writer> connectedClients;
    //维护一个线程池对象
    private ExecutorService executorService;

    public ChatServer()
    {
        //初始化map集合
        connectedClients=new ConcurrentHashMap<>();
        //线程池对象初始化---线程池里面固定线程数量
        executorService=Executors.newFixedThreadPool(10);
    }

    //新增客户端
    public synchronized void addClient(Socket socket) throws IOException {
        if(socket!=null)
        {
            //添加进集合
            connectedClients.put(socket.getPort(),new BufferedWriter(new OutputStreamWriter(socket.getOutputStream())));
            System.out.println("当前客户端["+socket.getPort()+"]已成功连接到服务器");
        }
    }

    //移除客户端
    public synchronized void removeClient(Socket socket) throws IOException {
        if(socket!=null)
        {
            //从集合中移除，并关闭相关流
            if(connectedClients.containsKey(socket.getPort()))
            {
                connectedClients.get(socket.getPort()).close();
                connectedClients.remove(socket.getPort());
                System.out.println("客户端[" + socket.getPort() + "]已断开连接");
            }
        }
    }

    //群发消息给其他客户端
    public synchronized void forwardMessage(Socket socket,String msg) throws IOException {
        //遍历集合，挨个转发消息--但是不发给自己
//        connectedClients.forEach(
//                (port,client)->
//                {
//                          if(!port.equals(socket.getPort()))
//                          {
//                              System.out.println("当前客户端["+port+"]");
//                              //这里最好还是把异常外抛出去，因为这里使用lambda写法，因此没法抛出去
//                              try
//                              {
//                                  client.write(msg+"\n");
//                                  client.flush();
//                              }
//                              catch (IOException e)
//                              {
//                                  e.printStackTrace();
//                              }
//                          }
//                }
//        );
        for (Integer id : connectedClients.keySet()) {
            if (!id.equals(socket.getPort())) {
                Writer writer = connectedClients.get(id);
                //这个地方必须要加\n,否则readline读取不到换行符，会阻塞住
                writer.write(msg+"\n");
                writer.flush();
            }
        }
    }

    //客户端是否准备好退出连接
    public boolean clientReadyToQuit(String msg)
    {
          return QUIT.equals(msg);
    }

    //关闭服务器端口
    public synchronized void close() throws IOException {
            serverSocket.close();
            System.out.println("关闭socket服务器");
    }

    //启动服务器
    public void start()
    {
        try
        {
            //绑定服务器与对应的端口
            ServerSocket serverSocket = new ServerSocket(SERVER_PORT);
            System.out.println("服务器启动,对应的端口为: "+SERVER_PORT);
            //等待客户端连接
            while(true)
            {
                //如果没有客户端连接，这边就会阻塞住
                System.out.println("等待客户端连接...");
                Socket socket = serverSocket.accept();
                System.out.println("客户端连接中...");
                //有客户端连接后，创建ChatHandler线程
               //new Thread(new ChatHandler(this,socket)).start();
               //使用线程池管理客户端连接---这里excute会调用线程的start方法
                executorService.execute(new ChatHandler(this,socket));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //启动服务
    public static void main(String[] args) {
        ChatServer chatServer=new ChatServer();
        chatServer.start();
    }
}

赞(0) 打赏
未经允许不得转载：IDEA激活码 » Java网络编程系列之基于BIO的多人聊天室设计与实现

相关推荐

  • 暂无文章

IDEA永久激活破解教程,IDEA激活码

pycharm激活破解教程,pycharm激活码

Java视频资源

Java电子书

DataGrip激活教程 Docker Git教程 IntellIJ idea IntelliJ IDEA激活码 Java Java IO Java基础 Java基础面试题 Java集合面试题 Java面试题 JVM JVM面试题 Linux MySQL MySQL面试题 Nginx PyCharm PyCharm激活 redis Redis SpringBoot SpringBoot SpringBoot面试题 Web开发 剑指Offer系列 多线程 多线程并发面试题 字符串面试题 数据库 数据库 数据结构与算法 未分类 爬虫 电子书 软件下载

一个分享Java & Python知识的社区