JAVA 零拷贝技术和主流中间件零拷贝技术应用

- 介绍
- - Java代码里面有哪些零拷贝技术
  - java 中`文件读写`方式主要分为
  - 什么是`FileChannel`
  - - mmap实现
    - sendfile实现
- 文件IO实战
- - 需求
  - 代码编写实战
  - - IOTest.java 文件
    - 上传阿里云，测试运行代码看耗时
    - - 为啥带buffer的IO比普通IO性能高？
      - BufferedInputStream为啥性能高点
- 性能差异分析
- - 原理分析
- 中间件零拷贝的应用
- - Nginx 使用就是 sendfile 零拷贝
  - RocketMQ
  - 其它中间件
  - 优缺点

介绍

Java代码里面有哪些零拷贝技术

Java NIO对mmap -> fileChannel.map()
Java NIO对sendfile -> fileChannel.transferTo()和 fileChannel.transferFrom()
API是否使用零拷贝依赖于底层的系统实现

java 中`文件读写`方式主要分为

IO输入输出流，存在于 java.io 中【普通】

 public static void inputStream(String inputFilePathStr, String outputFilePathStr) {
        long start = System.currentTimeMillis();
        try (InputStream fis = new FileInputStream(inputFilePathStr);
             FileOutputStream fos = new FileOutputStream(outputFilePathStr);
        ) {
            byte[] buf = new byte[1024];
            int len = 0;
            while ((len = fis.read(buf)) != -1) {
                fos.write(buf);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        long end = System.currentTimeMillis();
        System.out.println(end - start);
    }

FileChannel 文件通道，存在于java.nio.channels.FileChannel 中【高级】

什么是`FileChannel`

是一个连接到文件的通道,可以通过文件通道读写文件，该常被用于高效的网络/文件的数据传输和大文件拷贝
应用程序使用FileChannel 写完以后，数据是在PageCache上的，操作系统不定时的把PageCache的数据写入到磁盘
- 使用 channel.force(true) 把文件相关的数据强制刷入磁盘上去，避免宕机数据丢失
使用之前必须先打开它，但是无法直接 new 一个 FileChannel
常规通过使用一个InputStream、OutputStream或RandomAccessFile来获取一个FileChannel实例

RandomAccessFile randomAccessFile = new RandomAccessFile("/usr/data/xdclass_nio-data.txt", "rw");
FileChannel inChannel = randomAccessFile.getChannel();

mmap实现

- map方法，把文件映射成内存映射文件
- `MappedByteBuffer`，是抽象类 也是ByteBuffer的子类  ，具体实现子类是DirectByteBuffer，可被通道进行读写
- 一次 map 大小要限制 2G 内，过大 map 会增加虚拟内存回收和重新分配的压力 ，直接报错
	- `FileChannel.java` 中的 `map` 对 `long size` 进行了限制，不能大于 `Integer.MAX_VALUE`，否则就报错
	- JDK 层的为何要限制，是因为底层 C++的类型，无符号int类型最大是2^31 -1， 2^31 -1 字节就是 2GB - 1B

在这里插入图片描述

//position: 文件开始
//size：映射的文件区域大小
//mode: 访问该内存映射文件的方式： READ_ONLY(只读) READ_WRITE(读写),PRIVATE(创建一个修改副本)
MappedByteBuffer map(int mode,long position,long size);

sendfile实现

将字节从此通道的文件传输到给定的可写入字节通道
返回值为真实拷贝的size，最大拷贝2G，超出2G的部分将丢弃

//position - 文件中的位置,从此位置开始传输，必须非负数
//count - 要传输的最大字节数,必须非负数
//target - 目标通道
//返回：实际已传输的字节数，可能为零
fileChannel.transferTo(long position, long count, WritableByteChannel target)

将字节从给定的可读取字节通道传输到此通道的文件中
对比从源通道读取并将内容写入此通道的循环语句相比，此方法更高效

//src - 源通道
//position - 文件中的位置，从此位置开始传输,必须非负数
//count - 要传输的最大字节数, 必须非负数
//返回：实际已传输的字节数，可能为零
fileChannel.transferFrom(ReadableByteChannel src, long position, long count)

注意
- 上述方法允许将一个通道连接到另一个通道，不需要在用户态和内核态来回复制，同时通道间的内核态数据也无需复制
- transferTo()只有源为FileChannel才支持transfer这种高效的复制方式，其他如SocketChannel都不支持transfer模式
- 一般可以做FileChannel->FileChannel 和 FileChannel->SocketChannel的transfer零拷贝

文件IO实战

需求

实现一个文件拷贝，对比不同IO方式性能差异,文件大小 200MB～5GB
编码类型
- 普通java的io流
- 普通java的带buffer的io
- 零拷贝实现之mmap的io
- 零拷贝实现之sendfile的io
运行环境
- 阿里云Linux CentOS7.X
- 安装JDK11 配置全局环境变量
  - 配置 vim /etc/profile
  - 环境变量立刻生效
    - source /etc/profile
    - 查看安装情况 java -version

JAVA_HOME=/usr/local/software/jdk11
CLASSPATH=$JAVA_HOME/lib/
PATH=$PATH:$JAVA_HOME/bin
export PATH JAVA_HOME CLASSPATH

代码编写实战

IOTest.java 文件

package net.demo;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;

/**
 *  java IOTest.java "io" "source" "target"
 */
public class IOTest {
    public static void main(String[] args) {
        String type = args[0];
        String inputFilePath = args[1];
        String outputFilePath = args[2];
        if ("io".equalsIgnoreCase(type)) {
            inputStreamCopyFile(inputFilePath, outputFilePath);

        } else if ("buffer".equalsIgnoreCase(type)) {
            bufferInputStreamCopyFile(inputFilePath, outputFilePath);

        } else if ("mmap".equalsIgnoreCase(type)) {
            mmapCopyFile(inputFilePath, outputFilePath);

        } else if ("sendfile".equalsIgnoreCase(type)) {
            sendfileCopyFile(inputFilePath, outputFilePath);

        }
    }

    private static void sendfileCopyFile(String inputFilePath, String outputFilePath) {
        long start = System.currentTimeMillis();

        try (
                FileChannel channelIn = new FileInputStream(inputFilePath).getChannel();
                FileChannel channelOut  = new FileOutputStream(outputFilePath).getChannel();
        ) {
            // 代码一：针对小于2GB的问题，返回值为真实拷贝的size,最大拷贝2G，超出2G的部分将丢弃，最终拷贝文件大小只有2GB多点
            // channelIn.transferTo(0, channelIn.size(), channelOut);

            //代码二：针对大于2GB的文件，方案
            //获取文件总大小
            long size = channelIn.size();

            for (long left = size; left > 0; ) {
                //transferSize所拷贝过去的真实长度，size - left计算出下次要拷贝的位置
                long transferSize = channelIn.transferTo((size - left), left, channelOut);
                System.out.println("总大小："+size+"，拷贝大小："+transferSize);
                //left剩余字节多少
                left = left - transferSize;
            }

        } catch (IOException e) {
            e.printStackTrace();
        }

        long end = System.currentTimeMillis();
        System.out.println("耗时："+(end - start));
    }

    private static void mmapCopyFile(String inputFilePath, String outputFilePath) {
        long start = System.currentTimeMillis();

        try (
                FileChannel channelIn = new FileInputStream(inputFilePath).getChannel();
                FileChannel channelOut = new RandomAccessFile(outputFilePath, "rw").getChannel();

        ) {
            long size = channelIn.size();
            System.out.println("mappedFile:" + size);
            MappedByteBuffer mbbi = channelIn.map(FileChannel.MapMode.READ_ONLY, 0, size);
            MappedByteBuffer mbbo = channelOut.map(FileChannel.MapMode.READ_WRITE, 0, size);
            for (int i = 0; i < size; i++) {
                byte b = mbbi.get(i);
                mbbo.put(i, b);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }

        long end = System.currentTimeMillis();
        System.out.println("耗时："+(end - start));
    }

    private static void bufferInputStreamCopyFile(String inputFilePath, String outputFilePath) {
        long start = System.currentTimeMillis();
        try (
                BufferedInputStream bis = new BufferedInputStream( new FileInputStream(inputFilePath));
                BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(outputFilePath));
        ) {

//            byte[] buf = new byte[64];
            byte[] buf = new byte[1];//方便测试字节改用1
            int len;
            while ((len = bis.read(buf)) != -1) {
                bos.write(buf);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        long end = System.currentTimeMillis();
        System.out.println("耗时："+(end - start));
    }

    private static void inputStreamCopyFile(String inputFilePath, String outputFilePath) {

        long start = System.currentTimeMillis();
        try (
                FileInputStream fis = new FileInputStream(inputFilePath);
                FileOutputStream fos = new FileOutputStream(outputFilePath)
        ) {

//            byte[] buf = new byte[64];
            byte[] buf = new byte[1];//方便测试字节改用1
            int len;
            while ((len = fis.read(buf)) != -1) {
                fos.write(buf);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        long end = System.currentTimeMillis();
        System.out.println("耗时："+(end - start));

    }
}

上传阿里云，测试运行代码看耗时

#释放所有缓存
echo 3 > /proc/sys/vm/drop_caches

#查看内存使用
free -h -w -s 1

#查看机器负载
top 

java IOTest.java "io" "/usr/local/software/slow.log" "/usr/local/software/iotest/slow.log"
java IOTest.java "buffer" "/usr/local/software/slow.log" "/usr/local/software/iotest/slow.log"
java IOTest.java "mmap" "/usr/local/software/slow.log" "/usr/local/software/iotest/slow.log"
java IOTest.java "sendfile" "/usr/local/software/slow.log" "/usr/local/software/iotest/slow.log"

局部性原理：指计算机在执行某个程序时,倾向于使用最近使用的数据
- 时间局部性：如果程序中的某条指令一旦被执行，则不久的将来该指令可能再次被执行
- 空间局部性：一旦程序访问了某个存储单元，在不久的将来，其附近的存储单元也最有可能被访问

为啥带buffer的IO比普通IO性能高？

文件读取，OS的做了什么优化操作
- 每次读数据的时候，系统根据局部性原理，通过 DMA 会读入更多的数据到内核缓冲区里面
- OS根据局部性原理会在一次 read()系统调用过程中预读更多的文件数据缓存在内核IO缓冲区中
- 当继续访问的文件数据在缓冲区中时便直接拷贝数据到进程缓冲区，避免了再次的低效率磁盘IO操作
- OS已经帮减少磁盘IO操作次数，提高了性能

BufferedInputStream为啥性能高点

通过减少系统调用次数来提高性能了IO性能，即减少CPU在内核态和用户态的上下文切换次数
在 kernel buffer 把数据拷贝到 user buffer 的时候，把数据多拷贝到 user buffer 中
比如
- 进程user buffer想要向内核态读取4个字节，但是内核态上面有8个字节数据，大方点都拷贝到user buffer里面
- 当进程user buffer下次要再读取4个字节的时候，因为数据已经在user buffer中了，就不需要上下文切换

性能差异分析

普通拷贝
- 普通java的io流【慢】1800秒
- 普通java的带buffer的io【快】80秒
零拷贝（1～2g文件差别不大）
- 零拷贝实现之mmap的io【快】30秒
- 零拷贝实现之sendfile的io【快】30秒

原理分析

mmap
sendfile

中间件零拷贝的应用

Nginx 使用就是 sendfile 零拷贝

Web Server 处理静态页面请求时，是从磁盘中读取网页的内容，所以选择这个
- 因为 sendfile不能在应用程序中修改数据，所以适合静态文件服务器或者是直接转发数据的代理服务器

RocketMQ

主要是mmap，也有小部分使用sendfile
rocketMQ在消息存盘和网络发送使用mmap, 单个CommitLog文件大小默认1GB
- 要在用户进程内处理数据，然后再发送出去的话，用户空间和内核空间的数据传输就是不可避免的

其它中间件

Kafka ：主要是sendfile，也有小部分使用mmap
- kafka 在客户端和 broker 进行数据传输时，broker 使用 sendfile 系统调用，类似【FileChannel.transferTo】 API，将磁盘文件读到 OS 内核缓冲区后，直接转到 socket buffer 进行网络发送，即 Linux 的 sendfile
Hadoop、Tomcat、Kafka、Netty、Zookeeper、Rabbitmq… 等都有用到零拷贝

优缺点

零拷贝的目标
- 解放CPU，避免CPU做太多事情
- 减少内存带宽占用
- 减少用户态和内核态上下文切换过多
- 在文件较小的时候 mmap 耗时更短，当文件较大时 sendfile 的方式最优
零拷贝方式对比
- sendfile
  - 无法在调用过程中修改数据，只适用于应用程序不需要对所访问数据进行处理修改情况
  - 场景
    - 比如 静态文件传输，MQ的Broker发送消息给消费者
    - 如果想要在传输过程中修改数据，可以使用mmap系统调用
  - 文件大小：适合大文件传输
  - 切换和拷贝：2次上下文切换，最少 2 次数据拷贝
- mmap
  - 在mmap调用可以在应用程序中直接修改Page Cache中的数据，使用的是mmap+write两步
    - 调用比sendfile成本高，但优于传统I/O的拷贝实现方式，虽然比 sendfile 多了上下文切换
    - 但用户空间与内核空间并不需要数据拷贝，在正确使用情况下并不比 sendfile 效率差
  - 场景
    - 多个线程以只读的方式同时访问一个文件， mmap 机制下多线程共享同一物理内存空间，节约内存
  - 文件大小：适合小数据量读写
  - 切换和拷贝：4 次上下文切换，3 次数据拷贝