Java NIO 详解

一、NIO简介

NIO 是 Java SE 1.4 引入的一组新的 I/O 相关的 API，它提供了非阻塞式 I/O、选择器、通道、缓冲区等新的概念和机制。相比与传统的 I/O 多出的 N 不是单纯的 New，更多的是代表了 Non-blocking 非阻塞，NIO具有更高的并发性、可扩展性以及更少的资源消耗等优点。

二、NIO 与传统BIO

NIO：是同步非阻塞的，服务器实现模式为 一个线程处理多个连接。服务端只会创建一个线程负责管理Selector（多路复用器），Selector（多路复用器）不断的轮询注册其上的Channel（通道）中的 I/O 事件，并将监听到的事件进行相应的处理。每个客户端与服务端建立连接时会创建一个 SocketChannel 通道，通过 SocketChannel 进行数据交互。

BIO：全称是Blocking IO，同步阻塞式IO，是JDK1.4之前的传统IO模型，服务器实现模式为一个连接一个线程。每当客户端有连接请求时服务器端就需要启动一个线程进行处理。

两者主要区别如下：

1. 阻塞和非阻塞：NIO 使用非阻塞式 I/O，而 BIO 使用阻塞式 I/O。在阻塞式 I/O 中，当一个 I/O 操作完成之前，线程会一直被阻塞，直到 I/O 操作完成；在非阻塞式 I/O 中，线程可以继续执行其他任务，直到 I/O 操作完成并返回结果。
2. 线程模型：NIO 中的线程模型是基于事件驱动的，当一个 I/O 操作完成时，会触发相应的事件通知线程处理；而在 BIO 中，每个线程都负责处理一个客户端连接，需要不断地轮询客户端的输入输出流，以便及时响应客户端的请求。
3. 内存消耗：NIO 中使用的缓冲区（Buffer）可以重复利用，减少了频繁的内存分配和回收，从而减少了内存的消耗；而在 BIO 中，每个客户端连接都需要单独分配一个缓冲区，容易造成内存的浪费。
4. 并发性能：NIO 中使用非阻塞式 I/O，可以同时处理多个客户端连接，从而提高了并发处理能力；而在 BIO 中，由于每个客户端连接都需要一个线程来处理，当连接数量增加时，容易出现线程饥饿和资源耗尽的问题。

三、NIO 工作流程

1. 创建 Selector：Selector 是 NIO 的核心组件之一，它可以同时监听多个通道上的 I/O 事件，并且可以通过 select() 方法等待事件的发生。
2. 注册 Channel：通过 Channel 的 register() 方法将 Channel 注册到 Selector 上，这样 Selector 就可以监听 Channel 上的 I/O 事件。
3. 等待事件：调用 Selector 的 select() 方法等待事件的发生，当有事件发生时，Selector 就会通知相应的线程进行处理。
4. 处理事件：根据不同的事件类型，调用对应的处理逻辑。
5. 关闭 Channel：当 Channel 不再需要使用时，需要调用 Channel 的 close() 方法关闭 Channel，同时也需要调用 Buffer 的 clear() 方法清空 Buffer 中的数据，以释放内存资源。

Java NIO 的工作流程可以简单概括为：通过 Selector 监听多个 Channel 上的 I/O 事件，当事件发生时，通过对应的 Channel 进行读写操作，并在 Channel 不再需要使用时关闭 Channel。

四、NIO 核心的组件

1. Channel(通道)

Channel 是应用程序与操作系统之间交互事件和传递内容的直接交互渠道，应用程序可以从管道中读取操作系统中接收到的数据，也可以向操作系统发送数据。Channel和传统IO中的Stream很相似，其主要区别为：通道是双向的，通过一个Channel既可以进行读，也可以进行写；而Stream只能进行单向操作，通过一个Stream只能进行读或者写，比如InputStream只能进行读取操作，OutputStream只能进行写操作。

1.1 常用的Channel实现类

1. FileChannel：本地文件IO通道，从文件中读写数据。一般流程为：

1.获取文件通道，通过 FileChannel 的静态方法 open() 来获取，获取时需要指定文件路径和文件打开方式
FileChannel channel = FileChannel.open(Paths.get(fileName), StandardOpenOption.READ);

2.创建字节缓冲区
ByteBuffer buf = ByteBuffer.allocate(1024);

3.读/写操作
(1)、读操作
// 循环读取通道中的数据，并写入到 buf 中
while (channel.read(buf) != -1){ 
    // 缓存区切换到读模式
    buf.flip(); 
    // 读取 buf 中的数据
    while (buf.position() < buf.limit()){ 
    	// 将buf中的数据追加到文件中
        text.append((char)buf.get());
    }
    // 清空已经读取完成的 buffer，以便后续使用
    buf.clear();
}
(2)、写操作
// 循环读取文件中的数据，并写入到 buf 中
for (int i = 0; i < text.length(); i++) {
    // 填充缓冲区，需要将 2 字节的 char 强转为 1 自己的 byte
    buf.put((byte)text.charAt(i)); 
    // 缓存区已满或者已经遍历到最后一个字符
    if (buf.position() == buf.limit() || i == text.length() - 1) { 
        // 将缓冲区由写模式置为读模式
        buf.flip(); 
        // 将缓冲区的数据写到通道
        channel.write(buf); 
        // 清空已经读取完成的 buffer，以便后续使用
        buf.clear(); 
    }
}

4.将数据刷出到物理磁盘
channel.force(false);

5.关闭通道
channel.close();

2. SocketChannel：网络套接字IO通道，TCP协议，客户端通过 SocketChannel 与服务端建立TCP连接进行通信交互。与传统的Socket操作不同的是，SocketChannel基于非阻塞IO模式，可以在同一个线程内同时管理多个通信连接，从而提高系统的并发处理能力。

1.打开一个 SocketChannel 通道
SocketChannel channel = SocketChannel.open();

2.连接到服务端
channel.connect(new InetSocketAddress("localhost", 9001));

3.分配缓冲区
ByteBuffer buf = ByteBuffer.allocate(1024); 

4.配置是否为阻塞方式（默认为阻塞方式）
channel.configureBlocking(false); // 配置通道为非阻塞模式

5.将channel的连接、读、写等事件注册到selector中，每个chanel只能注册一个事件，最后注册的一个生效,
同时注册多个事件可以使用"|"操作符将常量连接起来
Selector selector = Selector.open();
channel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_WRITE | SelectionKey.OP_READ);

6.与服务端进行读写操作
channel.read(buf);
channel.write(buf);

7.关闭通道
channel.close();

3. ServerSocketChannel：网络套接字IO通道，TCP协议，服务端通过ServerSocketChannel监听来自客户端的连接请求，并创建相应的SocketChannel对象进行通信交互。ServerSocketChannel同样也是基于非阻塞IO模式，可以在同一个线程内同时管理多个通信连接，从而提高系统的并发处理能力。

1.打开一个 ServerSocketChannel 通道
ServerSocketChannel serverChannel = ServerSocketChannel.open();

2.绑定本地端口
serverChannel.bind(new InetSocketAddress(9001));

3.配置是否为阻塞方式（默认为阻塞方式）
serverChannel.configureBlocking(false); // 配置通道为非阻塞模式

4.分配缓冲区
ByteBuffer buf = ByteBuffer.allocate(1024); 

5.将serverChannel 的连接、读、写等事件注册到selector中，每个chanel只能注册一个事件，最后注册的一个生效,
同时注册多个事件可以使用"|"操作符将常量连接起来
Selector selector = Selector.open();
serverChannel.register(selector, SelectionKey.OP_ACCEPT| SelectionKey.OP_WRITE | SelectionKey.OP_READ);

6.与客服端进行读写操作
serverChannel.read(buf);
serverChannel.write(buf);

7.关闭通道
serverChannel.close();

4. DatagramChannel：DatagramChannel是Java NIO中对UDP协议通信的封装。通过DatagramChannel对象，我们可以实现发送和接收UDP数据包。它与TCP协议不同的是，UDP协议没有连接的概念，所以无需像SocketChannel一样先建立连接再开始通信。

1.打开一个 DatagramChannel 通道
DatagramChannel channel = DatagramChannel.open();

2.分配缓冲区
ByteBuffer buf = ByteBuffer.allocate(1024); 

3.配置是否为阻塞方式（默认为阻塞方式）
channel.configureBlocking(false); // 配置通道为非阻塞模式

4.与客服端进行读写操作
buffer.flip();
// 发送消息给服务端
channel.send(buffer, new InetSocketAddress("localhost", 9001));
buffer.clear();
// 接收服务端的响应信息
channel.receive(buffer);
buffer.flip();
// 打印出响应信息
while (buffer.hasRemaining()) {
     System.out.print((char) buffer.get());
}
buffer.clear();

7.关闭通道
channel.close();

1.2 常用的Channel方法

1. read(ByteBuffer)：从 Channel 中读取数据到 ByteBuffer 中。如果 Channel 中没有可读数据，则会阻塞等待直到有数据可读。
2. write(ByteBuffer)：将数据写入到 Channel 中。如果 Channel 中没有可写空间，则会阻塞等待直到有可写空间。
3. read(ByteBuffer, long)：从 Channel 中读取数据到 ByteBuffer 中，并设置读取超时时间。如果超时时间到了还没有读取到数据，则会抛出 TimeoutException 异常。
4. write(ByteBuffer, long)：将数据写入到 Channel 中，并设置写入超时时间。如果超时时间到了还没有写入完成，则会抛出 TimeoutException 异常。
5. flush()：将 Channel 中的缓冲区数据刷新到底层设备中，如果没有数据需要刷新，则会立即返回。
6. register(SelectionKey, int)：将 Channel 注册到 Selector 上，并设置注册的事件类型和操作。可以通过 Selector 监听 Channel 上的事件，当有事件发生时，Selector 就会通知相应的线程进行处理。
7. configureBlocking(boolean)：设置 Channel 是否为阻塞模式。如果为阻塞模式，则在读取或写入数据时会一直阻塞等待，直到有数据可读或写入完成；如果为非阻塞模式，则在读取或写入数据时会立即返回，如果没有数据可读或写入完成，则会返回 -1。
8. socket()：获取底层的 Socket 对象。
9. isConnected()：判断 Channel 是否已经连接到了远程主机。
10. isWritable()：判断 Channel 是否可以写入数据。
11. isReadable()：判断 Channel 是否可以读取数据。
12. isOpen()：检查 Channel 是否已经打开。
13. getRemoteAddress()：获取 Channel 对应的远程地址。
14. getLocalAddress()：获取 Channel 对应的本地地址。

2. Buffer(缓冲区)

NIO 中的数据都是通过 Buffer 对象来处理的，每个 Buffer 对象都关联着一个字节数组，可以保存多个相同类型的数据。在读取数据时，是从Buffer 中读取的，在写入数据时，也是写入到 Buffer 中的。

2.1 Buffer 常用子类

1. ByteBuffer：用于存储字节数据；
2. CharBuffer：用于存储字符数据；
3. ShortBuffer：用于存储Short类型数据；
4. IntBuffer：用于存储Int类型数据；
5. LongBuffer：用于存储Long类型数据；
6. FloatBuffer：用于存储Float类型数据；
7. DoubleBuffer：用于存储Double类型数据；

2.2 Buffer 重要属性

1. capacity(容量)：表示 Buffer 所占的内存大小，capacity不能为负，并且创建后不能更改。
2. limit(限制)：表示 Buffer 中可以操作数据的大小，limit不能为负，并且不能大于其capacity。写模式下，表示最多能往 Buffer 里写多少数据，即 limit 等于 Buffer 的capacity。读模式下，表示你最多能读到多少数据，其实就是能读到之前写入的所有数据。
3. position(位置)：表示下一个要读取或写入的数据的索引。缓冲区的位置不能为负，并且不能大于其限制。初始的 position 值为 0，最大可为 capacity – 1。当一个 byte、long 等数据写到 Buffer 后， position 会向前移动到下一个可插入数据的 Buffer 单元。
4. mark(标记)：表示记录当前 position 的位置。可以通过 reset() 恢复到 mark 的位置。
   

2.3 Buffer 常见方法

1. clear()：清空缓冲区并返回对缓冲区的引用；
2. flip()：将缓冲区的界限设置为当前位置，并将当前位置重置为 0；
3. capacity()：返回 Buffer 的 capacity 大小；
4. limit()：返回 Buffer 的界限(limit) 的位置；
5. limit(int n)：将设置缓冲区界限为 n，并返回一个具有新 limit 的缓冲区对象；
6. position()：返回缓冲区的当前位置 position；
7. position(int n)：将设置缓冲区的当前位置为 n， 并返回修改后的 Buffer 对象；
8. mark()：对缓冲区设置标记；
9. reset()：将位置 position 转到以前设置的mark 所在的位置；
10. rewind()：将位置设为为 0， 取消设置的 mark；
11. hasRemaining()：判断缓冲区中是否还有元素；
12. get()：读取单个字节；
13. get(byte[] dst)：读取多个字节；
14. get(int index)：读取指定索引位置的字节；
15. put(byte b)：将给定单个字节写入缓冲区的当前位置；
16. put(byte[] src)：将数组中的字节从当前位置依次写入到缓冲区中；
17. put(int index, byte b)：将指定字节写入缓冲区的索引位置；

2.4 Buffer 内存分配

1. 普通缓冲区：通过allocate()方法进行分配，可以在jvm堆上申请堆上内存。如果要作IO操作，会先从本进程的堆上内存复制到直接内存，再利用本地IO处理。

ByteBuffer readBuffer = ByteBuffer.allocate(1024);

2. 直接缓冲区：通过allocateDirect()方法进行分配，直接从本地内存中申请。如果要作IO操作，直接从本地内存中利用本地IO处理。使用直接内存会具有更高的效率，但是它比申请普通的堆内存需要耗费更高的性能。直接内存中的数据是在JVM之外的，因此它不会占用应用的内存，当有很大的数据要缓存，并且它的生命周期又很长，那么就比较适合使用直接内存。一般来说，如果不是能带来很明显的性能提升，还是推荐使用堆内存。

ByteBuffer directByteBuffer = ByteBuffer.allocateDirect(1024);

3. 缓冲区分片：通过slice()方法可以根据现有的缓冲区对象来创建一个子缓冲区，即在现有缓冲区上切出一片来作为一个新的缓冲区，但现有的缓冲区与创建的子缓冲区在底层数组层面上是数据共享的。

ByteBuffer readBuffer = ByteBuffer.allocate(1024);
ByteBuffer sliceBuffer = readBuffer.slice();

4. 只读缓冲区：通过asReadOnlyBuffer()方法可以将任何常规缓冲区转换为只读缓冲区，这个方法返回 一个与原缓冲区完全相同的缓冲区，并与原缓冲区共享数据，只不过它是只读的。如果原缓冲区的内容发生了变化，只读缓冲区的内容也随之发生变化。

ByteBuffer readBuffer = ByteBuffer.allocate(1024);
ByteBuffer readonlyBuffer = readBuffer.asReadOnlyBuffer();

3. Selector(选择器)

Selector 提供了选择已经就绪的任务的能力，Selector会不断的轮询注册在上面的所有channel，进行后续的IO操作。只需通过一个单独的线程就可以管理多个channel，从而管理多个网络连接。这就是Nio与传统I/O最大的区别，不用为每个连接都去创建一个线程。

3.1 Selector使用流程

1.获取选择器
Selector selector = Selector.open();

2.通道注册到选择器，进行监听
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

3.获取可操作的 Channel
selector.select();

4.获取可操作的 Channel 中的就绪事件集合
Set<SelectionKey> keys = selector.selectedKeys();

5.处理就绪事件
while (keys.iterator().hasNext()){
	SelectionKey key = keys.iterator().next();
	if (!key.isValid()){
		continue;
	}
    if (key.isAcceptable()){
		accept(key);
	}
	if(key.isReadable()){
		read(key);
	}
	if (key.isWritable()){
		write(key);
	}
	keyIterator.remove(); //移除当前的key
}

3.2 SelectionKey事件类型

每个 Channel向Selector 注册时，都会创建一个 SelectionKey 对象，通过 SelectionKey 对象向Selector 注册，且 SelectionKey 中维护了 Channel 的事件。常见的四种事件如下：

1. OP_READ：当操作系统读缓冲区有数据可读时就绪。
2. OP_WRITE：当操作系统写缓冲区有空闲空间时就绪。
3. OP_CONNECT：当 SocketChannel.connect()请求连接成功后就绪，该操作只给客户端使用。
4. OP_ACCEPT：当接收到一个客户端连接请求时就绪，该操作只给服务器使用。

五、简单实例

1. 服务端

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;

public class NioServiceTest {
    private Selector selector;
    private ServerSocketChannel serverSocketChannel;
    private ByteBuffer readBuffer = ByteBuffer.allocate(1024);//调整缓冲区大小为1024字节
    private ByteBuffer sendBuffer = ByteBuffer.allocate(1024);
    String str;

    public NioServiceTest(int port) throws IOException {
        // 打开服务器套接字通道
        this.serverSocketChannel = ServerSocketChannel.open();
        // 服务器配置为非阻塞 即异步IO
        this.serverSocketChannel.configureBlocking(false);
        // 绑定本地端口
        this.serverSocketChannel.bind(new InetSocketAddress(port));
        // 创建选择器
        this.selector = Selector.open();
        // 注册接收连接事件
        this.serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    }

    public void handle() throws IOException {
        // 无限判断当前线程状态，如果没有中断，就一直执行while内容。
        while(!Thread.currentThread().isInterrupted()){
            // 获取准备就绪的channel
            if (selector.select() == 0) {
                continue;
            }

            // 获取到对应的 SelectionKey 对象
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> keyIterator = keys.iterator();
            // 遍历所有的 SelectionKey 对象
            while (keyIterator.hasNext()){
                // 根据不同的SelectionKey事件类型进行相应的处理
                SelectionKey key = keyIterator.next();
                if (!key.isValid()){
                    continue;
                }
                if (key.isAcceptable()){
                    accept(key);
                }
                if(key.isReadable()){
                    read(key);
                }
                // 移除当前的key
                keyIterator.remove();
            }
        }
    }

    /**
     * 客服端连接事件处理
     *
     * @param key
     * @throws IOException
     */
    private void accept(SelectionKey key) throws IOException {
        SocketChannel socketChannel = this.serverSocketChannel.accept();
        socketChannel.configureBlocking(false);
        // 注册客户端读取事件到selector
        socketChannel.register(selector, SelectionKey.OP_READ);
        System.out.println("client connected " + socketChannel.getRemoteAddress());
    }

    /**
     * 读取事件处理
     *
     * @param key
     * @throws IOException
     */
    private void read(SelectionKey key) throws IOException{
        SocketChannel socketChannel = (SocketChannel) key.channel();
        //清除缓冲区，准备接受新数据
        this.readBuffer.clear();
        int numRead;
        try{
            // 从 channel 中读取数据
            numRead = socketChannel.read(this.readBuffer);
        }catch (IOException e){
            System.out.println("read failed");
            key.cancel();
            socketChannel.close();
            return;
        }
        str = new String(readBuffer.array(),0,numRead);
        System.out.println("read String is: " + str);
    }

    public static void main(String[] args) throws Exception {
        System.out.println("sever start...");
        new NioServiceTest(8000).handle();
    }
}

2. 客户端

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.SocketChannel;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;

public class NioClientTest {
    ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
    private SocketChannel sc;
    private Selector selector;

    public NioClientTest(String hostname, int port) throws IOException {
        // 打开socket通道
        sc = SocketChannel.open();
        // 配置为非阻塞 即异步IO
        sc.configureBlocking(false);
        // 连接服务器端
        sc.connect(new InetSocketAddress(hostname,port));
        // 创建选择器
        selector = Selector.open();
        // 注册请求连接事件
        sc.register(selector, SelectionKey.OP_CONNECT);
    }

    public void send() throws IOException{
        Scanner scanner = new Scanner(System.in);
        // 无限判断当前线程状态，如果没有中断，就一直执行while内容。
        while (!Thread.currentThread().isInterrupted()){
            // 获取准备就绪的channel
            if (selector.select() == 0) {
                continue;
            }

            // 获取到对应的 SelectionKey 对象
            Set<SelectionKey> keys = selector.selectedKeys();
            System.out.println("all keys is:"+keys.size());
            Iterator<SelectionKey> iterator = keys.iterator();
            // 遍历所有的 SelectionKey 对象
            while (iterator.hasNext()){
                SelectionKey key = iterator.next();
                //判断此通道上是否在进行连接操作
                if (key.isConnectable()){
                    sc.finishConnect();
                    //注册写操作
                    sc.register(selector, SelectionKey.OP_WRITE);
                    System.out.println("server connected...");
                    break;
                }else if (key.isWritable()){
                    System.out.println("please input message:");
                    String message = scanner.nextLine();
                    writeBuffer.clear();
                    writeBuffer.put(message.getBytes());
                    //将缓冲区各标志复位,因为向里面put了数据标志被改变要想从中读取数据发向服务器,就要复位
                    writeBuffer.flip();
                    sc.write(writeBuffer);
                }
                // 移除当前的key
                iterator.remove();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        new NioClientTest("localhost", 8000).send();
    }
}