Java NIO（六）：实现多人聊天室

Love The Way You Lie 2021-09-10 04:24 382阅读 0赞

> Demo 下载：[https://gitee.com/LinYuanTongXue/Demo-NIO-CharRoom][https_gitee.com_LinYuanTongXue_Demo-NIO-CharRoom]

### 服务端思路： ###

*  定义一个 Charset 字符集用于解析数据
 *  定义两个 ByteBuffered 缓冲区用于收发数据
 *  定义一个 Map<String,Channel> 用于存放客户端集合
 *  定义一个 Selector 用于监听通道事件
 *  通过 ServerSocketChannel 的 open 方法打开一个 ServerSocketChannel 通道
 *  设置为非阻塞模式，并传递一个 InetSocketAddress 绑定端口
 *  注册 ServerSocketChannel 的接收就绪事件通道到 Selector
 *  循环监听事件
    
     *  调用 Selector 的 select 方法阻塞直到有准备就绪的通道
     *  调用 Selector 的 selectedKeys 获取准备就绪的通道集合
     *  针对事件处理通道
        
         *  **接收就绪**：
            
             *  代表有客户端要连接，通过 SelectionKey 对象获取 ServerSocketChannel
             *  调用 accept 方法接收请求客户端通道 SocketChannel
             *  为 SocketChannel 通道注册可读数据事件到 Selector
             *  将客户端 SocketChannel 添加到客户端集合 map 中
         *  **可读就绪**:
            
             *  代表客户端向服务器端发送了消息，通过 SelectionKey 对象获取 SocketChannel
             *  调用 read 方法将数据读入通道，通过 Charset 的 decode 将数据解码
             *  调用 dispatch 方法将消息转发给各个客户端
                
                 *  dispach方法：遍历客户端集合 map，通过 Channel 的 write 方法将消息发送给各个客户端
 *  处理完事件后要清空 SelectionKey 集合，当下次该通道变成就绪时，Selector 才会再次将其放入 SelectionKey 中

public class NIOServer {
    
        private int port = 8888;
    
        // 用于字符集编解码
        private Charset charset = Charset.forName("UTF-8");
    
        // 用于接收数据的缓冲区
        private ByteBuffer rBuffer = ByteBuffer.allocate(1024);
    
        // 用于发送数据的缓冲区
        private ByteBuffer sBuffer = ByteBuffer.allocate(1024);
    
        // 用于存放客户端SocketChannel集合
        private Map<String, SocketChannel> clientMap = new HashMap();
    
        // 用于监听通道事件
        private static Selector selector;
    
        public NIOServer(int port) {
            this.port = port;
            try {
                init();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        // 初始化服务器
        private void init() throws IOException {
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            ServerSocket serverSocket = serverSocketChannel.socket();
            serverSocket.bind(new InetSocketAddress(port));
            selector = Selector.open();
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("服务器启动，端口为：" + port);
        }
    
        /**
         * 服务器端轮询监听，select 方法会一直阻塞直到有相关事件发生或超时
         */
        public void listen() {
            while (true) {
                try {
                    selector.select();   // 返回值为本次触发的事件数
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    selectionKeys.forEach(selectionKey -> handle(selectionKey));
                    selectionKeys.clear(); // 清除处理过的事件
                } catch (IOException e) {
                    e.printStackTrace();
                }
    
            }
        }
    
        /**
         * 处理事件
         */
        private void handle(SelectionKey selectionKey) {
            try {
                // 有客户端要连接
                if (selectionKey.isAcceptable()) {
                    ServerSocketChannel server = (ServerSocketChannel) selectionKey.channel();
                    SocketChannel client = server.accept();
                    client.configureBlocking(false);
                    client.register(selector, SelectionKey.OP_READ);
                    clientMap.put(getClientName(client), client);
                }
                // 客户端发送了消息
                else if (selectionKey.isReadable()) {
                    SocketChannel client = (SocketChannel) selectionKey.channel();
                    rBuffer.clear();
                    int bytes = client.read(rBuffer);
                    if (bytes > 0) {
                        rBuffer.flip();
                        String receiveText = String.valueOf(charset.decode(rBuffer));
                        System.out.println(client.toString() + ":" + receiveText);
                        dispatch(client, receiveText);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        /**
         * 转发消息给各个客户端
         */
        private void dispatch(SocketChannel client, String info) throws IOException {
            if (!clientMap.isEmpty()) {
                for (Map.Entry<String, SocketChannel> entry : clientMap.entrySet()) {
                    SocketChannel temp = entry.getValue();
                    if (!client.equals(temp)) {
                        sBuffer.clear();
                        sBuffer.put(charset.encode(getClientName(client) + ":" + info));
                        sBuffer.flip();
                        temp.write(sBuffer);
                    }
                }
            }
        }
    
        /**
         * 生成客户端名字
         */
        private String getClientName(SocketChannel client){
            Socket socket = client.socket();
            return "[" + socket.getInetAddress().toString().substring(1) + ":" + Integer.toHexString(client.hashCode()) + "]";
        }
    
        public static void main(String[] args) {
            NIOServer server = new NIOServer(7777);
            server.listen();
        }
    }

### 客户端思路： ###

*  定义一个 Charset 字符集用于解析数据
 *  定义两个 ByteBuffered 缓冲区用于收发数据
 *  定义一个 Selector 用于监听通道事件
 *  通过 SocketChannel 的 open() 打开一个 SocketChannel 实例
 *  设置 SocketChannel 实例为非阻塞模式，并调用 connect() 连接到服务端
 *  注册 SocketChannel 的连接就绪事件到 Selector
 *  循环监听事件
    
     *  调用 Selector 的 select 方法阻塞直到有准备就绪的通道
     *  调用 Selector 的 selectedKeys 获取准备就绪的通道集合
     *  针对事件处理通道
        
         *  **连接就绪**：
            
             *  调用 SocketChannel 的 isConnectionPending 判断否正在连接服务器端，如果是则调用 finishConnect 完成连接
             *  新建一个线程，通过 Scanner 接收用户输入并通过 SocketChannel 的 write 写入到通道
             *  设置监听 SocketChannel 的可读数据事件
         *  **可读就绪**：
            
             *  有从服务器端发送过来的信息，读取输出到屏幕上
 *  处理完事件后要清空 SelectionKey 集合，当下次该通道变成就绪时，Selector 才会再次将其放入 SelectionKey 中

public class NIOClient {
    
        // 服务端地址
        private InetSocketAddress SERVER;
    
        // 用于接收数据的缓冲区
        private ByteBuffer rBuffer = ByteBuffer.allocate(1024);
    
        // 用于发送数据的缓冲区
        private ByteBuffer sBuffer = ByteBuffer.allocate(1024);
    
        // 用于监听通道事件
        private static Selector selector;
    
        // 用于编/解码 buffer
        private Charset charset = Charset.forName("UTF-8");
    
        public NIOClient(int port) {
            SERVER = new InetSocketAddress("localhost", port);
            try {
                init();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        // 初始化客户端
        private void init() throws IOException {
            SocketChannel socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);
            selector = Selector.open();
            socketChannel.register(selector, SelectionKey.OP_CONNECT);
            socketChannel.connect(SERVER);
            while (true) {
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                selectionKeys.forEach(selectionKey -> handle(selectionKey));
                selectionKeys.clear(); // 清除处理过的事件
            }
        }
    
        /**
         * 处理事件
         */
        private void handle(SelectionKey selectionKey) {
            try {
                // 连接就绪事件
                if (selectionKey.isConnectable()) {
                    SocketChannel client = (SocketChannel) selectionKey.channel();
                    if (client.isConnectionPending()) {
                        client.finishConnect();
                        System.out.println("连接成功！");
                        // 启动线程监听客户端输入
                        new Thread() {
                            @Override
                            public void run() {
                                while (true) {
                                    try {
                                        sBuffer.clear();
                                        Scanner scanner = new Scanner(System.in);
                                        String sendText = scanner.nextLine();
                                        System.out.println(sendText);
                                        sBuffer.put(charset.encode(sendText));
                                        sBuffer.flip();
                                        client.write(sBuffer);
                                    } catch (IOException e) {
                                        e.printStackTrace();
                                    }
                                }
                            }
                        }.start();
                    }
                    // 注册可读事件
                    client.register(selector, SelectionKey.OP_READ);
                }
                // 可读事件，有从服务器端发送过来的信息，读取输出到屏幕上
                else if (selectionKey.isReadable()) {
                    SocketChannel client = (SocketChannel) selectionKey.channel();
                    rBuffer.clear();
                    int count = client.read(rBuffer);
                    if (count > 0) {
                        String receiveText = new String(rBuffer.array(), 0, count);
                        System.out.println(receiveText);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
    
        public static void main(String[] args) {
            new NIOClient(7777);
        }
    }

[https_gitee.com_LinYuanTongXue_Demo-NIO-CharRoom]: https://gitee.com/LinYuanTongXue/Demo-NIO-CharRoom