NIO（一）：从一个简单的NIO示例开始

Bertha 。 2023-06-27 05:42 42阅读 0赞

# 1，NIO和BIO比较 #

*  BIO以流的方法处理数据，而NIO以块的方式处理数据；效率提升很多
 *  BIO是阻塞的，NIO是非阻塞的
 *  BIO基于字节流和字符流进行操作，NIO基于`Buffer`缓冲区和`Channel`通道进行操作，基于`Selector`选择器监听通道中的时间，可以使用单线程监听多个通道数据

# [2，NIO三大核心概念][2_NIO] #

*  缓冲区（Buffer）
 *  通道（Channel）
 *  选择器（Selector）

# 3，NIO核心关系 #

*  每个`Channel`对应一个`Buffer`
 *  `Selector`对应一个线程，一个线程对应多个`Channel`（即连接）
 *  `Selector`进行轮询切换`Channel`，并由事件决定处理逻辑
 *  `Buffer`就是一个内存块，底层由数组构成
 *  数据的读取和写入通过`Buffer`，并通过`Channel`进行`Buffer`数据传递
 *  `Channel`是双向处理的

# 4，NIO执行流程 #

*  初始化服务端通道和选择器，绑定启动端口，并注册通道（`ServerSocketChannel`）到选择器（`Selector`）上，等待客户端连接
 *  客户端连接时，会通过`ServerSocketChannel`获取到一个`SocketChannel`
 *  `Selector`选择器会通过`select()`方法阻塞监听新建连接， 添加到内部`SelectionKey`集合中后，返回监听到的数量
 *  每一次通道注册到选择器后，会包装成一个`SelectionKey`返回，并会添加到内部`SelectionKey`集合中，与`Selector`关联
 *  `Selector`选择器通过`selectionKeys()`方法获取所有注册的事件，可遍历进行处理
 *  通过每一个`SelectionKey`反向获取`channel()`注册的通道`Channel`，并进行后续业务处理

# 5，NIO代码示例 #

*  服务端代码

package com.self.netty.nio.server;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    
    /**
     * 完整NIO服务端
     * 
     * @author Administrator
     *
     */
    public class NIOServer {
    
    	private Selector selector;
    
    	public void init() throws Exception {
    		// 初始化服务端通道, 并绑定端口
    		ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    		// 设置非阻塞
    		serverSocketChannel.configureBlocking(false);
    		serverSocketChannel.bind(new InetSocketAddress(8080));
    		// 初始化选择器
    		selector = Selector.open();
    		// 绑定通道到选择器上, 并初始化为可接收链接
    		serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    		System.out.println("初始化服务端成功, 监听端口: " + 8080);
    		// 开始进行处理
    		start();
    	}
    
    	private void start() throws Exception {
    		// 存在已经注册就绪的事件
    		while (selector.select() > 0) {
    			// 获取就绪的所有事件
    			Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
    			// 遍历事件进行处理
    			while (iterator.hasNext()) {
    				SelectionKey selectionKey = iterator.next();
    				// 处理数据
    				process(selectionKey);
    				iterator.remove();
    			}
    		}
    	}
    
    	private void process(SelectionKey selectionKey) throws IOException {
    		if (selectionKey.isAcceptable()) {
    			// 从服务链接中获取到客户端连接通道, 并注册为可读
    			ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
    			SocketChannel socketChannel = serverSocketChannel.accept();
    			socketChannel.configureBlocking(false);
    			socketChannel.register(selector, SelectionKey.OP_READ);
    			System.out.println("服务端接收到一个客户端链接请求, 并注册为读事件, 准备读取客户端数据数据...");
    		} else if (selectionKey.isReadable()) {
    			SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
    			// 初始化缓冲区
    			ByteBuffer byteBuffer = ByteBuffer.allocate(128);
    			int read = socketChannel.read(byteBuffer);
    			byteBuffer.flip();
    			String content = new String(byteBuffer.array(), 0, read);
    			System.out.println("服务端接收到客户端消息, 消息内容为: " + content);
    			// 携带一个attach, 准备进行返回
    			SelectionKey key = socketChannel.register(selector, SelectionKey.OP_WRITE);
    			key.attach("服务端返回数据: " + content);
    		} else if (selectionKey.isWritable()) {
    			SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
    			String content = (String) selectionKey.attachment();
    			System.out.println("服务端返回数据为: " + content);
    			selectionKey.attach(null);
    			if (null == content) {
    				socketChannel.register(selector, SelectionKey.OP_READ);
    				return;
    			}
    			// 初始化缓冲区
    			ByteBuffer byteBuffer = ByteBuffer.allocate(128);
    			// 写数据到客户端
    			byteBuffer.put(content.getBytes());
    			byteBuffer.flip();
    			socketChannel.write(byteBuffer);
    			byteBuffer.clear();
    			System.out.println("服务端响应客户端数据完成...");
                // 从新注册为读，准备重新读取数据
    			socketChannel.register(selector, SelectionKey.OP_READ);
    		}
    	}
    
    	public static void main(String[] args) throws Exception {
    		new NIOServer().init();
    	}
    
    }

*  客户端代码

# 6，NIO群聊系统实现 #

*  服务端

package com.self.netty.nio.chat;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.*;
    import java.util.Iterator;
    
    /**
     * 群聊服务端
     * 1, 接收客户端连接
     * 2, 对客户端上线和下线进行检测
     * 3, 客户端发送消息后, 转发显示到其他客户端
     *
     * @author pj_zhang
     * @create 2019-12-15 21:59
     **/
    public class GroupChatServer {
    
        /**
         * 选择器
         */
        private Selector selector;
    
        /**
         * 服务端通道
         */
        private ServerSocketChannel serverSocketChannel;
    
        /**
         * 服务端监听端口
         */
        private final int PORT = 8080;
    
        public GroupChatServer() throws Exception {
            // 初始化非阻塞服务端, 并绑定端口
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.bind(new InetSocketAddress(PORT));
            // 初始化选择器
            selector = Selector.open();
            // 注册通道到选择器上, 并初始化为监听
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("服务端启动成功, 监听端口: " + PORT);
        }
    
        public void start() throws Exception {
            while (selector.select() > 0) {
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    // 处理事件
                    process(selectionKey);
                    iterator.remove();
                }
            }
        }
    
        private void process(SelectionKey selectionKey) throws Exception {
            // 初始化链接
            if (selectionKey.isAcceptable()) {
                ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
                // 获取到客户端连接
                SocketChannel socketChannel = serverSocketChannel.accept();
                socketChannel.configureBlocking(false);
                // 注册到选择器, 并注册为可读
                socketChannel.register(selector, SelectionKey.OP_READ);
                // 服务端提示上线
                String message = socketChannel.getRemoteAddress() + " 上线了...";
                System.out.println(message);
                publishMessage(message, socketChannel);
            } else if (selectionKey.isReadable()) {
                SocketChannel socketChannel = null;
                try {
                    // 读取当前客户端发送消息
                    socketChannel = (SocketChannel) selectionKey.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    int readCount = socketChannel.read(buffer);
                    String showMessage = socketChannel.getRemoteAddress() + " 说: " + new String(buffer.array(), 0, readCount);
                    System.out.println(showMessage);
                    // 向其他客户端广播消息
                    publishMessage(showMessage, socketChannel);
                } catch (IOException e) {
                    if (null != socketChannel) {
                        // 读取消息失败, 说明客户端已经下线, 做下线处理
                        System.out.println(socketChannel.getRemoteAddress() + " 下线了...");
                        // 取消注册
                        selectionKey.cancel();
                        // 关闭通道
                        socketChannel.close();
                    } else {
                        e.printStackTrace();
                    }
                }
            }
        }
    
        private void publishMessage(String showMessage, SocketChannel socketChannel) throws IOException {
            System.out.println("服务端接收到消息, 现在进行转发...");
            // 初始化需要发送的消息为ByteBuffer
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            byteBuffer.put(showMessage.getBytes());
            // 遍历每一个注册的客户端进行消息发送
            for (SelectionKey selectionKey : selector.keys()) {
                Channel channel = selectionKey.channel();
                // 跳过自身
                if (channel instanceof SocketChannel && channel != socketChannel) {
                    SocketChannel targetChannel = (SocketChannel) channel;
                    // 消息发送前进行读写转换, 保证每一次都能发出有效数据
                    // 如果出现多个客户端只有一个接收到, 其他没有接受到, 但是服务单正常广播了
                    // 优先查看ByteBuffer问题
                    byteBuffer.flip();
                    targetChannel.write(byteBuffer);
                    System.out.println("发送消息成功, Address: " + targetChannel.getRemoteAddress());
                }
            }
            System.out.println("服务端转发消息成功...");
        }
    
        public static void main(String[] args) throws Exception {
            new GroupChatServer().start();
        }
    
    }

*  客户端

package com.self.netty.nio.chat;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.Buffer;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.Scanner;
    
    /**
     * NIO群聊系统_客户端
     * 1, 连接服务端
     * 2, 发送消息到服务端
     * 3, 接口服务端转发的消息
     *
     * @author pj_zhang
     * @create 2019-12-15 22:28
     **/
    public class GroupChatClient {
    
        // 服务端IP
        private final String HOST = "127.0.0.1";
    
        // 服务端端口
        private final int PORT = 8080;
    
        private SocketChannel socketChannel;
    
        private Selector selector;
    
        public GroupChatClient() throws IOException {
            // 初始化客户端SocketChannel
            socketChannel = SocketChannel.open(new InetSocketAddress(HOST, PORT));
            socketChannel.configureBlocking(false);
            // 初始化选择器
            selector = Selector.open();
            // 绑定事件, 绑定为读事件
            socketChannel.register(selector, SelectionKey.OP_READ);
            System.out.println("客户端已经准备完成, NickName: " + socketChannel.getLocalAddress());
        }
    
        public void start() throws IOException {
            // 接收服务端消息, 此处开线程接收, 保证读写不冲突,不会造成互相影响
            // 先开读, 再去写, 防止写造成的读执行不到
            receiveMessage();
            // 发送消息到服务端
            sendMessage();
        }
    
        private void receiveMessage() throws IOException {
            // 启动一个线程进行服务端数据接收
            new Thread(() -> {
                try {
                    // selector.select() 会阻塞, 直到有连接进入
                    while (selector.select() > 0) {
                        Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                        for (;iterator.hasNext();) {
                            SelectionKey selectionKey = iterator.next();
                            if (selectionKey.isReadable()) {
                                // 接收服务端消息并处理
                                SocketChannel readChannel = (SocketChannel) selectionKey.channel();
                                readChannel.configureBlocking(false);
                                ByteBuffer buffer = ByteBuffer.allocate(1024);
                                int readCount = readChannel.read(buffer);
                                buffer.flip();
                                System.out.println("接收服务端消息: " + new String(buffer.array(), 0, readCount));
                            }
                        }
                        iterator.remove();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
                System.out.println("客户端接收消息完成");
            }).start();
        }
    
        private void sendMessage() throws IOException {
            // 从控制台输入消息
            Scanner scanner = new Scanner(System.in);
            System.out.println("等待客户端输入消息: ");
            // 发送到服务端
            while (scanner.hasNextLine()) {
                String message = scanner.nextLine();
                socketChannel.write(ByteBuffer.wrap(message.getBytes()));
            }
        }
    
        public static void main(String[] args) throws IOException {
            new GroupChatClient().start();
        }
    
    }

[2_NIO]: https://blog.csdn.net/u011976388/article/details/103824155