文件操作之 FileChannel 与 mmap

方法	描述
open	创建 FileChannel
read/write	读写
force	强制刷盘
map	mmap 内存映射
transferTo/transferFrom	转入/转出通道
lock/tryLock	获取文件锁

1 2	`RandomAccessFile aFile = new RandomAccessFile("data/nio-data.txt", "rw"); FileChannel inChannel = aFile.getChannel();`

// 写
byte[] data = new byte[4096];
long position = 1024L;
// 指定 position 写入 4kb 的数据
fileChannel.write(ByteBuffer.wrap(data), position);
// 从当前文件指针的位置写入 4kb 的数据
fileChannel.write(ByteBuffer.wrap(data));

// 读
ByteBuffer buffer = ByteBuffer.allocate(4096);
long position = 1024L;
// 指定 position 读取 4kb 的数据
fileChannel.read(buffer,position)；
// 从当前文件指针的位置读取 4kb 的数据
fileChannel.read(buffer);

1 2	`srcChannel.transferTo(0, Integer.MAX_VALUE, dstChannel); srcChannel.transferFrom(fromChannel, 0, Integer.MAX_VALUE);`

1	`MappedByteBuffer map(int mode,long position,long size);`

FileChannel channel = rf.getChannel();
// 持有文件锁
FileLock lock = channel.lock(0L, 23L, false);
// 释放文件锁
lock.release();

// 普通 ByteBuffer
ByteBuffer buffer = ByteBuffer.allocate(10);

// 堆外内存
ByteBuffer buffer = ByteBuffer.allocateDirect(10);

// 直接 wrap
byte[] bytes = new byte[10];
ByteBuffer buffer = ByteBuffer.wrap(bytes);
// 等同于上面的方式
ByteBuffer buffer = ByteBuffer.wrap(bytes, 0, bytes.length);

1	`0 <= mark <= position <= limit <= capacity`

// 创建一个 ByteBuffer。 此刻 mark 为 -1， position 为 0， limit 和 capacity 都是 10
ByteBuffer buffer = ByteBuffer.allocate(10);
// capacity 是无法改变的， 可以通过方法改变 position 和 limit 的位置
buffer. position(2); // 改变 position 的位置到 2
buffer.limit(5); // 改变 limit 的位置到 5

ByteBuffer buffer = ByteBuffer.allocate(10); // mark = -1, position = 0
buffer.position(2);                          // mark = -1, position = 2
buffer.mark();                               // mark = 2,  position = 2
buffer.position(5);                          // mark = 2,  position = 5
buffer.reset();                              // mark = 2,  position = 2

// 初始化各个指针位置
ByteBuffer buffer = ByteBuffer.allocate(10); // mark = -1, position = 0, limit = 10
buffer.position(2);                          // mark = -1, position = 2, limit = 10
buffer.mark();                               // mark = 2,  position = 2, limit = 10
buffer.position(5);                          // mark = 2,  position = 5, limit = 10
buffer.limit(8);                             // mark = 2,  position = 5, limit = 8

1	`buffer.clear(); // mark = -1, position = 0, limit = 10`

1	`buffer.flip(); // mark = -1, position = 0, limit = 5`

1	`buffer.rewind(); // mark = -1, position = 0, limit = 8`

1	`buffer.compact(); // mark = -1, position = 3, limit = 10`

-	堆内内存	堆外内存
底层实现	数组，JVM堆内存	`unsafe.allocateMemory(size)` 分配直接内存
分配大小限制	`-Xms-Xmx` 限制	可以通过 `-XX:MaxDirectMemorySize` 从JVM层面限制，同时也受到物理内存限制
垃圾回收	gc 回收	DirectByteBuffer 不再被使用的时候，会发出内部的 cleaner 钩子，保险通过: `((DirectByteBuffer)buffer)).cleaner().clean()` 来手动回收
拷贝方式	用户态和内核态之间来回拷贝	内核态

public void readInOneThread() throws Exception {
    int bufferSize = 50 * 1024 * 1024;
    File file = new File("/demo.txt");
    FileChannel fileChannel = new RandomAccessFile(file, "rw").getChannel();
    ByteBuffer byteBuffer = ByteBuffer.allocate(bufferSize);
    fileChannel.read(byteBuffer);
}

// IOUtil#read
static int read(FileDescriptor var0, ByteBuffer var1, long var2, NativeDispatcher var4) throws IOException {
    if (var1.isReadOnly()) {
        throw new IllegalArgumentException("Read-only buffer");
    } else if (var1 instanceof DirectBuffer) { 
    // 直接内存走这里
        return readIntoNativeBuffer(var0, var1, var2, var4);
    } else {
      // 堆内内存走这里
        ByteBuffer var5 = Util.getTemporaryDirectBuffer(var1.remaining());

        int var7;
        try {
          // 将数据读取到直接内存里面
            int var6 = readIntoNativeBuffer(var0, var5, var2, var4);
            var5.flip();
            if (var6 > 0) {
                // 从直接内存拷贝到堆内内存
                var1.put(var5);
            }

            var7 = var6;
        } finally {
            Util.offerFirstTemporaryDirectBuffer(var5);
        }

        return var7;
    }
}

public class Util {
    private static ThreadLocal<Util.BufferCache> bufferCache;
    
    public static ByteBuffer getTemporaryDirectBuffer(int var0) {
        if (isBufferTooLarge(var0)) {
            return ByteBuffer.allocateDirect(var0);
        } else {
            // FOUCS ON THIS LINE
            Util.BufferCache var1 = (Util.BufferCache)bufferCache.get();
            ByteBuffer var2 = var1.get(var0);
            if (var2 != null) {
                return var2;
            } else {
                if (!var1.isEmpty()) {
                    var2 = var1.removeFirst();
                    free(var2);
                }

                return ByteBuffer.allocateDirect(var0);
            }
        }
    }
    // ..... 省略其他逻辑
}

public void directBufferCopy() throws Exception {
    File file = new File("/essd");
    FileChannel fileChannel = new RandomAccessFile(file, "rw").getChannel();
    ByteBuffer byteBuffer = ByteBuffer.allocate(50 * 1024 * 1024);
    ByteBuffer directByteBuffer = ByteBuffer.allocateDirect(4 * 1024);
    for (int i = 0; i < 12800; i++) {
        directByteBuffer.clear();
        // 读取到自己的堆外内存
        fileChannel.read(directByteBuffer, i * 4 * 1024);
        // 拷贝到堆内内存
        directByteBuffer.flip();
        byteBuffer.put(directByteBuffer);
    }
}

public static void systemGC() throws IOException, InterruptedException {
    ByteBuffer buffer = ByteBuffer.allocateDirect(1024 * 1024 * 1024);
    System.in.read();
    buffer = null;
    // 手动触发GC可以回收堆外内存
    System.gc();
    new CountDownLatch(1).await();
}

public static void cleanerGC() throws InterruptedException, IOException {
    ByteBuffer buffer = ByteBuffer.allocateDirect(1024 * 1024 * 1024);
    System.in.read();
    // 通过 cleaner 来回收
    ((DirectBuffer) buffer).cleaner().clean();
    new CountDownLatch(1).await();
}

1
2
3

void *mmap(void *start, size_t length, int prot, int flags, int fd, off_t offset);
int munmap( void * addr, size_t len);
int msync( void *addr, size_t len, int flags);

// 开启 mmap
FileChannel fileChannel = new RandomAccessFile(new File("test.log"), "rw").getChannel();
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, fileChannel.size());

// 写
byte[] data = new byte[4];
int position = 8;
// 从 mmap 当前指针写入 4b 数据
mappedByteBuffer.put(data);
// 指定 position 位置写入 4b 数据
MappedByteBuffer subMmap = (MappedByteBuffer) mappedByteBuffer.slice();
subMmap.position(position);
subMmap.put(data);

// 读
byte[] readData = new byte[4];
// 从当前指针读 4b
mappedByteBuffer.get(data);
// 从指定位置读 4b
subMmap.position(position);
subMmap.get(data);

FileChannel fileChannel = new RandomAccessFile(new File("test.log"), "rw").getChannel();
// 真正使用 MappedByteBuffer 进行读取了才会确保文件进入 pagecache
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 1);
for(int i = 0; i < _1GB; i += _4KB) {
  temp += map.get(i);
}

// 方法一， 一次写 4KB， 写 1GB
System.out.println(LocalDateTime.now());
FileChannel fileChannel = new RandomAccessFile(new File("test.log"), "rw").getChannel();
ByteBuffer byteBuffer = ByteBuffer.allocate(_4KB);
for (int i = 0; i < _4KB; i++) {
    byteBuffer.put((byte) 0);
}
for (int i = 0; i < _GB; i += _4KB) {
    byteBuffer.position(0);
    byteBuffer.limit(_4KB);
    fileChannel.write(byteBuffer);
}
System.out.println(LocalDateTime.now());


// 方法二， 一次写 1byte， 写 1GB
System.out.println(LocalDateTime.now());
FileChannel fileChannel = new RandomAccessFile(new File("test.log"), "rw").getChannel();
ByteBuffer byteBuffer = ByteBuffer.allocate(1);
byteBuffer.put((byte) 0);
for (int i = 0; i < _GB; i++) {
    byteBuffer.position(0);
    byteBuffer.limit(1);
    fileChannel.write(byteBuffer);
}
System.out.println(LocalDateTime.now());

System.out.println(LocalDateTime.now());
FileChannel fileChannel = new RandomAccessFile(new File("test.log"), "rw").getChannel();
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, _GB);
for (int i = 0; i < _GB; i++) {
    mappedByteBuffer.put((byte) 0);
}
System.out.println(LocalDateTime.now());

1	`((DirectByteBuffer) mmap).cleaner().clean()`