关于linux零拷贝技术可以先看下前面一篇文章IO零拷贝，因为java里的零拷贝底层也是依赖的操作系统实现，需要说明下，Linux提供的零拷贝技术Java并不是全支持，只支持2种：mmap内存映射、sendfile，分别是由FileChannel.map()与FileChannel.transferTo()/transferFrom()实现。
涉及的类主要有FileChannel,MappedByteBuffer,DirectByteBuffer。
MappedByteBuffer
先看下ChannelFile的map方法：
public abstract MappedByteBuffer map(MapMode mode,
long position, long size)throws IOException;
- mode 限定内存映射区域(MappedByteBuffer)对内存映像文件的访问模式，有只读，读写与写时拷贝三种。
- position 文件映射的起始地址，对应内存映射区域的首地址
- size 文件映射的字节长度，从position往后的字节数，对应内存映射区域的大小
- force(),对于处于READ_WRITE模式下的缓冲区，将对缓冲区内容个性强制刷新到本地文件
- load(),将缓冲区的内容载入物理内存中，并返回这个缓冲区的引用
- isLoad(),判断缓冲区的内容是否在物理内存中，是返回true,不是返回false
public class MappedByteBufferDemo {
public static final String CONTENT = "zero copy by MappedByteBuffer";
public static final String FILE_NAME= "zero_copy/mmap.txt";
public static final String CHARSET = "UTF-8";
/**
* 写文件数据：打开文件通道 fileChannel 并提供读权限、写权限和数据清空权限，
* 通过 fileChannel 映射到一个可写的内存缓冲区 mappedByteBuffer，
* 将目标数据写入 mappedByteBuffer，通过 force() 方法把缓冲区更改的内容强制写入本地文件。
*/
@Test
public void writeToFileByMappedByteBuffer(){
//文件路径根据实际来定，我是放在项目的resources目录下
Path path = Paths.get(getClass().getResource("/"+FILE_NAME).getPath());
byte[] bytes = CONTENT.getBytes(Charset.forName(CHARSET));
try(FileChannel fileChannel = FileChannel.open(path, StandardOpenOption.READ,
StandardOpenOption.WRITE,StandardOpenOption.TRUNCATE_EXISTING)){
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, bytes.length);
if (mappedByteBuffer != null){
mappedByteBuffer.put(bytes);
mappedByteBuffer.force();
}
}catch (IOException e){
e.printStackTrace();
}
}/**
*
* 读文件数据：打开文件通道 fileChannel 并提供只读权限，通过 fileChannel 映射到一个
* 只可读的内存缓冲区 mappedByteBuffer，读取 mappedByteBuffer 中的字节数组即可得到文件数据。
*/
@Test
public void readFileFromMappedByteBuffer(){
Path path = Paths.get(getClass().getResource("/"+FILE_NAME).getPath());
int length = CONTENT.getBytes(Charset.forName(CHARSET)).length;
try(FileChannel fileChannel = FileChannel.open(path,StandardOpenOption.READ)){
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_ONLY, 0, length);
if (mappedByteBuffer != null){
byte[] bytes = new byte[length];
mappedByteBuffer.get(bytes);
String content = new String(bytes, StandardCharsets.UTF_8);
assertEquals(content,"zero copy by MappedByteBuffer");
}
}catch (IOException e){
e.printStackTrace();
}
}
}这里我们再来看看map()方法，它是在FileChannelImpl类里实现的，来看下核心代码：
public MappedByteBuffer map(MapMode mode, long position, long size) throws IOException {
int pagePosition = (int)(position % allocationGranularity);
long mapPosition = position - pagePosition;
long mapSize = size + pagePosition;
try {
//第一次文件映射导致OOM，手动触发垃圾回收，100S后再尝试映射，如果再失败则抛出异常
addr = map0(imode, mapPosition, mapSize);
} catch (OutOfMemoryError x) {
System.gc();
try {
Thread.sleep(100);
} catch (InterruptedException y) {
Thread.currentThread().interrupt();
}
try {
//addr为内存映射区域的起始地址，通过起始地址+偏移量可以获取指定内存数据。底层是JNI调用C实现
addr = map0(imode, mapPosition, mapSize);
} catch (OutOfMemoryError y) {
throw new IOException("Map failed", y);
}
}int isize = (int)size;
Unmapper um = new Unmapper(addr, mapSize, isize, mfd);
//通过Util工人反射创建一个DirectByteBuffer实例
if ((!writable) || (imode == MAP_RO)) {
return Util.newMappedByteBufferR(isize, addr + pagePosition, mfd, um);
} else {
return Util.newMappedByteBuffer(isize, addr + pagePosition, mfd, um);
}
}
总结：
- MappedByteBuffer底层使用DirectByteBuffer申请堆外虚拟内存，分配的内存不受JVM的-Xmx限制
- MappedByteBuffer打开的文件只有在垃圾回收的时候才会被关闭
- MappedByteBuffer映射的内存需要用户程序通过java反射调用sum.misc.Cleaner的clean()方法手动释放
DirectByteBuffer可以分配堆外内存，它是通过Unsafe本地方法allocateMemory()进行分配的，底层调用的是操作系统的malloc()函数。创建DirectByteBuffer对象时还会创建一个Deallocate线程，并通过Cleaner的freeMemory()方法对直接内存进行回收操作，freeMomery()底层调用的是操作系统的free()函数。
DirectByteBuffer(int cap) {
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)) {
address = base + ps - (base & (ps - 1));
} else {
address = base;
}
cleaner = Cleaner.create(this, new Deallocator(base, size, cap));
att = null;
}
DirectByteBuffer是MappedByteBuffer的子类，之前我们有提到的FileChannel#map()方法中
Util.newMappedByteBuffer(isize, addr + pagePosition, mfd, um);
它底层就是通过反射创建了DirectByteBuffer实例，然后分配的堆外内存：
static MappedByteBuffer newMappedByteBuffer(int size, long addr, FileDescriptor fd,
Runnable unmapper) {
MappedByteBuffer dbb;
if (directByteBufferConstructor == null)
initDBBConstructor();
try {
dbb = (MappedByteBuffer)directByteBufferConstructor.newInstance(
new Object[] { new Integer(size), new Long(addr), fd, unmapper });
} catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
throw new InternalError(e);
}
return dbb;
}private static void initDBBRConstructor() {
AccessController.doPrivileged(new PrivilegedAction() {
public Void run() {
try {
Class cl = Class.forName("java.nio.DirectByteBufferR");
Constructor ctor = cl.getDeclaredConstructor(
new Class[] { int.class, long.class, FileDescriptor.class,
Runnable.class });
ctor.setAccessible(true);
directByteBufferRConstructor = ctor;
} catch (ClassNotFoundException | NoSuchMethodException |
IllegalArgumentException | ClassCastException x) {
throw new InternalError(x);
}
return null;
}});
} DirectByteBuffer本身也有文件内存映射的功能，另外还提供了MappedByteBuffer所没有的可以在内存映像文件进行随机读取get()与写入write()操作。
public byte get() {
return ((unsafe.getByte(ix(nextGetIndex()))));
}public byte get(int i) {
return ((unsafe.getByte(ix(checkIndex(i)))));
}public ByteBuffer put(byte x) {
unsafe.putByte(ix(nextPutIndex()), ((x)));
return this;
}public ByteBuffer put(int i, byte x) {
unsafe.putByte(ix(checkIndex(i)), ((x)));
return this;
}
内存映像文件的随机读写都是借助 ix() 方法实现定位的， ix() 方法通过内存映射空间的内存首地址（address）和给定偏移量 i 计算出指针地址，然后由 unsafe 类的 get() 和 put() 方法和对指针指向的数据进行读取或写入。
总结：
- DirectByteBuffer是MappedByteBuffer子类，它本身有文件映射内存功能，同时它还具有MappedByteBuffer所没有的在内存映像文件进行随机读取get()与写入write()功能。
- DirectByteBuffer是通过Unsafe本地方法申请的堆外内存，回收时需要应用程序本身使用Cleaner类进行回收
FileChannel是一个用于文件读写，映射和操作的通道，它定义了transferFrom()和transferTo()两个抽象方法，它通过在通道和通道之间建立连接实现数据传输。
//通过FileChannel将文件里面的数据写入一个WritableByteChannel的目的通道
public abstract long transferTo(long position, long count, WritableByteChannel target)
throws IOException;
//将一个源通道ReadableByteChannel中的数据读取到当前FileChannel的文件里面
public abstract long transferFrom(ReadableByteChannel src, long position, long count)
throws IOException;
示例：
public class FileChannelDemo {
public static final String CONTENT = "zero copy by FileChannel";
//两个文件放在项目的resources目录下
public static final String SOURCE_FILE = "/zero_copy/source.txt";
public static final String TARGET_FILE = "/zero_copy/target.txt";
public static final String CHARSET = "UTF-8";
//先将内容写入source.txt
@Before
public void setup(){
Path path = Paths.get(getClass().getResource(SOURCE_FILE).getPath());
byte[] bytes = CONTENT.getBytes(Charset.forName(CHARSET));
try(FileChannel fileChannel = FileChannel.open(path, StandardOpenOption.READ,
StandardOpenOption.WRITE,StandardOpenOption.TRUNCATE_EXISTING)){
fileChannel.write(ByteBuffer.wrap(bytes));
}catch (IOException e){
e.printStackTrace();
}
}//通过transferTo将fromChannel上的数据拷贝到toChannel
@Test
public void transferTo()throws Exception{
try(FileChannel fromChannel = new RandomAccessFile(getClass().getResource(SOURCE_FILE).getPath()
,"rw").getChannel();
FileChannel toChannel = new RandomAccessFile(getClass().getResource(TARGET_FILE).getPath()
,"rw").getChannel()){
long position = 0L;
long offset = fromChannel.size();
fromChannel.transferTo(position,offset,toChannel);
}
}
//通过transferFrom将fromChannel中的数据拷贝到toChannel
@Test
public void transferFrom()throws Exception{
try(FileChannel fromChannel = new RandomAccessFile(getClass().getResource(SOURCE_FILE).getPath()
,"rw").getChannel();
FileChannel toChannel = new RandomAccessFile(getClass().getResource(TARGET_FILE).getPath()
,"rw").getChannel()){
long position = 0L;
long offset = fromChannel.size();
toChannel.transferFrom(fromChannel,position,offset);
}
}
}transferTo()与transferFrom()底层都是基于sendfile实现数据传输的。下面以transferTo()源码为例进行说明：
public long transferTo(long position, long count, WritableByteChannel target)
throws IOException {
// 计算文件的大小
long sz = size();
// 校验起始位置
if (position > sz)
return 0;
int icount = (int)Math.min(count, Integer.MAX_VALUE);
// 校验偏移量
if ((sz - position) < icount)
icount = (int)(sz - position);
long n;
//内核如果支持sendfile，则使用transferToDirectly
if ((n = transferToDirectly(position, icount, target)) >= 0)
return n;
//内核不支持sendfile，则使用mmap的方式
if ((n = transferToTrustedChannel(position, icount, target)) >= 0)
return n;
//内核不支持sendfile与mmap，则使用传统的IO方式完成读写
return transferToArbitraryChannel(position, icount, target);
}
//先获取文件描述符targetFD，接着获取同步锁后执行transferToDirectlyInternal
private long transferToDirectly(long position, int icount, WritableByteChannel target)
throws IOException {
// 省略从target获取targetFD的过程
if (nd.transferToDirectlyNeedsPositionLock()) {
synchronized (positionLock) {
long pos = position();
try {
return transferToDirectlyInternal(position, icount,
target, targetFD);
} finally {
position(pos);
}
}
} else {
return transferToDirectlyInternal(position, icount, target, targetFD);
}
}
//transferToDirectlyInternal会调用本地方法transferTo0()尝试以sendfile的方式传输数据
private long transferToDirectlyInternal(long position, int icount,
WritableByteChannel target,
FileDescriptor targetFD) throws IOException {
assert !nd.transferToDirectlyNeedsPositionLock() ||
Thread.holdsLock(positionLock);
long n = -1;
int ti = -1;
try {
begin();
ti = threads.add();
if (!isOpen())
return -1;
do {
n = transferTo0(fd, position, icount, targetFD);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
if (n == IOStatus.UNSUPPORTED_CASE) {
if (target instanceof SinkChannelImpl)
pipeSupported = false;
if (target instanceof FileChannelImpl)
fileSupported = false;
return IOStatus.UNSUPPORTED_CASE;
}
if (n == IOStatus.UNSUPPORTED) {
transferSupported = false;
return IOStatus.UNSUPPORTED;
}
return IOStatus.normalize(n);
} finally {
threads.remove(ti);
end (n > -1);
}
}总结：
- FileChannel的transferTo()与transferFrom()底层都是基于sendfile实现数据传输的
- transferTo方法内部会判断系统是否支持sendfile，如果不支持会使用mmap的方式；如果系统也不支持mmap的方式，则会使用传统的IO方式进行数据传输
- transferToyyfi会调用本地方法transferTo0()
netty的零拷贝主要是通过对java.nio.channels.FileChannel的tranferTo()的包装，在文件传输时将文件缓冲区的数据直接发送到目的通道(Channel)。
- 使用Direct Buffers，Netty采用直接缓冲区直接在内存区域分配空间，避免数据的多次拷贝
- 使用CompositeByteBuf，它保存了多个ByteBuf的引用，对外提供统一封装后的ByteBuf接口，避免数据拷贝
- Netty的文件传输类DefaultFileRegion通过调用FileChannel.transferTo()方法实现零拷贝，文件缓冲区的数据会直接发送给目标Channel
mmap使用的是非阻塞式IO，基于多路复用处理，适用于小数据块/高频率的IO传输，大块数据会阻塞多路复用线程，sendfile使用的是阻塞式IO，适用于大数据块/低频率的IO传输。
| 零拷贝方式 | 优点 | 缺点 | |
|---|---|---|---|
| RocketMQ | mmap+write | 适用于小块文件传输，频繁调用时效率高 | 不能很好利用DMA方式，会比sendfile多消耗CPU，内存安全性控制复杂，需要避免JVM Crash问题 |
| Kafka | sendfile | 可以利用DMA方式，消耗CPU较少，大块文件传输效率高，无内存安全问题 | 小块文件效率低于mmap方式，只能是BIO方式传输，不能使用NIO方式 |
- FileChannel调用map()方法最终是使用DirectByteBuffer映射的堆外内存，然后使用MappedByteBuffer进行读写，这就是mmap实现的方式
- FileChannel调用transferTo()方法时底层使用本地方法transferTo0()实现的sendfile方法
- Netty,RocketMQ,Kafka里的零拷贝也简单提了一下
Java NIO-零拷贝实现
【Java里的零拷贝】深入剖析Linux IO原理和几种零拷贝机制的实现
