由浅入深Netty基础知识IO相关

向右看齐 2024-03-22 19:16 104阅读 0赞

#### 目录 ####

*  1 stream vs channel
 *  2 IO 模型
 *  3 零拷贝
 *   *  3.1 传统 IO 问题
     *  3.2 NIO 优化
 *  4 AIO
 *   *  4.1 文件 AIO
     *  4.2 守护线程
     *  4.3 网络 AIO

--------------------

## 1 stream vs channel ##

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

*  stream 不会自动缓冲数据，channel 会利用系统提供的发送缓冲区、接收缓冲区（更为底层）
 *  stream 仅支持阻塞 API，channel 同时支持阻塞、非阻塞 API，网络 channel 可配合 selector 实现多路复用
 *  二者均为全双工，即读写可以同时进行

## 2 IO 模型 ##

同步阻塞、同步非阻塞、同步多路复用、异步阻塞（没有此情况）、异步非阻塞

*  同步：线程自己去获取结果（一个线程）
 *  异步：线程自己不去获取结果，而是由其它线程送结果（至少两个线程）

当调用一次 channel.read 或 stream.read 后，会切换至操作系统内核态来完成真正数据读取，而读取又分为两个阶段，分别为：

*  等待数据阶段
 *  复制数据阶段

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

*  阻塞 IO

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

*  非阻塞 IO

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

*  多路复用

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

*  信号驱动
 *  异步 IO

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

*  阻塞 IO vs 多路复用

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

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

## 3 零拷贝 ##

### 3.1 传统 IO 问题 ###

传统的 IO 将一个文件通过 socket 写出

File f = new File("helloword/data.txt");
    RandomAccessFile file = new RandomAccessFile(file, "r");
    
    byte[] buf = new byte[(int)f.length()];
    file.read(buf);
    
    Socket socket = ...;
    socket.getOutputStream().write(buf);

内部工作流程是这样的：

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

1.  java 本身并不具备 IO 读写能力，因此 read 方法调用后，要从 java 程序的**用户态**切换至**内核态**，去调用操作系统（Kernel）的读能力，将数据读入**内核缓冲区**。这期间用户线程阻塞，操作系统使用 DMA（Direct Memory Access）来实现文件读，其间也不会使用 cpu
    
    > DMA 也可以理解为硬件单元，用来解放 cpu 完成文件 IO
2.  从**内核态**切换回**用户态**，将数据从**内核缓冲区**读入**用户缓冲区**（即 byte\[\] buf），这期间 cpu 会参与拷贝，无法利用 DMA
3.  调用 write 方法，这时将数据从**用户缓冲区**（byte\[\] buf）写入 **socket 缓冲区**，cpu 会参与拷贝
4.  接下来要向网卡写数据，这项能力 java 又不具备，因此又得从**用户态**切换至**内核态**，调用操作系统的写能力，使用 DMA 将 **socket 缓冲区**的数据写入网卡，不会使用 cpu

可以看到中间环节较多，java 的 IO 实际不是物理设备级别的读写，而是缓存的复制，底层的真正读写是操作系统来完成的

*  用户态与内核态的切换发生了 3 次，这个操作比较重量级
 *  数据拷贝了共 4 次

### 3.2 NIO 优化 ###

通过 DirectByteBuf

*  ByteBuffer.allocate(10) HeapByteBuffer 使用的还是 java 内存
 *  ByteBuffer.allocateDirect(10) DirectByteBuffer 使用的是操作系统内存

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

大部分步骤与优化前相同，不再赘述。唯有一点：java 可以使用 DirectByteBuf 将堆外内存映射到 jvm 内存中来直接访问使用

*  这块内存不受 jvm 垃圾回收的影响，因此内存地址固定，有助于 IO 读写
 *  java 中的 DirectByteBuf 对象仅维护了此内存的虚引用，内存回收分成两步
    
     *  DirectByteBuf 对象被垃圾回收，将虚引用加入引用队列
     *  通过专门线程访问引用队列，根据虚引用释放堆外内存
 *  减少了一次数据拷贝，用户态与内核态的切换次数没有减少

进一步优化（底层采用了 linux 2.1 后提供的 sendFile 方法），java 中对应着两个 channel 调用 transferTo/transferFrom 方法拷贝数据

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

1.  java 调用 transferTo 方法后，要从 java 程序的**用户态**切换至**内核态**，使用 DMA将数据读入**内核缓冲区**，不会使用 cpu
2.  数据从**内核缓冲区**传输到 **socket 缓冲区**，cpu 会参与拷贝
3.  最后使用 DMA 将 **socket 缓冲区**的数据写入网卡，不会使用 cpu

可以看到

*  只发生了一次用户态与内核态的切换
 *  数据拷贝了 3 次

进一步优化（linux 2.4）

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

1.  java 调用 transferTo 方法后，要从 java 程序的**用户态**切换至**内核态**，使用 DMA将数据读入**内核缓冲区**，不会使用 cpu
2.  只会将一些 offset 和 length 信息拷入 **socket 缓冲区**，几乎无消耗
3.  使用 DMA 将 **内核缓冲区**的数据写入网卡，不会使用 cpu

整个过程仅只发生了一次用户态与内核态的切换，数据拷贝了 2 次。所谓的【零拷贝】，并不是真正无拷贝，而是在不会拷贝重复数据到 jvm 内存中，零拷贝的优点有

*  更少的用户态与内核态的切换
 *  不利用 cpu 计算，减少 cpu 缓存伪共享
 *  零拷贝适合小文件传输

## 4 AIO ##

AIO 用来解决数据复制阶段的阻塞问题

*  同步意味着，在进行读写操作时，线程需要等待结果，还是相当于闲置
 *  异步意味着，在进行读写操作时，线程不必等待结果，而是将来由操作系统来通过回调方式由另外的线程来获得结果

> 异步模型需要底层操作系统（Kernel）提供支持
> 
>  *  Windows 系统通过 IOCP 实现了真正的异步 IO
>  *  Linux 系统异步 IO 在 2.6 版本引入，但其底层实现还是用多路复用模拟了异步 IO，性能没有优势

### 4.1 文件 AIO ###

先来看看 AsynchronousFileChannel

@Slf4j
    public class AioDemo1 {
        
        public static void main(String[] args) throws IOException {
        
            try{
        
                AsynchronousFileChannel s = 
                    AsynchronousFileChannel.open(
                    	Paths.get("1.txt"), StandardOpenOption.READ);
                ByteBuffer buffer = ByteBuffer.allocate(2);
                log.debug("begin...");
                s.read(buffer, 0, null, new CompletionHandler<Integer, ByteBuffer>() {
        
                    @Override
                    public void completed(Integer result, ByteBuffer attachment) {
        
                        log.debug("read completed...{}", result);
                        buffer.flip();
                        debug(buffer);
                    }
    
                    @Override
                    public void failed(Throwable exc, ByteBuffer attachment) {
        
                        log.debug("read failed...");
                    }
                });
    
            } catch (IOException e) {
        
                e.printStackTrace();
            }
            log.debug("do other things...");
            System.in.read();
        }
    }

输出

13:44:56 [DEBUG] [main] c.i.aio.AioDemo1 - begin...
    13:44:56 [DEBUG] [main] c.i.aio.AioDemo1 - do other things...
    13:44:56 [DEBUG] [Thread-5] c.i.aio.AioDemo1 - read completed...2
             +-------------------------------------------------+
             |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    +--------+-------------------------------------------------+----------------+
    |00000000| 61 0d                                           |a.              |
    +--------+-------------------------------------------------+----------------+

可以看到

*  响应文件读取成功的是另一个线程 Thread-5
 *  主线程并没有 IO 操作阻塞

### 4.2 守护线程 ###

默认文件 AIO 使用的线程都是守护线程，所以最后要执行 `System.in.read()` 以避免守护线程意外结束

### 4.3 网络 AIO ###

public class AioServer {
        
        public static void main(String[] args) throws IOException {
        
            AsynchronousServerSocketChannel ssc = AsynchronousServerSocketChannel.open();
            ssc.bind(new InetSocketAddress(8080));
            ssc.accept(null, new AcceptHandler(ssc));
            System.in.read();
        }
    
        private static void closeChannel(AsynchronousSocketChannel sc) {
        
            try {
        
                System.out.printf("[%s] %s close\n", Thread.currentThread().getName(), sc.getRemoteAddress());
                sc.close();
            } catch (IOException e) {
        
                e.printStackTrace();
            }
        }
    
        private static class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {
        
            private final AsynchronousSocketChannel sc;
    
            public ReadHandler(AsynchronousSocketChannel sc) {
        
                this.sc = sc;
            }
    
            @Override
            public void completed(Integer result, ByteBuffer attachment) {
        
                try {
        
                    if (result == -1) {
        
                        closeChannel(sc);
                        return;
                    }
                    System.out.printf("[%s] %s read\n", Thread.currentThread().getName(), sc.getRemoteAddress());
                    attachment.flip();
                    System.out.println(Charset.defaultCharset().decode(attachment));
                    attachment.clear();
                    // 处理完第一个 read 时，需要再次调用 read 方法来处理下一个 read 事件
                    sc.read(attachment, attachment, this);
                } catch (IOException e) {
        
                    e.printStackTrace();
                }
            }
    
            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
        
                closeChannel(sc);
                exc.printStackTrace();
            }
        }
    
        private static class WriteHandler implements CompletionHandler<Integer, ByteBuffer> {
        
            private final AsynchronousSocketChannel sc;
    
            private WriteHandler(AsynchronousSocketChannel sc) {
        
                this.sc = sc;
            }
    
            @Override
            public void completed(Integer result, ByteBuffer attachment) {
        
                // 如果作为附件的 buffer 还有内容，需要再次 write 写出剩余内容
                if (attachment.hasRemaining()) {
        
                    sc.write(attachment);
                }
            }
    
            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
        
                exc.printStackTrace();
                closeChannel(sc);
            }
        }
    
        private static class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object> {
        
            private final AsynchronousServerSocketChannel ssc;
    
            public AcceptHandler(AsynchronousServerSocketChannel ssc) {
        
                this.ssc = ssc;
            }
    
            @Override
            public void completed(AsynchronousSocketChannel sc, Object attachment) {
        
                try {
        
                    System.out.printf("[%s] %s connected\n", Thread.currentThread().getName(), sc.getRemoteAddress());
                } catch (IOException e) {
        
                    e.printStackTrace();
                }
                ByteBuffer buffer = ByteBuffer.allocate(16);
                // 读事件由 ReadHandler 处理
                sc.read(buffer, buffer, new ReadHandler(sc));
                // 写事件由 WriteHandler 处理
                sc.write(Charset.defaultCharset().encode("server hello!"), ByteBuffer.allocate(16), new WriteHandler(sc));
                // 处理完第一个 accpet 时，需要再次调用 accept 方法来处理下一个 accept 事件
                ssc.accept(null, this);
            }
    
            @Override
            public void failed(Throwable exc, Object attachment) {
        
                exc.printStackTrace();
            }
        }
    }

[39711012fcb745aa92949b5763e0e3a8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/0299696144d54ac3b65690f16f9baa3d.png
[af5d853781cb46b1b92ef27170e4d36a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/beef5ee93b9a4766bc9272969f0e6d49.png
[22c631b0768a43a8aa17481a2736a94d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/cda4081cd62e407f8af6d373497485ad.png
[20e342632bbf47d6810c3ee6e3bf3a49.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/ceb8474dfd8a42beaf1cd5204247f039.png
[4163efbc93f84bf4b4eeadb4abee59f3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/32d86581226c4735b589aeb7c53c1b91.png
[6b36ff94f221481aa0e373e2f8bab8a0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/0c74450e0dad4cff92fbc0086142d941.png
[eb8b84f60cbf4f298297b74d2b1740c6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/1a974fde083749df88ef39adb7a9d6f9.png
[9a2ffdf0a4904670a42de7896fd57890.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/4ccbeea9bb364e43af3128192d9955e6.png
[f504d2c52e854a9cb34f225451796145.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/f0a4b93953494d599d2592bfe5766bb1.png
[1e03cb81f1e748d491520497984a2ad6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/9bec4ea4fb5f4bfcbe0fd1a4940671ea.png
[b66bd20eb83d4a908023ed35fc1c1a90.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/64b14736ba5741378b359e657345e663.png
[888868cdf9354d58a314d937ac2b6a0c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/7e8d1538e0624623be3825d94a1388cf.png