IO、NIO实现简单聊天室，附带问题解析

2020 年 7 月 21 日
笔记
JAVA

本篇文章主要使用IO和NIO的形式来实现一个简单的聊天室，并且说明IO方法存在的问题，而NIO又是如何解决的。

大概的框架为，先提供思路和大概框架图——代码——问题及解决方式，这样会容易看一点。

1. IO写法

1.1 思路框架

下面编写一个简单的聊天室，大概需要的功能就是服务端维护一个聊天室，里边的客户端发送消息之后服务将其消息转发给其他客户端，达到一个聊天室的效果。

大致的思路：服务端区分职责，分成两部分，主线程负责接收连接并把连接放入到线程池中处理，维护一个线程池，所有对于socket的处理都交给线程池中的线程来处理。如下图。

下面贴上demo代码（代码中有几处为了方便并没有采用最规范的定义方式，如线程池的创建和Map初始化的时候未设置初始容量等）

代码分五个类，服务端（ChatServer，监听作用，为服务端主线程）、客户端（ChatClient）、服务端处理器（ServerHandler，可以理解为线程池中要执行的事情）、客户端处理器（ClientHandler，客户端读写服务器消息的处理），工具类（SocketUtils，只有一个发送消息方法）。

1.2 demo代码

服务端：

/**
 * 服务端启动类
 * 主要负责监听客户端连接
 */
public class ChatServer {

    public static void main(String[] args) {
        ServerSocket serverSocket = null;
        /*----------为了方便使用Executors创建线程-------------*/
        ExecutorService handlerThreadPool = Executors.newFixedThreadPool(100);
        try {
            serverSocket = new ServerSocket(8888);
            while (true) {
                System.out.println("-----------阻塞等待连接------------");
                Socket socket = serverSocket.accept();
                String key = socket.getInetAddress().getHostAddress() + ":" + socket.getPort();
                System.err.println(key + "已连接");
                // 主线程只接收，处理直接交给处理线程池
                handlerThreadPool.execute(new ServerHandler(socket));
            }
        } catch (IOException e) {
            e.printStackTrace();
            if (Objects.nonNull(serverSocket)) {
                try {
                    serverSocket.close();
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
            }
        }
    }

}

服务端处理类：


/**
 * 服务端socket事件处理类
 * 负责处理对应socket中的读写操作
 */
public class ServerHandler implements Runnable {

    /**
     * 连接到服务端的所有连接 socket的地址端口->socket
     */
    private static final Map<String, Socket> socketMap = new ConcurrentHashMap<>();

    /**
     * 维护名称和地址的map
     */
    private static final Map<String, String> nameMap = new ConcurrentHashMap<>();

    private Socket socket;

    /**
     * 每个socket的标识，使用地址+端口构成
     */
    private String key;

    public ServerHandler() {
    }

    public ServerHandler(Socket socket) {
        this.socket = socket;
        this.key = socket.getInetAddress().getHostAddress() + ":" + socket.getPort();
    }

    @Override
    public void run() {
        Socket s = socket;
        // 根据消息执行不同操作
        InputStream inputStream;
        // debug查看数据用
        // Map<String, Socket> tmpMap = socketMap;
        try {
            inputStream = s.getInputStream();
            Scanner scanner = new Scanner(inputStream);
            while (true) {
                String line = scanner.nextLine();
                if (line.startsWith("register")) {
                    // 登记
                    String[] split = line.split(":");
                    String name = split[1];
                    String msg;
                    // 校验是否存在
                    if (socketMap.containsKey(key)) {
                        msg = "请勿重复登记";
                        sendMsg(s, msg);
                        return;
                    }

                    if (nameMap.containsValue(name)) {
                        msg = "名称已被登记，请换一个名称";
                        sendMsg(s, msg);
                        return;
                    }

                    // 通知自己已连接
                    sendMsg(s, "已连接到服务器");

                    msg = name + "进入聊天室";
                    // 将消息转发给其他客户端
                    sendMsgToClients(msg);

                    // 放入socket池
                    socketMap.put(key, s);
                    nameMap.put(key, name);
                    System.err.println(name + "已登记");
                } else if (line.trim().equalsIgnoreCase("end")) {
                    if (notPassRegisterValidate()) {
                        continue;
                    }

                    // 断开连接
                    socketMap.remove(key);
                    String name = nameMap.get(key);
                    String msg = name + "离开聊天室";
                    System.err.println(msg);
                    // 将消息转发给其他客户端
                    sendMsgToClients(msg);

                    msg = "已断开连接";
                    // 发送给对应的连接断开信息
                    sendMsg(s, msg);
                    inputStream.close();
                    break;
                } else {
                    if (notPassRegisterValidate()) {
                        continue;
                    }

                    // 正常通信
                    String name = nameMap.get(key);
                    String msg = name + ":" + line;
                    // 将消息转发给其他客户端
                    sendMsgToClients(msg);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 是否已登录校验
     *
     * @return 是否已登录
     */
    private boolean notPassRegisterValidate() {
        boolean hasRegister = nameMap.containsKey(key);
        if (hasRegister) {
            return false;
        }

        String msg = "您还未登录，请先登录";
        sendMsg(socket, msg);
        return true;
    }

    /**
     * 往连接发送消息
     *
     * @param socket 客户端连接
     * @param msg    消息
     */
    private void sendMsg(Socket socket, String msg) {
        SocketUtils.sendMsg(socket, msg);
        if (socket.isClosed()) {
            socketMap.remove(key);
        }
    }

    /**
     * 发送给其他客户端信息
     *
     * @param msg 信息
     */
    private void sendMsgToClients(String msg) {
        for (Map.Entry<String, Socket> entry : socketMap.entrySet()) {
            if (this.key.equals(entry.getKey())) {
                continue;
            }

            sendMsg(entry.getValue(), msg);
        }
    }

}

工具类（一个发送消息的方法）：

public class SocketUtils {

    private SocketUtils() {
    }

    public static void sendMsg(Socket socket, String msg) {
        Socket s = socket;
        OutputStream outputStream = null;
        msg += "\r\n";
        try {
            outputStream = s.getOutputStream();
            outputStream.write(msg.getBytes(StandardCharsets.UTF_8));
            outputStream.flush();
        } catch (IOException e) {
            System.err.println("发送消息失败, 连接已断开");
            try {
                if (Objects.nonNull(outputStream)) {
                    outputStream.close();
                }
                socket.close();
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }

        }
    }

}

客户端：

/**
 * 客户端读和写各自使用一个线程
 */
public class ChatClient {

    public static void main(String[] args) {
        Socket socket;
        ExecutorService clientHandlerPool = Executors.newFixedThreadPool(2);
        try {
            socket = new Socket("localhost", 8888);

            // 写线程
            clientHandlerPool.execute(new ClientHandler(socket, 1));
            // 读线程
            clientHandlerPool.execute(new ClientHandler(socket, 0));

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}

客户端处理器：

/**
 * 客户端处理器
 * 根据type来区分是做读工作还是写工作
 */
public class ClientHandler implements Runnable {

    private Socket socket;

    /**
     * 处理类型，0-读、1-写
     */
    private int type;

    public ClientHandler() {
        throw new IllegalArgumentException("不能使用没有参数的构造函数");
    }

    public ClientHandler(Socket socket, int type) {
        this.socket = socket;
        this.type = type;
    }

    @Override
    public void run() {
        if (type == 1) {
            // 进行写操作
            doWriteJob();
            return;
        }

        // 默认读操作
        doReadJob();
    }

    /**
     * 读操作
     */
    private void doReadJob() {
        Socket s = socket;
        InputStream inputStream;
        try {
            inputStream = s.getInputStream();
            Scanner scanner = new Scanner(inputStream);
            while (true) {
                String line = scanner.nextLine();
                if (null != line && !"".equals(line)) {
                    System.err.println(line);
                }
                // 如果已退出了，那么关闭连接
                if ("已断开连接".equals(line)) {
                    socket.close();
                    break;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
            try {
                socket.close();
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }

    /**
     * 写线程
     */
    private void doWriteJob() {
        Socket s = socket;
        try {
            Scanner scanner = new Scanner(System.in);
            while (true) {
                String output = scanner.nextLine();
                if (Objects.nonNull(output) && !"".equals(output)) {
                    SocketUtils.sendMsg(s, output);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.err.println("错误发生了:" + e.getMessage());
        }
    }
}

结果：

思考：当前这样实现有什么瓶颈，可能会出现什么问题？

存在问题：

服务端使用accept阻塞接收线程，连接一个一个处理，在高并发下处理性能缓慢。

没有连接的时候线程一直处于阻塞状态造成资源的浪费（如果使用多线程接收处理并发，那么没连接的时候造成多个线程的资源浪费）。

2. 使用NIO实现聊天室

2.1 整体思路

那我们来看下NIO是怎么解决上方的问题的，首先上这个demo整体的架构图。

大概的逻辑为

服务端将ServerSocketChannel注册到Selector中，客户端连接进来的时候事件触发，将客户端的连接注册到selector中。

主线程负责selector的轮询工作，发现有事件可以处理就将其交给线程池。

客户端同理分成两个部分，写操作和读操作，每个操作由一个线程单独完成；但是如果读操作处理使用while循环不断轮询等待接收的话，CPU会飙升，所以需要客户端新建一个selector来解决这个问题，注意这个selector跟服务端不是同一个，没有啥关系。

代码分类大致跟IO写法一样，分成服务端、服务端处理器、客户端、客户端处理器，下面为demo。

2.2 代码

服务端：

public class ChatServer {

    private Selector selector;

    private ServerSocketChannel serverSocketChannel;

    private static final ExecutorService handlerPool = Executors.newFixedThreadPool(100);

    public ChatServer() throws IOException {
        this.selector = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        ServerSocket serverSocket = serverSocketChannel.socket();
        serverSocket.bind(new InetSocketAddress(9999));
        // 将服务端的socket注册到selector中，接收客户端，并将其注册到selector中，其本身也是selector中的一个I/O事件
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        System.err.println("聊天室服务端初始化结束");
    }

    /**
     * 启动方法
     * 1.监听，拿到之后进行处理
     */
    public void start() throws IOException {
        int count;
        while (true) {
            // 可能出现select方法没阻塞，空轮询导致死循环的情况
            count = selector.select();

            if (count > 0) {
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    // 交给线程池处理
                    handlerPool.execute(new ServerHandler(key, selector));
                    // 处理完成后移除
                    iterator.remove();
                }
            }
        }
    }

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

服务端处理器：


public class ServerHandler implements Runnable {

    private SelectionKey key;

    private Selector selector;

    public ServerHandler() {

    }

    /**
     * 本来可以通过key拿到selector，这里为了图方便就这样写了
     */
    public ServerHandler(SelectionKey key, Selector selector) {
        this.key = key;
        this.selector = selector;
    }

    @Override
    public void run() {
        try {
            if (key.isAcceptable()) {
                // 说明是服务端的事件，注意这里强转换为的是ServerSocketChannel
                ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                // 接收连接
                SocketChannel socket = channel.accept();
                if (Objects.isNull(socket)) {
                    return;
                }

                socket.configureBlocking(false);
                // 接收客户端的socket并且将其注册到服务端这边的selector中，注意客户端在此时跟服务端selector产生关联
                socket.register(selector, SelectionKey.OP_READ);
                System.err.println("服务端已接收连接");
            } else if (key.isReadable()) {
                // 客户端发送信息过来了
                doReadJob();
            }
        } catch (IOException e) {
            e.printStackTrace();
            // 错误处理
        }
    }

    /**
     * 读取操作
     */
    private void doReadJob() throws IOException {
        SocketChannel socketChannel = (SocketChannel) key.channel();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        int readCount = socketChannel.read(buffer);
        if (readCount > 0) {
            String msg = new String(buffer.array(), StandardCharsets.UTF_8);
            System.err.println(socketChannel.getRemoteAddress().toString() + "的信息为:" + msg);

            // 转发给其他客户端
            sendMsgToOtherClients(msg);
        }
    }

    /**
     * 转发消息给其他客户端
     *
     * @param msg 消息
     */
    private void sendMsgToOtherClients(String msg) throws IOException {

        SocketChannel self = (SocketChannel) key.channel();

        Set<SelectionKey> keys = selector.keys();
        Iterator<SelectionKey> iterator = keys.iterator();
        while (iterator.hasNext()) {
            SelectionKey selectionKey = iterator.next();
            SelectableChannel channel = selectionKey.channel();
            // 如果是本身或者不是socketChannel类型则跳过
            if (self.equals(channel) || channel instanceof ServerSocketChannel) {
                continue;
            }

            SocketChannel socketChannel = (SocketChannel) channel;
            ByteBuffer byteBuffer = ByteBuffer.wrap(msg.getBytes(StandardCharsets.UTF_8));
            socketChannel.write(byteBuffer);
        }
    }
}

客户端：


public class ChatClient {

    private Selector selector;

    private SocketChannel socketChannel;

    private static ExecutorService dealPool = Executors.newFixedThreadPool(2);

    public ChatClient() throws IOException {

        /*
         * 说明一下：
         * 客户端这边的selector跟刚才在服务端定义的selector是不同的两个selector
         * 客户端这边不需要selector也能实现功能，但是读取的时候必须不断的循环，会导致CPU飙升，
         * 所以使用selector是为了解决这个问题的，别跟服务端的selector搞混就好
         */
        selector = Selector.open();
        socketChannel = SocketChannel.open(new InetSocketAddress("localhost", 9999));
        socketChannel.configureBlocking(false);
        socketChannel.register(selector, SelectionKey.OP_READ);
    }


    public void start() throws IOException, InterruptedException {
        // 连接
//        socketChannel.connect(new InetSocketAddress("localhost", 9999));
        while (!socketChannel.finishConnect()) {
            System.err.println("正在连接...");
            TimeUnit.MILLISECONDS.sleep(200);
        }

        System.err.println("连接成功");

        // 使用两个线程来分别处理读取和写操作
        // 写数据
        dealPool.execute(new ClientHandler(selector, socketChannel, 1));

        // 读取数据
        dealPool.execute(new ClientHandler(selector, socketChannel, 0));
    }


    public static void main(String[] args) throws IOException, InterruptedException {
        new ChatClient().start();
    }
}

客户端处理器：


public class ClientHandler implements Runnable {

    private Selector selector;

    private SocketChannel socketChannel;

    /**
     * 0-读，1-写
     */
    private int type;

    public ClientHandler() {
    }

    public ClientHandler(Selector selector, SocketChannel socketChannel, int type) {
        // selector是为了解决读时候CPU飙升的问题，具体见客户端的启动类代码注释
        this.selector = selector;
        this.socketChannel = socketChannel;
        this.type = type;
    }

    @Override
    public void run() {
        try {
            if (type == 0) {
                doClientReadJob();
                return;
            }

            doClientWriteJob();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 写操作
     */
    private void doClientWriteJob() throws IOException {
        SocketChannel sc = socketChannel;
        Scanner scanner = new Scanner(System.in);
        while (true) {
            if (scanner.hasNextLine()) {
                String line = scanner.nextLine();
                if (null != line && !"".equals(line)) {
                    ByteBuffer buffer = ByteBuffer.wrap(line.getBytes(StandardCharsets.UTF_8));
                    sc.write(buffer);
                }
            }
        }
    }

    /**
     * 读操作
     */
    private void doClientReadJob() throws IOException {
        SocketChannel sc = socketChannel;
        ByteBuffer buf = ByteBuffer.allocate(1024);
        while (true) {
            int select = selector.select();
            if (select > 0) {
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    // 这是必须的，不然下方的remove会出错
                    SelectionKey next = iterator.next();
                    // 这里因为只有本身这个客户端注册到客户端的selector中，所以有事件一定是它的，也就不用从key拿了，直接操作就行
                    buf.clear();
                    int read = sc.read(buf);
                    if (read > 0) {
                        String msg = new String(buf.array(), StandardCharsets.UTF_8);
                        System.err.println(msg);
                    }
                    // 事件处理完之后要移除这个key，否则的话selector.select()方法不会再读到这个key，即便有新的时间到这个channel来
                    iterator.remove();
                }
            }
        }
    }

}

结果图：

在编写的过程中发现了以下两点：

select方法之后如果存在key，并且接下来的操作未对这个selectionKey做remove操作，那么下次的select不会再将其选入，即便有事件发生，也就是说，select方法不会选择之前已经选过的key。

selector.select()方法中偶尔会出现不阻塞的情况。这就是NIO中的空轮询bug，也就是说，没有连接又不阻塞的话，while(true) … 的写法就是一个死循环，会导致CPU飙升。

第二点问题在NIO框架（如netty）中都采用了比较好的解决方法，可以去查下如何解决的。接下来看下NIO的写法是否解决了IO写法中存在的问题：

服务端使用accept阻塞接收线程，连接一个一个处理，在高并发下处理性能缓慢。

答：上述写法中还是使用一个ServerSocketChannel来接收客户端，没有解决这个问题；但是可以通过使用线程池的方式来解决。也就是说将服务端的事件分成两个部分，第一个部分为接收客户端，使用一个线程池来维护；第二个部分为客户端的事件处理操作，也维护一个线程池来执行这些事件。

这样性能上去了，由于selector的存在也不会出现资源浪费的事情，netty就是这么做的哦。

没有连接的时候线程一直处于阻塞状态造成资源的浪费（如果使用多线程接收处理并发，那么没连接的时候造成多个线程的资源浪费）。

答：解决。NIO写法主要有selector不断轮询，不会出现没连接不作为的情况，而且多个连接的话也没有问题（参考1的回答）。

3. 小结

两种写法都有Reactor模式的影子，但是IO写法有明显的缺点就是如果没有连接会造成资源浪费的问题（采用多个接收连接的话更甚），而NIO中selector轮询机制就很好的解决了无连接时无作为的情况，并且在性能方面可以通过职责分类和线程池来得到改善，所以，NIO，永远滴神。

需要压力，需要努力。

Tags: JAVA

IO、NIO实现简单聊天室，附带问题解析

1. IO写法

1.1 思路框架

1.2 demo代码

2. 使用NIO实现聊天室

2.1 整体思路

2.2 代码

3. 小结

VirMach 便宜 VPS

QNews

IO、NIO实现简单聊天室，附带问题解析

1. IO写法

1.1 思路框架

1.2 demo代码

2. 使用NIO实现聊天室

2.1 整体思路

2.2 代码

3. 小结

分享此文：

Related Posts

【.NET 与树莓派】数模转换

后深度学习时代的一大研究热点？论因果关系及其构建思路

大型Java进阶专题(八) 设计模式之适配器模式、装饰者模式、观察者模式

WPF 有缩放时显示线条的问题

VirMach 便宜 VPS

QNews

热门搜寻