Java NIO 实现多人聊天系统案例

深藏阁楼爱情的钟 2023-05-29 05:02 22阅读 0赞

## 1、NIO简述 ##

Java NIO 主要由三个核心部分组成：

1.      ** Channels**
2.  **     Buffers**
3.  **     Selectors**

另有其它组件，如Pipe和FileLock，与三个核心组件共同使用的工具类。

### Channel 和 Buffer ###

IO 在NIO 中都从一个Channel 开始。Channel 象流。 数据可以从Channel读到Buffer中，也可以从Buffer 写到Channel中。这里有个图示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWhhbzExODY_size_16_color_FFFFFF_t_70][]

Channel和Buffer有好几种类型。下面是JAVA NIO中的一些主要Channel的实现：

1.  **FileChannel**
2.  **DatagramChannel**
3.  **SocketChannel**
4.  **ServerSocketChannel**

正如你所看到的，这些通道涵盖了UDP 和 TCP 网络IO，以及文件IO。

与这些类一起的有一些有趣的接口，但为简单起见，我尽量在概述中不提到它们。本教程其它章节与它们相关的地方我会进行解释。

以下是Java NIO里关键的Buffer实现：

1.  **ByteBuffer**
2.  **CharBuffer**
3.  **DoubleBuffer**
4.  **FloatBuffer**
5.  **IntBuffer**
6.  **LongBuffer**
7.  **ShortBuffer**

这些Buffer覆盖了你能通过IO发送的基本数据类型：byte, short, int, long, float, double 和 char。

Java NIO 还有个 MappedByteBuffer，用于表示内存映射文件， 我也不打算在概述中说明。

## Selector ##

Selector允许单线程处理多个 Channel。如果你的应用打开了多个连接（通道），但每个连接的流量都很低，使用Selector就会很方便。例如，在一个聊天服务器中。

这是在一个单线程中使用一个Selector处理3个Channel的图示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWhhbzExODY_size_16_color_FFFFFF_t_70 1][]

要使用Selector，得向Selector注册Channel，然后调用它的select()方法。这个方法会一直阻塞到某个注册的通道有事件就绪。一旦这个方法返回，线程就可以处理这些事件，事件的例子有如新连接进来，数据接收等。

## 2、NIO代码实现多人聊天系统代码实现 ##

### 2.1、服务端 ###

package com.nio;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.*;
    import java.nio.charset.Charset;
    import java.util.Iterator;
    import java.util.Set;
    
    /**
     * NIO服务器端
     */
    public class NioServer {
    
        /**
         * 主方法
         * @param args
         */
        public static void main(String[] args) throws IOException {
            new NioServer().start();
        }
    
        /**
         * 启动
         */
        public void start() throws IOException {
            /**
             * 1. 创建Selector
             */
            Selector selector = Selector.open();
    
            /**
             * 2. 通过ServerSocketChannel创建channel通道
             */
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    
            /**
             * 3. 为channel通道绑定监听端口
             */
            serverSocketChannel.bind(new InetSocketAddress(8000));
    
            /**
             * 4. **设置channel为非阻塞模式**
             */
            serverSocketChannel.configureBlocking(false);
    
            /**
             * 5. 将channel注册到selector上，监听连接事件
             */
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("服务器启动成功！");
    
            /**
             * 6. 循环等待新接入的连接
             */
            for (;;) { // while(true) c for;;
                /**
                 * TODO 获取可用channel数量
                 */
                int readyChannels = selector.select();
    
                /**
                 * TODO 为什么要这样！！？
                 */
                if (readyChannels == 0) continue;
    
                /**
                 * 获取可用channel的集合
                 */
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
    
                Iterator iterator = selectionKeys.iterator();
    
                while (iterator.hasNext()) {
                    /**
                     * selectionKey实例
                     */
                    SelectionKey selectionKey = (SelectionKey) iterator.next();
    
                    /**
                     * **移除Set中的当前selectionKey**
                     */
                    iterator.remove();
    
                    /**
                     * 7. 根据就绪状态，调用对应方法处理业务逻辑
                     */
                    /**
                     * 如果是 接入事件
                     */
                    if (selectionKey.isAcceptable()) {
                        acceptHandler(serverSocketChannel, selector);
                    }
    
                    /**
                     * 如果是 可读事件
                     */
                    if (selectionKey.isReadable()) {
                        readHandler(selectionKey, selector);
                    }
                }
            }
        }
    
        /**
         * 接入事件处理器
         */
        private void acceptHandler(ServerSocketChannel serverSocketChannel,
                                   Selector selector)
                throws IOException {
            /**
             * 如果要是接入事件，创建socketChannel
             */
            SocketChannel socketChannel = serverSocketChannel.accept();
    
            /**
             * 将socketChannel设置为非阻塞工作模式
             */
            socketChannel.configureBlocking(false);
    
            /**
             * 将channel注册到selector上，监听 可读事件
             */
            socketChannel.register(selector, SelectionKey.OP_READ);
    
            /**
             * 回复客户端提示信息
             */
            socketChannel.write(Charset.forName("UTF-8")
                    .encode("你与聊天室里其他人都不是朋友关系，请注意隐私安全"));
        }
    
        /**
         * 可读事件处理器
         */
        private void readHandler(SelectionKey selectionKey, Selector selector)
                throws IOException {
            /**
             * 要从 selectionKey 中获取到已经就绪的channel
             */
            SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
    
            /**
             * 创建buffer
             */
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    
            /**
             * 循环读取客户端请求信息
             */
            String request = "";
            while (socketChannel.read(byteBuffer) > 0) {
                /**
                 * 切换buffer为读模式
                 */
                byteBuffer.flip();
    
                /**
                 * 读取buffer中的内容
                 */
                request += Charset.forName("UTF-8").decode(byteBuffer);
            }
    
            /**
             * 将channel再次注册到selector上，监听他的可读事件
             */
            socketChannel.register(selector, SelectionKey.OP_READ);
    
            /**
             * 将客户端发送的请求信息 广播给其他客户端
             */
            if (request.length() > 0) {
                // 广播给其他客户端
                broadCast(selector, socketChannel, request);
            }
        }
    
        /**
         * 广播给其他客户端
         */
        private void broadCast(Selector selector,
                               SocketChannel sourceChannel, String request) {
            /**
             * 获取到所有已接入的客户端channel
             */
            Set<SelectionKey> selectionKeySet = selector.keys();
    
            /**
             * 循环向所有channel广播信息
             */
            selectionKeySet.forEach(selectionKey -> {
                Channel targetChannel = selectionKey.channel();
    
                // 剔除发消息的客户端
                if (targetChannel instanceof SocketChannel
                        && targetChannel != sourceChannel) {
                    try {
                        // 将信息发送到targetChannel客户端
                        ((SocketChannel) targetChannel).write(
                                Charset.forName("UTF-8").encode(request));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    
    
    
    }

**说明：NIO SelectionKey中定义的4种事件**

1）SelectionKey.OP\_ACCEPT —— 接收连接继续事件，**表示服务器监听到了客户连接，服务器可以接收这个连接了**

2）SelectionKey.OP\_CONNECT —— 连接就绪事件，**表示客户与服务器的连接已经建立成功**

3）SelectionKey.OP\_READ —— 读**就绪**事件，**表示通道中已经有了可读的数据，可以执行读操作了（通道目前有数据，可以进行读操作了）**

4) SelectionKey.OP\_WRITE —— 写**就绪**事件，**表示已经可以向通道写数据了（通道目前可以用于写操作）**

这里 注意，下面两种，SelectionKey.OP\_READ ，SelectionKey.OP\_WRITE ，

① 当向通道中注册SelectionKey.OP\_READ事件后，如果客户端有向缓存中write数据，下次轮询时，则会 isReadable()=true；

② 当向通道中注册SelectionKey.OP\_WRITE事件后，这时你会发现当前轮询线程中isWritable()一直为ture，如果不设置为其他事件

### 2.2、客户端 ###

package com.nio;
    
    import java.io.IOException;
    
    public class ClientA {
    
        public static void main(String[] args)
                throws IOException {
            new NioClient().start("ClientA");
        }
    
    }

package com.nio;
    
    import java.io.IOException;
    
    public class ClientB {
    
        public static void main(String[] args)
                throws IOException {
            new NioClient().start("ClientB");
        }
    
    }

package com.nio;
    
    import java.io.IOException;
    
    public class ClientC {
    
        public static void main(String[] args)
                throws IOException {
            new NioClient().start("ClientC");
        }
    
    }

3、客户端实现类

package com.nio;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.nio.charset.Charset;
    import java.util.Scanner;
    
    /**
     * NIO客户端
     */
    public class NioClient {
    
        /**
         * 启动
         */
        public void start(String nickname) throws IOException {
            /**
             * 连接服务器端
             */
            SocketChannel socketChannel = SocketChannel.open(
                    new InetSocketAddress("127.0.0.1", 8000));
    
            /**
             * 接收服务器端响应
             */
            // 新开线程，专门负责来接收服务器端的响应数据
            // selector ， socketChannel ， 注册
            Selector selector = Selector.open();
            socketChannel.configureBlocking(false);
            socketChannel.register(selector, SelectionKey.OP_READ);
            new Thread(new NioClientHandler(selector)).start();
    
            /**
             * 向服务器端发送数据
             */
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()) {
                String request = scanner.nextLine();
                if (request != null && request.length() > 0) {
                    socketChannel.write(
                            Charset.forName("UTF-8")
                                    .encode(nickname + " : " + request));
                }
            }
    
        }
    
        public static void main(String[] args) throws IOException {
    //        new NioClient().start();
        }
    
    }

package com.nio;
    
    import java.io.IOException;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.nio.charset.Charset;
    import java.util.Iterator;
    import java.util.Set;
    
    /**
     * 客户端线程类，专门接收服务器端响应信息
     */
    public class NioClientHandler implements Runnable {
        private Selector selector;
    
        public NioClientHandler(Selector selector) {
            this.selector = selector;
        }
    
        @Override
        public void run() {
    
            try {
                for (;;) {
                    int readyChannels = selector.select();
    
                    if (readyChannels == 0) continue;
    
                    /**
                     * 获取可用channel的集合
                     */
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
    
                    Iterator iterator = selectionKeys.iterator();
    
                    while (iterator.hasNext()) {
                        /**
                         * selectionKey实例
                         */
                        SelectionKey selectionKey = (SelectionKey) iterator.next();
    
                        /**
                         * **移除Set中的当前selectionKey**
                         */
                        iterator.remove();
    
                        /**
                         * 7. 根据就绪状态，调用对应方法处理业务逻辑
                         */
    
                        /**
                         * 如果是 可读事件
                         */
                        if (selectionKey.isReadable()) {
                            readHandler(selectionKey, selector);
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        /**
         * 可读事件处理器
         */
        private void readHandler(SelectionKey selectionKey, Selector selector)
                throws IOException {
            /**
             * 要从 selectionKey 中获取到已经就绪的channel
             */
            SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
    
            /**
             * 创建 buffer
             */
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    
            /**
             * 循环读取服务器端响应信息
             */
            String response = "";
            while (socketChannel.read(byteBuffer) > 0) {
                /**
                 * 切换buffer为读模式
                 */
                byteBuffer.flip();
    
                /**
                 * 读取buffer中的内容
                 */
                response += Charset.forName("UTF-8").decode(byteBuffer);
            }
    
            /**
             * 将channel再次注册到selector上，监听他的可读事件
             */
            socketChannel.register(selector, SelectionKey.OP_READ);
    
            /**
             * 将服务器端响应信息打印到本地
             */
            if (response.length() > 0) {
                System.out.println(response);
            }
        }
    }

本文参考：

[https://www.imooc.com/learn/1118][https_www.imooc.com_learn_1118]  （张小喜 老师之 《解锁网络编程之NIO的前世今生》）

[https://www.cnblogs.com/liuxiuhao/p/5785027.html][https_www.cnblogs.com_liuxiuhao_p_5785027.html]

[https://ifeve.com/overview/][https_ifeve.com_overview]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWhhbzExODY_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191104105004718.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWhhbzExODY=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWhhbzExODY_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191104105248878.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ppYWhhbzExODY=,size_16,color_FFFFFF,t_70
[https_www.imooc.com_learn_1118]: https://www.imooc.com/learn/1118
[https_www.cnblogs.com_liuxiuhao_p_5785027.html]: https://www.cnblogs.com/liuxiuhao/p/5785027.html
[https_ifeve.com_overview]: https://ifeve.com/overview/