NIO工作方式浅析

java Socket 工作机制

• Socket是描述计算机之前相互通信的一种抽象功能。通过基于TCP/IP的流套接字协议建立连接
• A机器B机器通信—建立Socket连接—通过TCP连接（端口号指定唯一应用）----IP寻址（寻找唯一主机）----最终找到唯一主机上的唯一应用

建立通信链路

1. 客户端：

• 客户端建立链路之前需要有一个Socket实例，操作系统为Socket实例做如下操作

  • 分配一个没被使用的端口号
  • 创建包含本地地址 + 远程地址 + 端口号 的套接字数据结构

• 这个数据结构会一直保存在系统，直到链接关闭

• 创建Socket实例返回之前，需要进行TCP三次握手协议，完成TCP握手协议才算完成Socket创建

2. 服务端：

• 端口分配+监听：与客户的对应创建ServerSocket实例，只需要端口没被占就能创建成功
  • 操作系统页会创建底层数据接口包括：监听端口 + 监听地址通配符（例如 *，匹配所有）
• 阻塞+创建套接字数据结构：之后调用accept（），进入阻塞，等待客户端请求，当一个请求到了，操作系统将为这个连接创建一个套接字数据结构，包括如下信息
  • 请求源地址端口 + 请求源端口
• 完成TCP三次握手：将创建的套接字关联到ServerSocket实例的一个未完成的链接数据结构列表中（此时还未建立与客户端的Socket）
• 完成套接字数据状态转换：等到客户端与服务端TCP三次握手完成后，将这个ServerSocket才算创建完成，此时会将对应套接字数据结构从未完成列表移动到已完成列表
• 如上即与ServerSocket所关联的列表中每一个数据结构都代表与一个客户端建立的TCP链接

NIO工作方式

BIO瓶颈

• BIO即阻塞I/O，不管是磁盘还是网络I/O，数据写入OutputStream或者从InputStream读取都会阻塞，一旦阻塞，线程就失去了CPU的使用权，这在高并发情况下性能是无法接收到。
• 情况一在需要大量HTTP长连接的场景下，例如实时聊天，服务端不可能同时维护几百万用户的HTTP链接，因为这些链接并不是每时每刻都在传输数据。这种情况下保有大量HTTP是不可取的，即使线程数够，也会浪费大量的硬件机器资源，同时这种情况我们也无法判断那个HTTP的优先级更高，先处理哪个
• 另外一种情况客户端多线程并发获取服务端资源，此时需要同步服务端资源情况，这种同步情况比单线程复杂的多。

NIO工作机制

• NIO中关键类：Channel 和Selector 是NIO中核心概念
• Channel类比Socket，但是比Socket更具体，可以看成是数据的运输工具
• Selector类似运输工具的调度系统，控制Channel的状态，是正在发送，还是已经卸货等，他在轮询每个Channel的状态
• Buffer类 类比Stream，但是比Stream更具体，Buffer是数据的容器，但是粒度比Channel要小，一个Channel中可能有N个Buffer。
  • Stream不一样，他只代表一个大的数据容器，在数据装载之前并不知道任何信息，因为他都封装在运输工具Socket中了
• NIO引入Channel，Buffer，Selector就是将信息具体化，每个细节都开放接口处理给程序员控制。
• 有如下案例

/**
 * @author liaojiamin
 * @Date:Created in 15:11 2022/7/29
 */
public class ServerSelectorDemo {

    public void selector() throws IOException {
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        Selector selector = Selector.open();
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.socket().bind(new InetSocketAddress(8080));
        //注册监听事件
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        while (true){
            //获取所有key集合
            Set selectedKeys = selector.selectedKeys();
            Iterator iterator = selectedKeys.iterator();
            while (iterator.hasNext()){
                SelectionKey selectionKey = (SelectionKey) iterator.next();
                if((selectionKey.readyOps() & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT){
                    ServerSocketChannel serverSocketChannel1 = (ServerSocketChannel) selectionKey.channel();
                    //接受服务端请求
                    SocketChannel socketChannel = serverSocketChannel1.accept();
                    socketChannel.configureBlocking(false);
                    socketChannel.register(selector, SelectionKey.OP_READ);
                    iterator.remove();
                }else if ((selectionKey.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ){
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    while (true){
                        byteBuffer.clear();
                        //读取数据
                        int n = socketChannel.read(byteBuffer);
                        if(n <= 0){
                            break;
                        }
                        byteBuffer.flip();
                    }
                    iterator.remove();
                }
            }
        }

    }
}
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
45

• 如上代码中是基础的NIO模式：
  • 通过Selector静态工厂创建一个选择器
  • 创建一个服务端Channel，并且绑定到一个Socket对象，将将这个通信信道注册到选择器上
  • 设置是否阻塞模式，接着就能调用Selector的 selectedKeys检查这个已经注册的选择器上所有通信信道是否有需要的事情发生
  • 如果有某个指定类型事件，就返回所有的SelectionKey，通过这个对象的Channel可以获取到通信信道对象，从而读取通信的数据 – Buffer，这个Buffer是我们可以控制的缓冲器。
• Selector可以同时监听一组通信信道（Channel）上的I/O状态，前提是这个Selector已经注册到这些通信信道中

Buffer 工作方式

• Buffer中索引及说明
  • capacity：缓冲区数组总长度
  • position：下一个要操作的数据元素数据
  • limit：缓冲区数组中不可操作的下一个元素的位置， limit <= capacity
  • mark：用于记录当前position的前一个位置或者默认是0
• Buffer 通过ByteBuffer.allocate(11) 方式创建一个byte数组缓冲区，初始状态写入如下图，position位置0 ，capacity和limit默认都是数组长度，当写入11个byte时候，位置变化如下，代码如下

//初始化方法
ByteBuffer(int mark, int pos, int lim, int cap,   // package-private
                 byte[] hb, int offset)
    {
        super(mark, pos, lim, cap);
        this.hb = hb;
        this.offset = offset;
    }
//super是调用Byffer中的初始化方法
Buffer(int mark, int pos, int lim, int cap) {       // package-private
        if (cap < 0)
            throw new IllegalArgumentException("Negative capacity: " + cap);
        this.capacity = cap;
        limit(lim);
        position(pos);
        if (mark >= 0) {
            if (mark > pos)
                throw new IllegalArgumentException("mark > position: ("
                                                   + mark + " > " + pos + ")");
            this.mark = mark;
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

• 当写入4个byte后的示意图
  

• 当调用byteBuffer.flip(）方法时候，数组的状态如下
  

• 此时底层操作系统就可以从缓冲区中正确读取这5个字节数据，并且发送出去，在下一次写入数据之前我们在调用clear()方法，缓冲区的索引又会回到初始位置。

• mark的作用：当调用mark()方法时候，会记录当前position的前一个位置，我们需要调用reset时候，position恢复mark记录的值

allocate 与 allocateDirect的区别

• 通过allocate分配的内存，我们会通过Clannel获取I/O 数据，

  • 首先要通过操作系统Socket缓冲区：这个操作系统缓冲区就是底层TCO管理的RecvQ或者SendQ队列
  • 接着再将数据复制到Buffer

• 在这两个区域的复制过程（操作系统缓冲区 ----- 用户缓冲区）是很耗性能的

• 通过allocateDirect分配的内存：直接操作操作系统缓冲区

  • 方法返回与底层存储空间关联的缓冲区，通过Native代码操作费JVM堆内存
  • 缺点是必须每次创建或者释放都调用一次System.gc()
  • 如下allocateDirect初始化源码，都是用的unsafe 去分配的内存

DirectByteBuffer(int cap) {                   // package-private
        super(-1, 0, cap, cap);
        boolean pa = VM.isDirectMemoryPageAligned();
        int ps = Bits.pageSize();
        long size = Math.max(1L, (long)cap + (pa ? ps : 0));
        Bits.reserveMemory(size, cap);
        long base = 0;
        try {
            base = unsafe.allocateMemory(size);
        } catch (OutOfMemoryError x) {
            Bits.unreserveMemory(size, cap);
            throw x;
        }
        unsafe.setMemory(base, size, (byte) 0);
        if (pa && (base % ps != 0)) {
            // Round up to page boundary
            address = base + ps - (base & (ps - 1));
        } else {
            address = base;
        }
        cleaner = Cleaner.create(this, new Deallocator(base, size, cap));
        att = null;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

NIO数据访问方式

• NIO提供了比传统文件访问更好的方法，两个优化方法：FileChannel.transferTO，FileChannel.transferFrom， 另外一个是FileChannel.map

  • FileChannel.transferXXX：与传统访问文件方式比较，减少数据从内核到用户控件的复制过程。数据直接在内核空间中移动。

• 传统的数据访问方式
  

• FileChannel.transferXXX的访问方式
  

• 如上图中，可以看到，不管是读还是写的方式，都能减少用户地址空间到内核地址空间数据复制的这一个步骤

• FileChannel.map的方式，也同样的能按照一定大小块映射为内存区域。当范问这块内存的时候就是直接操作文件了。这样就省去了数据从内核到用户空间的复制

• 这种方式适合对大文件的只读操作。比如文件的MD5校验等，

• 如下一个实现案例。

/**
 * @author liaojiamin
 * @Date:Created in 14:09 2022/8/1
 */
public class FileChannelMapCopyFile {

    public static void main(String[] args) throws FileNotFoundException {
        int BUFFER_SIZE = 1024;
        String fileName = "E:\\learn\\问题汇总\\MYSQL.md";
        long fileLength = new File(fileName).length();
        int bufferCount = 1+ (int) (fileLength / BUFFER_SIZE);
        MappedByteBuffer[] byteBuffers = new MappedByteBuffer[bufferCount];
        long remaining = fileLength;

        String fileName_1 = "E:\\learn\\问题汇总\\MYSQL_1.md";
        FileOutputStream fileOutputStream = new FileOutputStream(fileName_1);
        FileChannel writeChannel = fileOutputStream.getChannel();

        for (int i = 0; i < bufferCount; i++) {
            RandomAccessFile file;
            try {
                file = new RandomAccessFile(fileName, "r");
                Integer size = (int)Math.min(remaining, BUFFER_SIZE);
                byteBuffers[i] = file.getChannel().map(FileChannel.MapMode.READ_ONLY, i * BUFFER_SIZE, size);
                ByteBuffer byteBuffers1 = byteBuffers[i].get(new byte[size]);
                byteBuffers1.flip();
                writeChannel.write(byteBuffers1);
            }catch (Exception e){
                e.printStackTrace();
            }
            remaining -= BUFFER_SIZE;
        }
    }
}
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