NIO的介绍（与BIO的区别及使用模型）

本是古典何须时尚 2023-02-12 05:44 4阅读 0赞

### 阻塞与同步 ###

**1、阻塞和非阻塞**  
阻塞和非阻塞是进程在访问数据的时候，数据是否准备就绪的一种处理方式，当数据没有准备的时候一直等待，直到有数据返回，否則一直等待在那里。

非阻塞：当我们的进程访问我们的数据缓冲区的时候，如果数据没有准备好则直接返回，不会等待。如果数据已经准备好，也直接返回。

如果用一个烧水例子解释这个问题：

阻塞：就相当于水壶放在放到火上，立等水开才能知道水是否开了。  
非堵塞：就相当于把水壶放在火上，随时去看，都知道水开还是没开而不是要等到水开了，才能告知结论。如果采用这种方案就是，可以看电视时不时去看，水是否开了。（同步阻塞）

NIO同步非堵塞IO:  
当此此线程中没有数据返回的时候，可以先处理其他线程的任务，不用卡在此线程一直等待数据返回。

**2、同步(Synchronization)和异步(Async)的方式**

同步和异步都是基于应用程序私操作系统处理IO事件所采用的方式，比如同步：是应用程序要直接参与IO读写的操作。异步：所有的IO读写交给搡作系统去处理，应用程序只需要等待通知。

如果用一个烧水例子解释这个问题：

同步：就相当于水壶放在放到火上，水开的时候并不会通知水开了，而需要一个线程去看水是否开了。  
异步：当水开了，会通知我们水已经开了。

异步非阻塞IO：  
获取数据的时候，无论是否有数据都立即返回，并且在内核在缓冲区里填入了数据之后自动调用内存拷贝。

### BIO代码演示 ###

BIO客户端：

public class BIOClient {
    	private static Charset charset = Charset.forName("UTF-8");
    
    	public static void main(String[] args) throws Exception {
    	//开启一个8080端口的线程
    		Socket s = new Socket("localhost", 8080);
    		OutputStream out = s.getOutputStream();
    
    		Scanner scanner = new Scanner(System.in);
    		System.out.println("请输入：");
    		String msg = scanner.nextLine();
    		out.write(msg.getBytes(charset));
    		scanner.close();
    		s.close();
    	}
    
    }

BIO客户端：

bio由于Socket request = serverSocket.accept(); 获取连接和  
InputStream inputStream = request.getInputStream(); 从连接获取流是阻塞的故，当有一个请求，只能创建一个线程去处理，线程一直阻塞利用率并不高。创建的线程过多导致服务器的新能下降明显。

public class BIOServer {
    
        private static ExecutorService threadPool = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                60L, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>());
    
        public static void main(String[] args) throws Exception {
            ServerSocket serverSocket = new ServerSocket(8080);
            System.out.println("服务器启动成功");
            while (!serverSocket.isClosed()) {
                //获取socket 连接数阻塞的
                Socket request = serverSocket.accept();
                System.out.println("收到新连接 : " + request.toString());
                threadPool.execute(() -> {
                    try {
                        // 接收数据、打印（从连接中获取数据同样是阻塞的）
                        InputStream inputStream = request.getInputStream();
                        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, "utf-8"));
                        String msg;
                        while ((msg = reader.readLine()) != null) {
                            if (msg.length() == 0) {
                                break;
                            }
                            System.out.println(msg);
                        }
    
                        System.out.println("收到数据,来自："+ request.toString());
                        // 响应结果 200
                        OutputStream outputStream = request.getOutputStream();
                        outputStream.write("HTTP/1.1 200 OK\r\n".getBytes());
                        outputStream.write("Content-Length: 11\r\n\r\n".getBytes());
                        outputStream.write("Hello World".getBytes());
                        outputStream.flush();
                    } catch (IOException e) {
                        e.printStackTrace();
                    } finally {
                        try {
                            request.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
            serverSocket.close();
        }
    }

产生结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70]

### NIO代码演示 ###

**1、缓冲区Buffer**

缓冲区实际上是一个容器对象，可以看做一个功能增强的数组（有位置、固定容量、标记），在NIO库中，所有数据都是用缓冲区处理的。在读取数据时，它是直接读到缓冲区中的； 在写入数据时，它也是写入到缓冲区中的；而在面向流I/O系统中，所有数据都是直接写入或者直接将数据读取到Stream对象中。

容量（capacity）：  
缓冲区能够容纳的数据元素的最大数量。这一容量在缓冲区创建时被设定，并且永远不能被改变  
上界（limit）：  
缓冲区的第一个不能被读或写的元素。或者说，缓冲区中现存元素的计数  
位置（position）：  
下一个要被读或写的元素的索引。位置会自动由相应的 get( )和 put( )函数更新  
标记（mark）：  
下一个要被读或写的元素的索引。位置会自动由相应的 get( )和 put( )函数更新一个备忘位置。

**2、通道Channel**

通道是一个对象，通过它可以读取和写入数据，当然了所有数据都通过Buffer对象来处理。我们永远不会将字节直接写入通道中，相反是将数据写入包含一个或者多个字节的缓冲区。同样不会直接从通道中读取字节，而是将数据从通道读入缓冲区，再从缓冲区获取这个字节。

![在这里插入图片描述][20200524201307977.png]

**3、选择器**

SelectableChannel 可被注册到 Selector 对象上，同时可以指定对那个选择器而言，哪种操作是感兴趣的，当发生的事件和该事件所发生的具体的SelectableChannel，以获得客户端发送过来的数据。一个通道可以被注册到多个选择器上，但对每个选择器而言，只能被注册一次，通道在被注册到一个选择器上之前，必须先设置为非阻塞模式，通过调用通道的configureBlocking(false)方法即可。这意味着不能将FileChannel与Selector一起使用，因为FileChannel不能切换到非阻塞模式，而套接字通道都可以。

在nio的代码写的过程中有很多需要注意的地方：  
https://www.cnblogs.com/pingh/p/3224990.html

下面设置关于通过nio来演示客户端与服务端的传输

客户端

public class NIOClient {
    
        public static void main(String[] args) throws Exception {
            SocketChannel socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);
            socketChannel.connect(new InetSocketAddress("127.0.0.1", 8080));
            while (!socketChannel.finishConnect()) {
                // 没连接上,则一直等待
                Thread.yield();
            }
            Scanner scanner = new Scanner(System.in);
            System.out.println("请输入：");
            // 发送内容
            String msg = scanner.nextLine();
            ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
            while (buffer.hasRemaining()) {
                socketChannel.write(buffer);
            }
            // 读取响应
            System.out.println("收到服务端响应:");
            ByteBuffer requestBuffer = ByteBuffer.allocate(1024);
    
            while (socketChannel.isOpen() && socketChannel.read(requestBuffer) != -1) {
                // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
                if (requestBuffer.position() > 0){
                    break;
                }
            }
            //nio 切换模式
            requestBuffer.flip();
            byte[] content = new byte[requestBuffer.limit()];
            requestBuffer.get(content);
            System.out.println(new String(content));
            scanner.close();
            socketChannel.close();
        }
    
    }

服务端：  
此处一个selector监听所有事件,演示

public class NIOServerV2 {
    
        public static void main(String[] args) throws Exception {
            // 1. 创建网络服务端ServerSocketChannel
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            // 设置为非阻塞模式
            serverSocketChannel.configureBlocking(false);
    
            // 2. 构建一个Selector选择器,并且将channel注册上去
            Selector selector = Selector.open();
            // 将serverSocketChannel注册到selector
            SelectionKey selectionKey = serverSocketChannel.register(selector, 0, serverSocketChannel);
            // 对serverSocketChannel上面的accept事件感兴趣(serverSocketChannel只能支持accept操作)
            selectionKey.interestOps(SelectionKey.OP_ACCEPT);
            // 3. 绑定端口
            serverSocketChannel.socket().bind(new InetSocketAddress(8080));
    
            System.out.println("启动成功");
    
            while (true) {
                // 用下面轮询事件的方式.select方法有阻塞效果,直到有事件通知才会有返回
                selector.select();
                // 获取事件
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                // 遍历查询结果
                Iterator<SelectionKey> iter = selectionKeys.iterator();
                while (iter.hasNext()) {
                    // 被封装的查询结果
                    SelectionKey key = iter.next();
                    iter.remove();
                    // 关注 Read 和 Accept两个事件
                    if (key.isAcceptable()) {
                        ServerSocketChannel server = (ServerSocketChannel) key.attachment();
                        // 将拿到的客户端连接通道,注册到selector上面
                        SocketChannel clientSocketChannel = server.accept();
                        clientSocketChannel.configureBlocking(false);
                        clientSocketChannel.register(selector, SelectionKey.OP_READ, clientSocketChannel);
                        System.out.println("收到新连接 : " + clientSocketChannel.getRemoteAddress());
                    }
    
                    if (key.isReadable()) {
                        SocketChannel socketChannel = (SocketChannel) key.attachment();
                        try {
                            ByteBuffer requestBuffer = ByteBuffer.allocate(1024);
                            while (socketChannel.isOpen() && socketChannel.read(requestBuffer) != -1) {
                                // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
                                if (requestBuffer.position() > 0) break;
                            }
                            // 如果没数据了, 则不继续后面的处理
                            if(requestBuffer.position() == 0) continue;
                            //切换模式
                            requestBuffer.flip();
                            byte[] content = new byte[requestBuffer.limit()];
                            requestBuffer.get(content);
                            System.out.println(new String(content));
                            System.out.println("收到数据,来自：" + socketChannel.getRemoteAddress());
    
                            // 响应结果 200
                            String response = "HTTP/1.1 200 OK\r\n" +
                                    "Content-Length: 11\r\n\r\n" +
                                    "Hello World";
                            ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
                            while (buffer.hasRemaining()) {
                                socketChannel.write(buffer);
                            }
                        } catch (IOException e) {
                            // e.printStackTrace();
                            // 取消事件订阅
                            key.cancel(); 
                        }
                    }
                }
                selector.selectNow();
            }
         
        }
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 1]

### NIO多路复用模型演示 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 2]

一个selector监听所有事件,个线程处理所有请求事件，这里通过多路复用来展示这个过程，这也是netty采用的方式，只是则是netty的简化版本。

/**
     * NIO selector 多路复用reactor线程模型
     */
    public class NIOServerV3 {
        abstract class EventLoop extends Thread {
    
            Selector selector;
            LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();
    
            /**
             * Selector监听到有事件后,调用这个方法
             */
            public abstract void handler(SelectableChannel channel) throws Exception;
    
            private EventLoop() throws IOException {
                selector = Selector.open();
            }
    
            volatile boolean running = false;
    
            @Override
            public void run() {
                // 轮询Selector事件
                while (running) {
                    try {
                        // 执行队列中的任务
                        Runnable task;
                        while ((task = taskQueue.poll()) != null) {
                            task.run();
                        }
                        selector.select(1000);
    
                        // 获取查询结果
                        Set<SelectionKey> selected = selector.selectedKeys();
                        // 遍历查询结果
                        Iterator<SelectionKey> iter = selected.iterator();
                        while (iter.hasNext()) {
                            // 被封装的查询结果
                            SelectionKey key = iter.next();
                            iter.remove();
                            int readyOps = key.readyOps();
                            // 关注 Read 和 Accept两个事件
                            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                                try {
                                    SelectableChannel channel = (SelectableChannel) key.attachment();
                                    channel.configureBlocking(false);
                                    handler(channel);
                                    if (!channel.isOpen()) {
                                        key.cancel(); // 如果关闭了,就取消这个KEY的订阅
                                    }
                                } catch (Exception ex) {
                                    key.cancel(); // 如果有异常,就取消这个KEY的订阅
                                }
                            }
                        }
                        selector.selectNow();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
    
    
            private SelectionKey register(SelectableChannel channel) throws Exception {
                FutureTask<SelectionKey> futureTask = new FutureTask<>(() -> channel.register(selector, 0, channel));
                taskQueue.add(futureTask);
                return futureTask.get();
            }
    
            private void doStart() {
                if (!running) {
                    running = true;
                    start();
                }
            }
        }
    
        private ServerSocketChannel serverSocketChannel;
        // 1、创建多个线程 - acceptor
        private EventLoop[] bossThreads = new EventLoop[1];
        // 2、创建多个线程 - io
        private EventLoop[] workThreads = new EventLoop[1];
    
        /**
         * 初始化线程组
         */
        private void newGroup() throws IOException {
            // 创建Boss线程, 只负责处理serverSocketChannel
            for (int i = 0; i < bossThreads.length; i++) {
                bossThreads[i] = new EventLoop() {
                    AtomicInteger incr = new AtomicInteger(0);
    
                    @Override
                    public void handler(SelectableChannel channel) throws Exception {
                        // 只做请求分发，不做具体的数据读取
                        ServerSocketChannel ch = (ServerSocketChannel) channel;
                        SocketChannel socketChannel = ch.accept();
                        socketChannel.configureBlocking(false);
                        // 收到连接建立的通知之后，分发给work线程继续去读取数据
                        int index = incr.getAndIncrement() % workThreads.length;
                        EventLoop workEventLoop = workThreads[index];
                        workEventLoop.doStart();
                        SelectionKey selectionKey = workEventLoop.register(socketChannel);
                        selectionKey.interestOps(SelectionKey.OP_READ);
                        System.out.println("收到新连接 : " + socketChannel.getRemoteAddress());
                    }
                };
            }
    
            // 创建IO线程,负责处理客户端连接以后socketChannel的IO读写
            for (int i = 0; i < workThreads.length; i++) {
                workThreads[i] = new EventLoop() {
                    @Override
                    public void handler(SelectableChannel channel) throws IOException {
                        // work线程只负责处理IO处理，不处理accept事件
                        SocketChannel ch = (SocketChannel) channel;
                        ByteBuffer requestBuffer = ByteBuffer.allocate(1024);
                        while (ch.isOpen() && ch.read(requestBuffer) != -1) {
                            // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
                            if (requestBuffer.position() > 0){
                                break;
                            }
                        }
                        // 如果没数据了, 则不继续后面的处理
                        if (requestBuffer.position() == 0){
                            return;
                        }
                        requestBuffer.flip();
                        byte[] content = new byte[requestBuffer.limit()];
                        requestBuffer.get(content);
                        System.out.println(new String(content));
                        System.out.println("收到数据,来自：" + ch.getRemoteAddress());
    
                        // 响应结果 200
                        String response = "HTTP/1.1 200 OK\r\n" +
                                "Content-Length: 11\r\n\r\n" +
                                "Hello World";
                        ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
                        while (buffer.hasRemaining()) {
                            ch.write(buffer);
                        }
                    }
                };
            }
        }
    
        /**
         * 初始化channel,并且绑定一个eventLoop线程
         *
         * @throws IOException IO异常
         */
        private void initAndRegister() throws Exception {
            // 1、 创建ServerSocketChannel
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            // 2、 将serverSocketChannel注册到selector
            int index = new Random().nextInt(bossThreads.length);
            bossThreads[index].doStart();
            SelectionKey selectionKey = bossThreads[index].register(serverSocketChannel);
            selectionKey.interestOps(SelectionKey.OP_ACCEPT);
        }
    
        /**
         * 绑定端口
         *
         * @throws IOException IO异常
         */
        private void bind() throws IOException {
            //  1、 正式绑定端口，对外服务
            serverSocketChannel.bind(new InetSocketAddress(8080));
            System.out.println("启动完成，端口8080");
        }
    
        public static void main(String[] args) throws Exception {
            NIOServerV3 nioServerV3 = new NIOServerV3();
            nioServerV3.newGroup();
            nioServerV3.initAndRegister();
            nioServerV3.bind();
            int read = System.in.read();
            System.out.println(read);
        }
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200524201908385.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU=,size_16,color_FFFFFF,t_70
[20200524201307977.png]: https://img-blog.csdnimg.cn/20200524201307977.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200524222407591.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200524223646676.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU=,size_16,color_FFFFFF,t_70