都将会
都将会

NIO

 都将会 收录于 202308
  约 3656 字 预计阅读 8 分钟  - 次阅读  

NIO 于 jdk1.4 引入，支持面向缓冲区的、基于通道的IO操作。

通道负责传输，缓冲区负责存储。

Buffer 缓冲区

高效的数据容器，除了布尔类型，所有原始数据类型都有相应的 Buffer 实现。

1
2

// 获取缓冲区，单位：B 字节
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

核心参数

mark <= position <= limit <= capacity

  • mark：当前 position 保存点，通过 reset() 恢复到 mark 位置
    • mark - position 区域：未保存
  • position：下一个读写位置的索引
    • position - limit 区域：可读写
  • limit：缓冲区界限
    • limit - capacity 区域：不可读写
  • capacity：缓冲区容量

核心方法

  • allocate(int capacity)：分配非直接缓冲区，位于 JVM 堆内存之中。

  • allocateDirect(int capacity)：分配直接缓冲区，位于物理内存之中。（多个虚拟空间指向同一个物理地址）

  • put()：写入数据，position++,capacity = limit

  • flip()：读写切换

    • 读模式： limit = position+1, position = 0
    • 写模式：position = limit +1, limit = position

  • get()：读取数据，position++

  • rewind()：只能在读模式下使用，恢复参数到 get() 前

  • clean()：指针归位，数据仍然存在，position = 0,capacity = limit

  • mark()：mark = position

  • reset()：position = mark

实践

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

import java.nio.ByteBuffer;
public class Demo1 {
    public static void main(String[] args) {
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        printByteBuffer(byteBuffer, "初始参数");

        System.out.println("------put()------");
        System.out.println("放入10个数据");
        byte bt = 1;
        for (int i = 0; i < 10; i++) {
            byteBuffer.put(bt);
        }
        printByteBuffer(byteBuffer, "放入后参数");

        System.out.println("------flip()-get()------");
        System.out.println("读取一个数据");
        // 切换模式
        byteBuffer.flip();
        byteBuffer.get();

        printByteBuffer(byteBuffer, "读取后参数");

        System.out.println("------rewind()------");
        byteBuffer.rewind();
        printByteBuffer(byteBuffer, "恢复后参数");

        System.out.println("------clear()------");
        // 清空缓冲区，这里只是恢复了各个属性的值，但是缓冲区里的数据依然存在
        // 但是下次写入的时候会覆盖缓冲区中之前的数据
        byteBuffer.clear();
        printByteBuffer(byteBuffer, "清空后参数");
        System.out.println("清空后获得旧数据");
        System.out.println(byteBuffer.get());

    }

    private static void printByteBuffer(ByteBuffer byteBuffer, String string) {
        System.out.println("|=======" + string + "=============");
        System.out.println("|position " + byteBuffer.position());
        System.out.println("|limit    " + byteBuffer.limit());
        System.out.println("|capacity " + byteBuffer.capacity());
        System.out.println("|=============================");
    }
}

输出：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

|=======初始参数=============
|position 0
|limit    1024
|capacity 1024
|=============================
------put()------
放入10个数据
|=======放入后参数=============
|position 10
|limit    1024
|capacity 1024
|=============================
------flip()-get()------
读取一个数据
|=======读取后参数=============
|position 1
|limit    10
|capacity 1024
|=============================
------rewind()------
|=======恢复后参数=============
|position 0
|limit    10
|capacity 1024
|=============================
------clear()------
|=======清空后参数=============
|position 0
|limit    1024
|capacity 1024
|=============================
清空后获得旧数据
1

Channel 通道

Channel 表示 IO 源与目标打开的连接，它本身不能直接访问数据，Channel 只能与Buffer 进行交互。channel 类似在 Linux 之类操作系统上看到的文件描述符，是 NIO 中被用来支持批量式 IO 操作的一种抽象。File 或者 Socket，通常被认为是比较高层次的抽象，而 Channel 则是更加操作系统底层的一种抽象，这也使得 NIO 得以充分利用现代操作系统底层机制，获得特定场景的性能优化，例如，DMA（Direct Memory Access）等。

  • FileChanner：本地文件 IO，阻塞式
    • 通过 FileInputStream 获取的 channel 只能读
    • 通过 FileOutputStream 获取的 channel 只能写
    • 通过 RandomAccessFile 是否能读写根据构造 RandomAccessFile 时的读写模式决定
  • SocketChannel：客户端 TCP 传输
  • ServerSocketChannel：服务端 TCP 传输
  • DatagramChannel：UDP 传输

获得通道的方式

  1. stream.getChannel() 或socket.getChannel()获得通道 + allocate() 获取非直接缓冲区（最慢）
  2. FileChannel.open() 获得通道 + fileChannel.map() 获取直接缓冲区
    1. 直接写入内存映射文件：inMapBuf.get(bytes); outMapBuf.put(bytes)（对应 Linux 中的 mmap() 操作，内核缓冲区与 socket 缓冲区共享，节省了一次 CPU拷贝。读写速度虽快，但内存占用高）
    2. 通道间直接传输：inChannel.transferTo((0,inChannel.size(),outChannel))（最快）

Stream 与 Channel 的区别

  1. stream 不会自动缓冲数据，channel 会利用系统提供的缓冲区。
  2. stream 仅支持阻塞 API，channel 同时支持阻塞、非阻塞 API，网络 channel 可配合 selector 实现多路复用。

Selector 选择器

多路复用是非阻塞 IO 的核心，通过 Selector 可以实现单个线程管理多个 Channel，适合连接数多，但流量较少的场景。

  • 多路复用仅针对网络 IO：Selector可以检测到注册在 Selector 上的多个 Channel 中，是否有 Channel 处于就绪状态，进而实现了单线程对多 Channel 的高效管理。
  • 普通文件 IO 无法利用多路复用。
1
2
3
4
5
6

// 创建选择器
Selector selector = Selector.open();
// Q：设置 channel 为非阻塞模式 ？
socketChannel.configureBlocking(false);
// 监听状态，多个状态通过“|”分隔
channel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_ACCEPT);

A：阻塞模式下，注册操作是不允许的，会抛出 IllegalBlockingModeException 异常。

SelectionKey 绑定的事件类型

  • connect：客户端连接成功时触发
  • accept：服务端连接成功时触发
  • read：数据可读入时
  • write：数据可写出时

Selector 监听类型

  • select()：阻塞直到绑定事件发生
  • select(long)：阻塞直到绑定事件发生，或超时
  • selectNow：不会阻塞，检查是否有事件，立刻返回

事件处理

事件发生后，要么正常处理 accept()，要么 catch 异常时取消 cancel()，如果什么都不做，下次该事件仍会触发。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// 获取所有事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();           
// 使用迭代器遍历事件
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
	SelectionKey key = iterator.next();                
	// 判断key的类型，此处为Accept类型
	if(key.isAcceptable()) {
        // 获得key对应的channel
        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
        // 获取连接并处理，而且是必须处理，否则需要取消
        SocketChannel socketChannel = channel.accept();
        // 处理完毕后移除，否则会导致已被处理过的事件再次被处理。
        iterator.remove();
	}
}

消息边界处理

  • 固定消息长度：浪费带宽。
  • 按分隔符拆分：效率低。
  • TLV 格式（Type 类型、Length 长度、Value 数据）：需要提前分配，如果内容过大，影响 server 吞吐量。

多线程优化

将选择器 Selector 分成两组：

  • 单线程配一个选择器 Boss，专门处理 accept（） 事件。

    • 接受并处理Accepet事件，当Accept事件发生后，调用Worker的register(SocketChannel socket)方法，让Worker去处理Read事件，其中需要根据标识robin去判断将任务分配给哪个Worker。
    1
2
3
4
5
6

    // 创建固定数量的Worker
Worker[] workers = new Worker[4];
// 用于负载均衡的原子整数
AtomicInteger robin = new AtomicInteger(0);
// 负载均衡，轮询分配Worker
workers[robin.getAndIncrement()% workers.length].register(socket);
    • 添加任务后需要调用selector.wakeup()来唤醒被阻塞的Selector，select类似LockSupport中的park，wakeup的原理类似LockSupport中的unpark。

  • 创建 CPU 核心数的线程 Worker，每个线程配一个选择器，轮流处理 read() 事件。

    • 从同步队列中获取注册任务，并处理Read事件。

IO 演进

五种 IO 模型

BIO

在while(true)中调用accept()方法等待客户端。一个线程只能同时处理一个连接请求。

4 次上下文切换 + 2 次 DMA 拷贝 + 2 次 CPU 拷贝

DMA：Direct Memory Access，即直接内存访问，DMA 本质上是一块主板上的芯片，它直接参与外设设备和内存存储器之间的 IO 数据传输，过程中不需要 CPU 参与。

NIO

内核会立即返回，已获得足够的CPU事件继续做其他事情，用户进程只是在等待阶段非阻塞，需要不断主动询问kernel数据是否准备好，第二阶段仍然阻塞。

img

多路复用

通过 Selector 实现，没有事件时调用select方法会被阻塞住，发生事件后就会处理。进程首先阻塞在select/poll上，再阻塞在读操作的第二阶段上。

img

信号驱动IO

在IO执行的数据准备阶段，不会阻塞用户进程，当用户进程收到信号后，才去查收数据。

异步 IO

线程 1 调用方法后立即返回，由线程 2 返回最终结果，需要底层操作系统（Kernel）提供支持。

  • Windows 系统通过 IOCP 实现了真正的异步 IO
  • Linux 系统异步 IO 在 2.6 版本引入，但其底层实现还是用多路复用模拟了异步 IO，性能没有优势
img

零拷贝

数据不经过 JVM 内存，直接拷贝到目的区域，适合小文件传输。

  1. ByteBuffer.allocateDirect(10)：将堆外操作系统内存映射到 JVM 内存中来直接访问使用，减少一次数据拷贝，内存回收通过将虚引用加入引用队列，再调用 Unsafe.freeMemory() 方法释放底层资源。

  2. sendFile()：于 Linux2.1 引入，应用于channel.transferTo()/transferFrom()方法。

    img
    • 1 次状态切换：调用方法后，用户态 -> 内核态
    • 3 次数据拷贝：上图中，1、3 为 DMA拷贝，2 为 CPU 拷贝

  3. sendFile：于 Linux2.4 优化，DMA直接将数据写入网卡，只会将一些 offset 和 length 信息拷入 socket 缓冲区，几乎无消耗。

    img
    1. 1 次状态切换：用户态 -> 内核态
    2. 2 次都是 DMA 拷贝，其中 2 是通过 Linux2.4 中的 scatter/gather 实现的 SG-DMA。

Netty 中的零拷贝

  1. CompositeByteBuf 类：将多个 ByteBuf 合并为逻辑上的 ByteBufe，避免 ByteBuf 之间的拷贝。
  2. byte 数组包装称 ByteBuf 对象，包装过程中不会产生内存拷贝。
  3. ByteBuf 支持 slice() 操作，因此可以将它分解为多个共享同一存储区域的 ByteBuf，避免内存拷贝。
  4. 通过 FileRegion 包装的 FileChannel.transferTo 实现文件传输，直接将文件缓冲区的数据发送到目标 channel，避免传统通过循环 write 方式导致的内存拷贝问题。

参考资料

更新于 2023-08-15 
CC BY-NC 4.0
NIOJavaNetWork
返回 | 主页
0%