前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >BIO在聊天室项目中的演化

BIO在聊天室项目中的演化

作者头像
喜欢天文的pony站长
发布2020-07-08 15:49:44
4730
发布2020-07-08 15:49:44
举报
文章被收录于专栏:RabbitMQ实战

通过聊天室项目的演化。介绍BIO的基本用法与优缺点。

image.png

  • 提示:注意阅读代码和注释。

# 提要:

  1. 第一版: echo聊天室
  • 服务器接收到客户端发送的消息,并打印
  • 服务端将客户端发送的消息经过包装后再次发送给客户端
  • 客户端断开连接
  • eg:
    • client: greet from socket.
    • server: echo from server: <greet from socket.>
  1. 第二版: 群聊聊天室
  • 服务器接收客户端发送的消息,并打印
  • 服务端将客户端发送的消息转发给其他在线的客户端
  • 客户端可一直保持在线状态

# 基础配置与工具类

  • 基础配置
代码语言:javascript
复制
/**
 * 常量
 *
 * @author futao
 * @date 2020/7/2
 */
public class Constants {

    /**
     * 服务器端口
     */
    public static final int SERVER_PORT = 9507;

    /**
     * 关键字-客户端退出系统
     */
    public static final String KEY_WORD_QUIT = "quit";

    /**
     * 使用的字符集编码
     */
    public static final Charset CHARSET = StandardCharsets.UTF_8;
}
  • IO工具类
代码语言:javascript
复制

/**
 * @author futao
 * @date 2020/7/2
 */
public class IOUtils {

    /**
     * 从输入流中读取字符串
     *
     * @param is 输入流
     * @return 读取到的字符串
     * @throws IOException
     */
    public static String readString(InputStream is) throws IOException {
        //使用带有缓冲区的BufferInputStream以提高读取性能
        BufferedInputStream bufferedInputStream = new BufferedInputStream(is);
        //从缓冲区中一次读取的数据
        byte[] buffer = new byte[1024 * 4];
        //读取的字符串
        StringBuilder fullMessage = new StringBuilder();
        //本次读取到的字节个数
        int curBufferSize;
        //循环将数据写入缓冲区buffer,并返回读取到的字节个数。当前数据读取完毕会返回-1。
        // curBufferSize这个参数的作用有两个
        //  1. 判断是否读取到了流的末尾(==-1?)
        //  2. 缓冲区字节数组buffer可能并没有写满,只写了curBufferSize,
        //  那么我们只需要将字节数组中前面curBufferSize个字节转换成字符串就行。
        while ((curBufferSize = bufferedInputStream.read(buffer)) != -1) {
            //将buffer中的数据转换成字符串,从buffer的第0个字节开始,读取curBufferSize个字节
            fullMessage.append(new String(buffer, 0, curBufferSize));
        }
        return fullMessage.toString();
    }
}

1. echo聊天室

  • 需求描述:

  • 服务器接收到客户端发送的消息,并打印
  • 服务端将客户端发送的消息经过包装后再次发送给客户端
  • 客户端断开连接
  • eg:
    • client: greet from socket.
    • server: echo from server: <greet from socket.>
  • 实现思路:
    • 创建服务端ServerSocket并绑定所监听的端口
    • 调用serverSocket.accept()阻塞监听客户端的接入
    • 客户端接入后获取到客户端Socket,并将该Socket上流的读写操作交给子线程去处理,主线程继续阻塞在accept()监听客户端的接入,否则同一时刻只能有一个客户端接入。
  • 服务器端 BioChatServer
代码语言:javascript
复制
/**
 * @author futao
 * @date 2020/7/2
 */
public class BioChatServer {

    private static final Logger logger = LoggerFactory.getLogger(BioChatServer.class);

    /**
     * 启动服务器
     */
    public void start() {
        //创建服务端ServerSocket,并监听端口
        try (ServerSocket serverSocket = new ServerSocket(Constants.SERVER_PORT)) {
            logger.debug("========== 基于BIO的聊天室在[{}]端口启动成功 ==========", Constants.SERVER_PORT);
            //循环accept()监听
            while (true) {
                //accept()将阻塞,直到有客户端Socket接入。并在服务端创建一个Socket与其对应
                Socket socket = serverSocket.accept();
                logger.debug("客户端[{}]成功接入", socket.getPort());
                //将获取到的客户端连接交给子线程去处理,不影响主线程继续监听,等待下一个客户端连接
                new Thread(() -> {
                    try (
                            //用于从客户端读取数据
                            InputStream inputStream = socket.getInputStream();
                            //用于将数据写给客户端
                            OutputStream outputStream = socket.getOutputStream()
                    ) {
                        //从输入流中读取数据
                        String fullMessage = IOUtils.readString(inputStream);
                        logger.info("接收到客户端【{}】发来的消息[{}]", socket.getPort(), fullMessage);
                        //编码与响应
                        outputStream.write(String.format("echo from server: <%s>", fullMessage).getBytes(Constants.CHARSET));
                    } catch (IOException e) {
                        logger.error("客户端异常", e);
                    }
                }).start();
            }

        } catch (IOException e) {
            logger.error("服务器启动失败", e);
            return;
        }
    }

    public static void main(String[] args) {
        new BioChatServer().start();
    }
}
  • 客户端 BioChatClient
代码语言:javascript
复制
/**
 * @author futao
 * @date 2020/7/2
 */
public class BioChatClient {

    private static final Logger logger = LoggerFactory.getLogger(BioChatClient.class);

    /**
     * 启动客户端
     */
    private void start() {
        try (
                //尝试连接到服务器
                Socket socket = new Socket("localhost", Constants.SERVER_PORT);
                //获取到输入流
                InputStream inputStream = socket.getInputStream();
                //输出流
                OutputStream outputStream = socket.getOutputStream()
        ) {
            logger.debug("========== 成功连接到聊天服务器 ==========");
            //获取到用户输入的字符串
            String userInputStr = new Scanner(System.in).nextLine();
            //将字符串转换成字节,写入输出流
            outputStream.write(userInputStr.getBytes(Constants.CHARSET));
            //刷新缓冲区
            outputStream.flush();
            //【重要】关闭输出流,通知服务器客户端消息已经发送完毕
            socket.shutdownOutput();
            //读取服务端的响应
            logger.info("接收到消息:[{}]", IOUtils.readString(inputStream));
        } catch (UnknownHostException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        new BioChatClient().start();
    }
}
  • 需要注意的代码为BioChatClient.start()socket.shutdownOutput();,如果不将客户端socket对应的输出流关闭,服务器端将不知道客户端消息是否发送完毕,服务端会一直阻塞在inputstream.read()
  • 测试
    • 启动服务端
    • 分别启动两个客户端,向服务端发送消息
  • 两个客户端分别发送了一条消息,并接收到了服务器的响应。

image.png

  • 通过服务器端日志可以看出
    • 客户端接入事件都是在主线程main线程上发生的。
    • 而客户端消息的收发都是在新的子线程上发生的。而且每一个连接都需要一个全新的线程来处理。
  • 通过对线程运行状态的分析也可以看出,子线程在完成消息读取和发送之后立马就销毁了

image.png

  • 缺点:
    • 客户端每次接入只能发送一条消息就下线了,无法保持长期在线。
    • 每有一个客户端接入,就需要创建一个线程,如果有大量客户端接入,将对服务器产生较大压力。
    • 且每个创建的线程只执行了非常少量的任务就被销毁了,对资源的消耗比较大。

2. 群聊聊天室

  • 服务器接收客户端发送的消息,并打印
  • 服务端将客户端发送的消息转发给其他在线的客户端
  • 客户端可一直保持在线状态
2.1 常规思路
  • 服务端

image.png

  • 客户端

image.png

  • 但是测试下来会发现,如果客户端不断开连接,服务端将一直阻塞在Inputstream.read(),因为服务端根本不知道客户端的数据已经发送完毕了。
  • 所以现在的问题是:如何告知对方数据已经发送完毕?

# 如何告知对方数据已经发送完毕?

客户端打开一个输出流,如果不做约定,也不关闭它,那么服务端永远不知道客户端是否发送完消息,那么服务端会一直等待下去,直到读取超时。

  1. 关闭Socket连接。
    • 缺点: 客户端Socket关闭后,将不能接受服务端发送的消息,也不能再次发送消息。
  2. 关闭Socket的输出流(而不是Socket)
    • 关闭输出流socket.shutdownOutput();而不是outputStream.close();,这样还能继续监听从服务端响应的输入流。
    • 参照Echo聊天室的实现
    • 缺点:还是不能再次发送消息给服务端。
  3. 通过约定的符号
    • 先在输出流的第一个字节写入本次传输将会传递的数据的字节大小,接收方在获取到这个值之后,从输入流中读取指定个数的字节即可。
    • Reader.readLine()将会在读取到回车\r,换行\n或者回车紧跟着换行\r\n时返回读取到的数据。
    1. 通过Writer.writeLine()/Reader.readLine()
    2. 通过指定长度告知对方已发送完命令
2.2 通过约定的符号\r\n,标识消息发送完毕。
  • 服务端代码
代码语言:javascript
复制

/**
 * 使用标记符号的方式通知消息发送完毕
 *
 * @author futao
 * @date 2020/7/2
 */
public class BioChatServer {

    private static final Logger logger = LoggerFactory.getLogger(BioChatServer.class);

    private static final Set<Socket> CLIENT_SOCKET_SET = new HashSet<Socket>() {
        @Override
        public synchronized boolean add(Socket o) {
            return super.add(o);
        }

        @Override
        public synchronized boolean remove(Object o) {
            return super.remove(o);
        }
    };


    public void start() {
        //创建服务端ServerSocket,并监听端口
        try (ServerSocket serverSocket = new ServerSocket(Constants.SERVER_PORT)) {
            logger.debug("========== 基于BIO的聊天室在[{}]端口启动成功 ==========", Constants.SERVER_PORT);
            //循环accept()监听
            while (true) {
                //accept()将阻塞,直到有客户端Socket接入。并在服务端创建一个Socket与其对应
                Socket socket = serverSocket.accept();
                logger.debug("客户端[{}]上线", socket.getPort());
                CLIENT_SOCKET_SET.add(socket);
                //将获取到的客户端连接交给子线程去处理,不影响主线程继续监听,等待下一个客户端连接
                new Thread(() -> {
                    try {
                        //用于从客户端读取数据(将字节流转换成字符流)
                        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(new BufferedInputStream(socket.getInputStream())));
                        while (true) {
                            //从输入流中读取数据
                            String message = bufferedReader.readLine();
                            if (StringUtils.isNotBlank(message)) {
                                logger.info("接收到客户端【{}】发来的消息[{}]", socket.getPort(), message);
                            } else {
                                isQuit(message, socket);
                            }
                            //判断是否为下线
                            boolean isQuit = isQuit(message, socket);
                            //转发消息
                            if (!isQuit) {
                                forwardMessage(socket.getPort(), String.format("<from %s>", socket.getPort()) + message);
                            } else {
                                break;
                            }
                        }
                    } catch (IOException e) {
                        logger.error("客户端异常", e);
                    }
                }).start();
            }

        } catch (IOException e) {
            logger.error("服务器启动失败", e);
            return;
        }
    }


    public boolean isQuit(String message, Socket socket) throws IOException {
        boolean isQuit = Constants.KEY_WORD_QUIT.equals(message);
        if (isQuit) {
            CLIENT_SOCKET_SET.remove(socket);
            int port = socket.getPort();
            socket.close();
            logger.debug("客户端[{}]下线", port);
        }
        return isQuit;
    }

    /**
     * 转发消息
     *
     * @param curSocketPort 当前发送消息的客户端Socket的端口
     * @param message       需要转发的消息
     */
    public void forwardMessage(int curSocketPort, String message) {
        message += "\r\n";
        if (StringUtils.isBlank(message)) {
            return;
        }
        for (Socket socket : CLIENT_SOCKET_SET) {
            if (socket.isClosed() || socket.getPort() == curSocketPort) {
                continue;
            }
            if (socket.getPort() != curSocketPort) {
                try {
                    OutputStream outputStream = socket.getOutputStream();
                    //将字符串编码之后写入客户端
                    outputStream.write(message.getBytes(Constants.CHARSET));
                    //刷新缓冲区
                    outputStream.flush();
                } catch (IOException e) {
                    logger.error("消息转发失败", e);
                }
            }
        }
    }

    public static void main(String[] args) {
        new BioChatServer().start();
    }
}
  • 客户端
代码语言:javascript
复制

/**
 * @author futao
 * @date 2020/7/2
 */
public class BioChatClient {

    private static final Logger logger = LoggerFactory.getLogger(BioChatClient.class);
    /**
     * 开启这个线程的目的是,当用户输入了退出指令,需要通知监听响应的线程也结束,
     * 否则如果监听响应的线程还处于阻塞状态的话,客户端应用是无法停止的
     */
    private static final ExecutorService executorService = Executors.newSingleThreadExecutor();

    private void start() {
        try {
            Socket socket = new Socket("localhost", Constants.SERVER_PORT);
            BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), Constants.CHARSET));
            OutputStream outputStream = socket.getOutputStream();

            logger.debug("========== 成功连接到聊天服务器 ==========");
            new Thread(() -> {
                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in, Constants.CHARSET));
                while (true) {
                    try {
                        String userInputStr = bufferedReader.readLine();
                        //需要加上换行符
                        outputStream.write((userInputStr + "\n").getBytes(Constants.CHARSET));
                        outputStream.flush();
                        if (Constants.KEY_WORD_QUIT.equals(userInputStr)) {
                            reader.close();
                            outputStream.close();
                            socket.close();
                            //通知监听响应的线程也结束
                            executorService.shutdownNow();
                            break;
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                logger.debug("========== 退出聊天 ==========");
            }).start();


            executorService.execute(() -> {
                //线程一直监听服务端发送的消息
                String message;
                try {
                    while (!socket.isInputShutdown() && (message = reader.readLine()) != null) {
                        logger.info("接收到消息:[{}]", message);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        } catch (UnknownHostException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        new BioChatClient().start();
    }
}

  • 测试
    • 启动服务端与客户端(1个服务端,3个客户端)
  • 客户端发送消息与转发
  • 马化腾下线

image.png

image.png

  • 测试通过,达成目标~

# 对线程的考虑

  • 从前面对BIO的实现可以看出,需要大量用到线程,现在来测试一下关于线程的问题。

启动50个客户端,观察服务器端线程的情况。

  • 测试代码
代码语言:javascript
复制
/**
 * @author futao
 * @date 2020/7/5
 */
public class ClientRunner {

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(50);
        for (int i = 0; i < 50; i++) {
            executorService.execute(() -> {
                new BioChatClient().start();
            });
        }
    }
}
  • 日志

image.png

  • 线程

服务端有50个线程与客户端对应。

# 伪异步IO

服务端在accept()与客户端建立Socket连接之后,将该任务交给线程池去处理,而不是每次都开启一个新的线程。

  • 改动服务端代码的两行代码

image.png

  • 再次测试

image.png

  • 服务端的线程数维持在了10个,保护了服务端的安全~

# 参考

  • 【Socket】Java Socket编程基础及深入讲解
  • Java 网络编程基础 BIO、NIO、AIO 总结

欢迎在评论区留下你看文章时的思考,及时说出,有助于加深记忆和理解,还能和像你一样也喜欢这个话题的读者相遇~

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2020-07-05,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 喜欢天文 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • # 提要:
  • # 基础配置与工具类
  • 1. echo聊天室
  • 2. 群聊聊天室
    • 2.1 常规思路
    • # 如何告知对方数据已经发送完毕?
      • 2.2 通过约定的符号\r\n,标识消息发送完毕。
      • # 对线程的考虑
        • # 伪异步IO
        • # 参考
        领券
        问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档