Java NIO深入详解

傷城~ 2023-06-12 15:26 1阅读 0赞

想要学习Netty，NIO的了解必不可少。

### 什么是NIO ###

IO的方式通常分为几种: 同步阻塞的BIO、同步非阻塞的NIO、异步非阻塞的AIO。

这里简单提及下

*  BIO全称是Blocking IO，是JDK1.4之前的传统IO模型，本身是同步阻塞模式，针对网络通信都是一请求一应答的方式
 *  NIO 即New IO，是JDK1.4中引入的库，是一种非阻塞式IO(non-blocking IO)。本身是基于事件驱动思想来完成的。
 *  AIO（Asynchronous IO），其实是NIO的改进优化，也被称为NIO2.0，在JDK7中发布。提供了异步非堵塞的 IO 操作方式。
 *  关于这三者的具体区别可以参考以下这些博客
    
    > [http://www.imooc.com/article/265871][http_www.imooc.com_article_265871]
    > 
    > [https://blog.51cto.com/14084556/2325377][https_blog.51cto.com_14084556_2325377]
    > 
    > [https://www.cnblogs.com/baxianhua/p/9303354.html][https_www.cnblogs.com_baxianhua_p_9303354.html]

### NIO的核心组件 ###

NIO中涉及到核心三个组件:Selector (选择器)、Channel (通道)、Buffer (缓冲区) 。传统的IO是基于流进行操作的，包括字节流和字符流。而NIO是基于Buffer缓冲区进行数据传输。并且传统IO的流都是单向的，比如各种输入输出流，只能用于输入或者输出。在NIO中数据可以由Buffer写入到Channel(通道)中，也可以由Channel写入到Buffer中。

#### Selector ####

NIO的核心处理器，属于多路复用器 ，实现异步非阻塞IO操作。一个Selector能够处理多个Channel，检测多个Channel上的事件，因此不需要为每一个channel分配一个线程。

#### Channel ####

Channel类似IO中的流，不过流是单向的，Channel是双向的，Channel打开后可以读取，写入或这读写。既可以从通道中读取数据，又可以写数据到通道，并且通道中的数据必须读到一个Buffer中，或者从一个Buffer中写入。

Java NIO中channel的主要实现：

*  SocketChannel （TCP client）
 *  ServerSocketChannel (TCP server)
 *  DatagramChannel（UDP）
 *  FileChannel （文件IO）

#### Buffer ####

Buffer是NIO中的缓冲区，主要和Channel交互，负责从Channel中读取数据，或者写入数据到Channel。

Buffer的实现有

*  ByteBuffer
 *  CharBuffer
 *  DoubleBuffer
 *  FloatBuffer
 *  IntBuffer
 *  LongBuffer
 *  ShortBuffer

分别对应基本类型中的byte, short, int, long, float, double 和 char。要注意上面这些Buffer都是抽象类，每一种都有各自的实现类或者继承了这些类的其他抽象Buffer，类似HeapByteBuffer,HeapIntBuffer,还有Direct(XX)Buffer,Direct(XX)BufferR等。

一个Selector可以处理多个Channel,每个Channel上都可以有Buffer进行读写。需要注意这并不涉及对应关系,只是工作流程

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

### NIO简单示例 ###

先看如下代码，功能是将数据写入到Buffer中再读出来

public static void main(String[] args) {
        
        IntBuffer intBuffer = IntBuffer.allocate(10);
        //向buffer中写入随机数
        SecureRandom secureRandom = new SecureRandom();
        for (int i = 0; i < intBuffer.capacity(); i++) {
        
            int num = secureRandom.nextInt(10);
            intBuffer.put(num);
        }
        //切换模式
        intBuffer.flip();
        //输出buffer中的内容
        while (intBuffer.hasRemaining()) {
        
            System.out.println(intBuffer.get());
        }
    }

其中主要步骤：

*  通过`allocate`方法创建一个Buffer
 *  向Buffer中写入数据
 *  使用`flip`方法切换模式，写状态变为读状态
 *  从Buffer中读取数据

再看一个使用NIO读取文件内容的代码

public static void main(String[] args) {
        
        try (FileInputStream fileInputStream = new FileInputStream("README.md")) {
        
            //获取文件Channel
            FileChannel channel = fileInputStream.getChannel();
            //申请一个Buffer
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            //将Channel中的内容读到buffer中
            channel.read(byteBuffer);
            //将读模式切换为写模式
            byteBuffer.flip();
            //读取buffer中的内容
            while (byteBuffer.hasRemaining()) {
        
                byte b = byteBuffer.get();
                System.out.println((char) b);
            }
        } catch (IOException e) {
        
            e.printStackTrace();
        }
    }

其中主要步骤：

*  通过`FileInputStream`创建一个`Channel`
 *  创建一个`ByteBuffer`
 *  从Channel中向Buffer中写入数据
 *  写完后通过`flip`方法切换模式
 *  从Buffer中读取数据

可以注意到每次Buffer中被写入内容后再读取内容都需要调用`flip`方法切换模式，这是为什么呢？这解决这些问题，必须深入了解Buffer的构成。

### Buffer详解 ###

Buffer意为缓冲区，实际上是一个容器，每种类型的Buffer，底层都是使用对应类型的数组存储数据。向Buffer中写入或读取数据就是对底层数组中写入或读取数据。  
Buffer中有几个重要属性，分别是：

*  capacity(容量)
 *  limit(上限)
 *  position(位置)
 *  mark(标记)

**capacity**表示Buffer的最大容量，最大读取存储量，是`allocate`方法决定的，一个Buffer写满了，必须将其清空才能继续写入数据。

**limit**在初始情况下是和capacity的值一样，当写入数据后，如果Buffer没写满，切换到读模式，limit的就是写入数据的容量，表示写入的数据的容量。例如，初始capacity，limit都为10的buffer，写入6个数据后切换到读模式，此时的limit就是6，表示数据的最大容量。limit的值始终是小于等于capacity的

**position**表示操作数据的当前位置。初始position的值是0，当数据写入到Buffer中，position通过向前移动一位始终表示当前可写的位置。当Buffer从写模式切换到读模式时position又会被置为0，表示当前可读的位置。position的值始终小于limit

**mark**表示标记的一个位置，默认是-1，记录了当前position的前一个位置。可以通过`reset`方法回到mark标记的位置

它们三者的大小：**0<=mark<=position<=limit<=capacity**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NobGF0WmVk_size_16_color_FFFFFF_t_70 1]  
示例：当使用一个小容量Buffer读写文件，需要不断的切换读写模式，并清空Buffer的缓冲。

public static void main(String[] args) throws IOException {
        
        FileInputStream fileInputStream = new FileInputStream("input.txt");
        FileOutputStream fileOutputStream = new FileOutputStream("output.txt");
        FileChannel inputChannel = fileInputStream.getChannel();
        FileChannel outputChannel = fileOutputStream.getChannel();
        ByteBuffer buffer = ByteBuffer.allocate(4);
        while (true) {
        
            //一轮读取后，需要使用clear方法清空缓冲区
            buffer.clear();
            int read = inputChannel.read(buffer);
            if (read == -1) {
        
                break;
            }
            //切换模式
            buffer.flip();
            outputChannel.write(buffer);
        }
        inputChannel.close();
        outputChannel.close();
        fileOutputStream.close();
        fileInputStream.close();
    }

当数据读取到Buffer中被写入后，再次读取数据需要调用clear方法，清除缓冲区。clear,flip方法都是涉及到这几个属性的变化。

查看flip和clear方法的源码就知道为何要调用这些方法，以及这几个属性的变化。

public final Buffer flip() {
        
        limit = position;
        position = 0;
        mark = -1;
        return this;
    }

上面是flip方法的源码，当数据被写入到Buffer后，通过flip方法，limit的值为写入的position的值，position被重置为0，即数据最开始的位置，mark的标记也被重置。例如，容量为10的Buffer，写入6个数据后，此时的position为6(从0开始的)，指向了下一个能被写入的位置。调用flip方法后，limit的值为6，position为0，数据会从position的位置一直读到limit。

public final Buffer clear() {
        
        position = 0;
        limit = capacity;
        mark = -1;
        return this;
    }

在clear方法中，position被重置为0，limit被重置为capacity的大小。所以数据又会从头写入到Buffer中。

你可能会注意到虽然名字叫clear，但是实际上并没有擦除数据，只是将一些索引值重新初始化了。查看源码的Javadoc上也说明了

> This method does not actually erase the data in the buffer, but it is named as if it did because it will most often be used in situations in which that might as well be the case.

这个方法实际上并不会清除buffer中数据，但是它被命名为好像它删除了一样，因为它常常用于这种情况（指清除数据）

因此调用clear方法后其实仍能从buffer中获取到数据。

public static void main(String[] args) {
        
        IntBuffer intBuffer = IntBuffer.allocate(3);
        for (int i = 0; i < 3; i++) {
        
            intBuffer.put(i);
        }
        intBuffer.flip();
        while (intBuffer.hasRemaining()) {
        
            System.out.println(intBuffer.get());
        }
        intBuffer.clear();
        while (intBuffer.hasRemaining()) {
        
            System.out.println(intBuffer.get());
        }
    }

输出结果

0
    1
    2
    0
    1
    2

思考一下，如果上述文件读写示例中input.txt的内容为`0123456789`,注释掉clear方法，文件output.txt中最后会是什么内容。答案是会`0123`一直重复下去

### ByteBuffer的实现类 ###

前面说过ByteBuffer之类的都是抽象类，那他们的实现有哪些呢，在IDEA中继承ByteBuffer的有5个类

![ByteBuffer的实现类][ByteBuffer]

查看UML类图

![查看UML类图][UML]

其中HeapByteBuffer，HeapByteBufferR，DirectByteBuffer,DirectByteBufferR是实现类，MappedByteBuffer是抽象类。注意这些Buffer的实现类都是nio包下可见的，无法在自己的类中引入这些类的。

ByteBuffer中4个重要的属性capacity，limit,position,mark就是在Buffer顶层抽象类中定义的。

查看ByteBuffer的allocate方法可以看到，返回的就是HeapByteBuffer

public static ByteBuffer allocate(int capacity) {
        
        if (capacity < 0)
            throw new IllegalArgumentException();
        return new HeapByteBuffer(capacity, capacity);
    }

HeapByteBufferR是通过ByteBuffer的`asReadOnlyBuffer`返回的，这个方法顾名思义，返回一个只读的Buffer。

HeapByteBufferR是只读Buffer。查看HeapByteBufferR的几个put方法就发现，方法体中直接抛出异常。

public ByteBuffer put(byte x) {
        
        throw new ReadOnlyBufferException();
    }
    public ByteBuffer put(int i, byte x) {
        
        throw new ReadOnlyBufferException();
    }
    public ByteBuffer put(ByteBuffer src) {
        
        throw new ReadOnlyBufferException();
    }

DirectByteBuffer是Buffer的`allocateDirect`方法生成的，意为直接缓冲Buffer，它与HeapByteBuffer 非直接缓冲buffer有着较大的区别

public static ByteBuffer allocateDirect(int capacity) {
        
        return new DirectByteBuffer(capacity);
    }

首先简单了解一下堆内内存和堆外内存的概念。

堆外内存是相对于堆内内存的一个概念。堆内内存是由JVM所管控的Java进程内存，我们平时在Java中创建的对象都处于堆内内存中，并且它们遵循JVM的内存管理机制，JVM会采用垃圾回收机制统一管理它们的内存。那么堆外内存就是存在于JVM管控之外的一块内存区域，因此它是不受JVM的管控。

DirectByteBuffer实现堆外内存的创建，HeapByteBuffer实现堆内内存的创建。关于这两者的具体区别可以参考其他博客的详细介绍

> [https://www.jianshu.com/p/007052ee3773][https_www.jianshu.com_p_007052ee3773]

而DirectByteBufferR和HeapByteBufferR类似，也是属于只读Buffer。

MappedByteBuffer 直接使用内存映射（堆外内存），一般多用于操作大文件。他通过FileChannel的map方法创建，关于它的详细使用可以参考这个

> [https://www.cnblogs.com/xubenben/p/4424398.html][https_www.cnblogs.com_xubenben_p_4424398.html]
> 
> [https://www.jianshu.com/p/f90866dcbffc][https_www.jianshu.com_p_f90866dcbffc]

以下是一个使用MappedByteBuffer的示例，input.txt的内容为`0123456789`

public static void main(String[] args) throws IOException {
        
         RandomAccessFile file = new RandomAccessFile("input.txt", "rw");
         FileChannel channel = file.getChannel();
         //读写模式，从0开始读取5个数据到内存中
         MappedByteBuffer map = channel.map(FileChannel.MapMode.READ_WRITE, 0, 5);
         //修改前两个数据为AB
         map.put(0, (byte) 'A');
         map.put(1, (byte) 'B');
         file.close();
     }

代码运行后从新打开就会发现内存变成了`AB23456789`,注意**在IDEA中打开input.txt会发现内存还是原来的，从资源管理器中打开文件会发现其实内容已经被修改了**

PS 如果查看ByteBuffer的源码可以发现源码格式很乱，包含大量空行，并且在顶部注明`// -- This file was mechanically generated: Do not edit! -- //`，如果好奇为什么会是这样的，可以参考[这里][Link 1]

### Channel ###

在上面这些例子中已经见识到了FileChannel的使用，其余的几个Channel因为都涉及到网络，所有会和Selector一起讲解。

### Selector详解 ###

Selector是SelectableChannel对象的多路复用器，ServerSocketChannel，SocketChannel和DatagramChannel都继承了SelectableChannel。

可以通过调用此类的open方法来创建选择器，该方法将使用系统的默认值selector provider创建一个新的Selector。 还可以通过调用自定义选择器提供程序的openSelector方法来创建Selector。 Selector保持打开，直到通过其close方法关闭。

创建一个Selector`Selector selector = Selector.open();`

将Channel注册到Selector上通过`register`方法：`Channel.register(selector, Selectionkey);`

例如

ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    //serverSocketChannel注册一个accept事件到selector上
    serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

register方法的第二个参数表示`interest set`,可选值由4中，意为channel注册到selector上后对什么事件有反应

*  `SelectionKey.OP_READ`
 *  `SelectionKey.OP_WRITE`
 *  `SelectionKey.OP_CONNECT`
 *  `SelectionKey.OP_ACCEPT`

Selector的可选Channel注册是由SelectKey对象表示，Selector维持了三组SelectKey

*  key set 表示当前channel注册到selector中所有SelectKey，该集合由`keys`方法返回
 *  selected-key set 已注册事件的Channel至少一个事件准备就绪
 *  cancelled-key

在刚初始化的Selector对象中，这三个集合都是空的。 **通过Selector的select（）方法可以选择已经准备好进行I/O操作通道** （这些通道包含你感兴趣的的事件）。比如你对读就绪的通道感兴趣，那么select（）方法就会返回读事件已经就绪的那些通道。

以下是一个socket server的示例，作用是接收到消息原文返回

public static void main(String[] args) throws IOException {
        
        //通过open方法获取一个selector
        Selector selector = Selector.open();
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        //设置成非阻塞模式
        serverSocketChannel.configureBlocking(false);
        //绑定10000端口
        ServerSocket serverSocket = serverSocketChannel.socket();
        serverSocket.bind(new InetSocketAddress(10000));
        //serverSocketChannel注册一个accept事件到selector上
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        while (true) {
        
            System.out.println("重新进行select");
            selector.select();
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            selectionKeys.forEach(selectionKey -> {
        
                try {
        
                    if (selectionKey.isAcceptable()) {
        
                        //当接收到连接就会执行以下代码
                        ServerSocketChannel socketChannel = (ServerSocketChannel) selectionKey.channel();
                        SocketChannel client = socketChannel.accept();
                        client.configureBlocking(false);
                        //建立连接后注册READ事件
                        client.register(selector, SelectionKey.OP_READ);
                        System.out.println("获取客户端连接：" + socketChannel);
                    } else if (selectionKey.isReadable()) {
        
                        SocketChannel client = (SocketChannel) selectionKey.channel();
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        while (true) {
        
                            buffer.clear();
                            int read = client.read(buffer);
                            if (read <= 0) {
        
                                break;
                            }
                            buffer.flip();
                            client.write(buffer);
                        }
                        System.out.println("接收到客户端消息:" + new String(buffer.array()));
                    }
                } catch (Exception e) {
        
                    e.printStackTrace();
                } finally {
        
                    selectionKeys.remove(selectionKey);
                }
            });
        }
    
    }

关于更多selector的介绍可以参考这里

> [https://www.cnblogs.com/snailclimb/p/9086334.html][https_www.cnblogs.com_snailclimb_p_9086334.html]

以下是一个服务端与客户端的示例,client端可以通过控制台输入发送消息到server上

Server

public static void main(String[] args) throws IOException {
        
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        ServerSocket serverSocket = serverSocketChannel.socket();
        serverSocket.bind(new InetSocketAddress(8899));
    
        Selector selector = Selector.open();
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        while (true) {
        
            selector.select();
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            selectionKeys.forEach(selectionKey -> {
        
                try {
        
                    SocketChannel client;
                    if (selectionKey.isAcceptable()) {
        
                        ServerSocketChannel serverSocketChannel1 = (ServerSocketChannel) selectionKey.channel();
                        client = serverSocketChannel1.accept();
                        client.configureBlocking(false);
                        client.register(selector, SelectionKey.OP_READ);
                        String key = "[" + UUID.randomUUID().toString() + "]";
                        map.put(key, client);
                    } else if (selectionKey.isReadable()) {
        
                        client = (SocketChannel) selectionKey.channel();
                        ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                        int count = client.read(readBuffer);
                        if (count > 0) {
        
                            readBuffer.flip();
                            String receivedMessage = String.valueOf(StandardCharsets.UTF_8.decode(readBuffer).array());
                            System.out.println(client + ":" + receivedMessage);
                            String senderKey = null;
                            for (Map.Entry<String, SocketChannel> entry : map.entrySet()) {
        
                                String key = entry.getKey();
                                SocketChannel socketChannel = entry.getValue();
                                if (client == socketChannel) {
        
                                    senderKey = key;
                                    break;
                                }
                            }
                            for (Map.Entry<String, SocketChannel> entry : map.entrySet()) {
        
                                SocketChannel socketChannel = entry.getValue();
                                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                                byteBuffer.put((senderKey + ":" + receivedMessage).getBytes());
                                byteBuffer.flip();
                                socketChannel.write(byteBuffer);
                            }
                        }
                    }
                    System.out.println("length:" + selectionKeys.size());
                    selectionKeys.clear();
                } catch (Exception e) {
        
                    e.printStackTrace();
                } finally {
        
                }
            });
        }
    
    }

Client

public static void main(String[] args) throws IOException {
        
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.configureBlocking(false);
        Selector selector = Selector.open();
        socketChannel.register(selector, SelectionKey.OP_CONNECT);
        socketChannel.connect(new InetSocketAddress("localhost", 8899));
        while (true) {
        
            selector.select();
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            System.out.println("selectionKeys length:" + selectionKeys.size());
            selectionKeys.forEach(selectionKey -> {
        
                try {
        
                    if (selectionKey.isConnectable()) {
        
                        SocketChannel client = (SocketChannel) selectionKey.channel();
                        if (client.isConnectionPending()) {
        
                            client.finishConnect();
                            ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
                            writeBuffer.put((LocalDateTime.now() + " 连接成功").getBytes());
                            writeBuffer.flip();
                            client.write(writeBuffer);
                            ExecutorService executorService = Executors.newSingleThreadExecutor();
                            executorService.submit(() -> {
        
                                while (true) {
        
                                    writeBuffer.clear();
                                    InputStreamReader reader = new InputStreamReader(System.in);
                                    BufferedReader bufferedReader = new BufferedReader(reader);
                                    String sendMessage = bufferedReader.readLine();
                                    writeBuffer.put(sendMessage.getBytes());
                                    writeBuffer.flip();
                                    client.write(writeBuffer);
                                }
                            });
                        }
                        client.register(selector, SelectionKey.OP_READ);
                    } else if (selectionKey.isReadable()) {
        
                        SocketChannel client = (SocketChannel) selectionKey.channel();
                        ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                        int read = client.read(readBuffer);
                        if (read > 0) {
        
                            String receiveMessage = new String(readBuffer.array(), 0, read);
                            System.out.println(receiveMessage);
                        }
                    }
                } catch (IOException e) {
        
                    e.printStackTrace();
                } finally {
        
                    selectionKeys.clear();
                }
            });
        }
    }

[http_www.imooc.com_article_265871]: http://www.imooc.com/article/265871
[https_blog.51cto.com_14084556_2325377]: https://blog.51cto.com/14084556/2325377
[https_www.cnblogs.com_baxianhua_p_9303354.html]: https://www.cnblogs.com/baxianhua/p/9303354.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NobGF0WmVk_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191113134817650.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NobGF0WmVk,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NobGF0WmVk_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191113134923832.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NobGF0WmVk,size_16,color_FFFFFF,t_70
[ByteBuffer]: https://img-blog.csdnimg.cn/20191113134953189.png
[UML]: https://img-blog.csdnimg.cn/20191113135019765.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0NobGF0WmVk,size_16,color_FFFFFF,t_70
[https_www.jianshu.com_p_007052ee3773]: https://www.jianshu.com/p/007052ee3773
[https_www.cnblogs.com_xubenben_p_4424398.html]: https://www.cnblogs.com/xubenben/p/4424398.html
[https_www.jianshu.com_p_f90866dcbffc]: https://www.jianshu.com/p/f90866dcbffc
[Link 1]: https://blog.csdn.net/ChlatZed/article/details/103186012
[https_www.cnblogs.com_snailclimb_p_9086334.html]: https://www.cnblogs.com/snailclimb/p/9086334.html