Buffer源碼深入分析

本機(jī)環(huán)境：
Linux 4.4.0-21-generic #37-Ubuntu SMP Mon Apr 18 18:33:37 UTC 2016 x86_64 x86_64 x86_64 GNU/Linux

Buffer

Buffer的類圖如下：

除了Boolean，其他基本數(shù)據(jù)類型都有對應(yīng)的Buffer，但是只有ByteBuffer才能和Channel交互。只有ByteBuffer才能產(chǎn)生Direct的buffer，其他數(shù)據(jù)類型的Buffer只能產(chǎn)生Heap類型的Buffer。ByteBuffer可以產(chǎn)生其他數(shù)據(jù)類型的視圖Buffer，如果ByteBuffer本身是Direct的，則產(chǎn)生的各視圖Buffer也是Direct的。

Direct和Heap類型Buffer的本質(zhì)

首選說說JVM是怎么進(jìn)行IO操作的。

JVM在需要通過操作系統(tǒng)調(diào)用完成IO操作，比如可以通過read系統(tǒng)調(diào)用完成文件的讀取。read的原型是：ssize_t read(int fd,void *buf,size_t nbytes)，和其他的IO系統(tǒng)調(diào)用類似，一般需要緩沖區(qū)作為其中一個參數(shù)，該緩沖區(qū)要求是連續(xù)的。

Buffer分為Direct和Heap兩類，下面分別說明這兩類buffer。

Heap

Heap類型的Buffer存在于JVM的堆上，這部分內(nèi)存的回收與整理和普通的對象一樣。Heap類型的Buffer對象都包含一個對應(yīng)基本數(shù)據(jù)類型的數(shù)組屬性（比如：final **[] hb），數(shù)組才是Heap類型Buffer的底層緩沖區(qū)。
但是Heap類型的Buffer不能作為緩沖區(qū)參數(shù)直接進(jìn)行系統(tǒng)調(diào)用，主要因?yàn)橄旅鎯蓚€原因。

• JVM在GC時可能會移動緩沖區(qū)（復(fù)制-整理），緩沖區(qū)的地址不固定。

• 系統(tǒng)調(diào)用時，緩沖區(qū)需要是連續(xù)的，但是數(shù)組可能不是連續(xù)的（JVM的實(shí)現(xiàn)沒要求連續(xù)）。

所以使用Heap類型的Buffer進(jìn)行IO時，JVM需要產(chǎn)生一個臨時Direct類型的Buffer，然后進(jìn)行數(shù)據(jù)復(fù)制，再使用臨時Direct的Buffer作為參數(shù)進(jìn)行操作系統(tǒng)調(diào)用。這造成很低的效率，主要是因?yàn)閮蓚€原因：

• 需要把數(shù)據(jù)從Heap類型的Buffer里面復(fù)制到臨時創(chuàng)建的Direct的Buffer里面。

• 可能產(chǎn)生大量的Buffer對象，從而提高GC的頻率。所以在IO操作時，可以通過重復(fù)利用Buffer進(jìn)行優(yōu)化。

Direct

Direct類型的buffer，不存在于堆上，而是JVM通過malloc直接分配的一段連續(xù)的內(nèi)存，這部分內(nèi)存成為直接內(nèi)存，JVM進(jìn)行IO系統(tǒng)調(diào)用時使用的是直接內(nèi)存作為緩沖區(qū)。
-XX:MaxDirectMemorySize，通過這個配置可以設(shè)置允許分配的最大直接內(nèi)存的大?。∕appedByteBuffer分配的內(nèi)存不受此配置影響）。
直接內(nèi)存的回收和堆內(nèi)存的回收不同，如果直接內(nèi)存使用不當(dāng)，很容易造成OutOfMemoryError。JAVA沒有提供顯示的方法去主動釋放直接內(nèi)存，sun.misc.Unsafe類可以進(jìn)行直接的底層內(nèi)存操作，通過該類可以主動釋放和管理直接內(nèi)存。同理，也應(yīng)該重復(fù)利用直接內(nèi)存以提高效率。

MappedByteBuffer和DirectByteBuffer之間的關(guān)系

This is a little bit backwards: By rights MappedByteBuffer should be a subclass of DirectByteBuffer, but to keep the spec clear and simple, and for optimization purposes, it's easier to do it the other way around.This works because DirectByteBuffer is a package-private class.（本段話摘自MappedByteBuffer的源碼）

實(shí)際上，MappedByteBuffer屬于映射buffer（自己看看虛擬內(nèi)存），但是DirectByteBuffer只是說明該部分內(nèi)存是JVＭ在直接內(nèi)存區(qū)分配的連續(xù)緩沖區(qū)，并不一是映射的。也就是說MappedByteBuffer應(yīng)該是DirectByteBuffer的子類，但是為了方便和優(yōu)化，把MappedByteBuffer作為了DirectByteBuffer的父類。另外，雖然MappedByteBuffer在邏輯上應(yīng)該是DirectByteBuffer的子類，而且MappedByteBuffer的內(nèi)存的GC和直接內(nèi)存的GC類似（和堆GC不同），但是分配的MappedByteBuffer的大小不受-XX:MaxDirectMemorySize參數(shù)影響。
MappedByteBuffer封裝的是內(nèi)存映射文件操作，也就是只能進(jìn)行文件IO操作。MappedByteBuffer是根據(jù)mmap產(chǎn)生的映射緩沖區(qū)，這部分緩沖區(qū)被映射到對應(yīng)的文件頁上，屬于直接內(nèi)存在用戶態(tài)，通過MappedByteBuffer可以直接操作映射緩沖區(qū)，而這部分緩沖區(qū)又被映射到文件頁上，操作系統(tǒng)通過對應(yīng)內(nèi)存頁的調(diào)入和調(diào)出完成文件的寫入和寫出。

MappedByteBuffer

通過FileChannel.map(MapMode mode,long position, long size)得到MappedByteBuffer，下面結(jié)合源碼說明MappedByteBuffer的產(chǎn)生過程。

FileChannel.map的源碼：

public MappedByteBuffer map(MapMode mode, long position, long size)
        throws IOException    {        ensureOpen();        if (position < 0L)            throw new IllegalArgumentException("Negative position");        if (size < 0L)            throw new IllegalArgumentException("Negative size");        if (position + size < 0)            throw new IllegalArgumentException("Position + size overflow");        //最大2G
        if (size > Integer.MAX_VALUE)            throw new IllegalArgumentException("Size exceeds Integer.MAX_VALUE");        int imode = -1;        if (mode == MapMode.READ_ONLY)
            imode = MAP_RO;        else if (mode == MapMode.READ_WRITE)
            imode = MAP_RW;        else if (mode == MapMode.PRIVATE)
            imode = MAP_PV;        assert (imode >= 0);        if ((mode != MapMode.READ_ONLY) && !writable)            throw new NonWritableChannelException();        if (!readable)            throw new NonReadableChannelException();        long addr = -1;        int ti = -1;        try {            begin();
            ti = threads.add();            if (!isOpen())                return null;            //size()返回實(shí)際的文件大小
            //如果實(shí)際文件大小不符合，則增大文件的大小，文件的大小被改變，文件增大的部分默認(rèn)設(shè)置為0。
            if (size() < position + size) { // Extend file size
                if (!writable) {                    throw new IOException("Channel not open for writing " +                        "- cannot extend file to required size");
                }                int rv;                do {                   //增大文件的大小
                    rv = nd.truncate(fd, position + size);
                } while ((rv == IOStatus.INTERRUPTED) && isOpen());
            }            //如果要求映射的文件大小為0，則不調(diào)用操作系統(tǒng)的mmap調(diào)用，只是生成一個空間容量為0的DirectByteBuffer
            //并返回
            if (size == 0) {
                addr = 0;                // a valid file descriptor is not required
                FileDescriptor dummy = new FileDescriptor();                if ((!writable) || (imode == MAP_RO))                    return Util.newMappedByteBufferR(0, 0, dummy, null);                else
                    return Util.newMappedByteBuffer(0, 0, dummy, null);
            }            //allocationGranularity的大小在我的系統(tǒng)上是4K
            //頁對齊，pagePosition為第多少頁
            int pagePosition = (int)(position % allocationGranularity);            //從頁的最開始映射
            long mapPosition = position - pagePosition;            //因?yàn)閺捻摰淖铋_始映射，增大映射空間
            long mapSize = size + pagePosition;            try {                // If no exception was thrown from map0, the address is valid
                //native方法，源代碼在openjdk/jdk/src/solaris/native/sun/nio/ch/FileChannelImpl.c,
                //參見下面的說明
                addr = map0(imode, mapPosition, mapSize);
            } catch (OutOfMemoryError x) {                // An OutOfMemoryError may indicate that we've exhausted memory
                // so force gc and re-attempt map
                System.gc();                try {
                    Thread.sleep(100);
                } catch (InterruptedException y) {
                    Thread.currentThread().interrupt();
                }                try {
                    addr = map0(imode, mapPosition, mapSize);
                } catch (OutOfMemoryError y) {                    // After a second OOME, fail
                    throw new IOException("Map failed", y);
                }
            }            // On Windows, and potentially other platforms, we need an open
            // file descriptor for some mapping operations.
            FileDescriptor mfd;            try {
                mfd = nd.duplicateForMapping(fd);
            } catch (IOException ioe) {                unmap0(addr, mapSize);                throw ioe;
            }            assert (IOStatus.checkAll(addr));            assert (addr % allocationGranularity == 0);            int isize = (int)size;
            Unmapper um = new Unmapper(addr, mapSize, isize, mfd);            if ((!writable) || (imode == MAP_RO)) {                return Util.newMappedByteBufferR(isize,
                                                 addr + pagePosition,
                                                 mfd,
                                                 um);
            } else {                return Util.newMappedByteBuffer(isize,
                                                addr + pagePosition,
                                                mfd,
                                                um);
            }
        } finally {
            threads.remove(ti);            end(IOStatus.checkAll(addr));
        }
    }

map0的源碼實(shí)現(xiàn)：

JNIEXPORT jlong JNICALLJava_sun_nio_ch_FileChannelImpl_map0(JNIEnv *env, jobject this,
                                     jint prot, jlong off, jlong len){    void *mapAddress = 0;
    jobject fdo = (*env)->GetObjectField(env, this, chan_fd);    //linux系統(tǒng)調(diào)用是通過整型的文件id引用文件的，這里得到文件id
    jint fd = fdval(env, fdo);    int protections = 0;    int flags = 0;    if (prot == sun_nio_ch_FileChannelImpl_MAP_RO) {
        protections = PROT_READ;
        flags = MAP_SHARED;
    } else if (prot == sun_nio_ch_FileChannelImpl_MAP_RW) {
        protections = PROT_WRITE | PROT_READ;
        flags = MAP_SHARED;
    } else if (prot == sun_nio_ch_FileChannelImpl_MAP_PV) {
        protections =  PROT_WRITE | PROT_READ;
        flags = MAP_PRIVATE;
    }    //這里就是操作系統(tǒng)調(diào)用了，mmap64是宏定義，實(shí)際最后調(diào)用的是mmap
    mapAddress = mmap64(        0,                    /* Let OS decide location */
        len,                  /* Number of bytes to map */
        protections,          /* File permissions */
        flags,                /* Changes are shared */
        fd,                   /* File descriptor of mapped file */
        off);                 /* Offset into file */

    if (mapAddress == MAP_FAILED) {        if (errno == ENOMEM) {            //如果沒有映射成功，直接拋出OutOfMemoryError
            JNU_ThrowOutOfMemoryError(env, "Map failed");            return IOS_THROWN;
        }        return handle(env, -1, "Map failed");
    }    return ((jlong) (unsigned long) mapAddress);
}

雖然FileChannel.map()的zise參數(shù)是long，但是size的大小最大為Integer.MAX_VALUE,也就是最大只能映射最大2G大小的空間。實(shí)際上操作系統(tǒng)提供的MMAP可以分配更大的空間，但是JAVA限制在2G，ByteBuffer等Buffer也最大只能分配2G大小的緩沖區(qū)。
MappedByteBuffer是通過mmap產(chǎn)生得到的緩沖區(qū)，這部分緩沖區(qū)是由操作系統(tǒng)直接創(chuàng)建和管理的，最后JVM通過unmmap讓操作系統(tǒng)直接釋放這部分內(nèi)存。

Haep****Buffer

下面以ByteBuffer為例，說明Heap類型Buffer的細(xì)節(jié)。
該類型的Buffer可以通過下面方式產(chǎn)生：

• ByteBuffer.allocate(int capacity)

• ByteBuffer.wrap(byte[] array)
  使用傳入的數(shù)組作為底層緩沖區(qū)，變更數(shù)組會影響緩沖區(qū)，變更緩沖區(qū)也會影響數(shù)組。

• ByteBuffer.wrap(byte[] array,int offset, int length)
  使用傳入的數(shù)組的一部分作為底層緩沖區(qū)，變更數(shù)組的對應(yīng)部分會影響緩沖區(qū)，變更緩沖區(qū)也會影響數(shù)組。

DirectByteBuffer

DirectByteBuffer只能通過ByteBuffer.allocateDirect(int capacity) 產(chǎn)生。
ByteBuffer.allocateDirect()源碼如下：

      public static ByteBuffer allocateDirect(int capacity) {        return new DirectByteBuffer(capacity);
    }

DirectByteBuffer()源碼如下：

    DirectByteBuffer(int cap) {                   // package-private

        super(-1, 0, cap, cap);        //直接內(nèi)存是否要頁對齊，我本機(jī)測試的不用
        boolean pa = VM.isDirectMemoryPageAligned();        //頁的大小，本機(jī)測試的是4K
        int ps = Bits.pageSize();        //如果頁對齊，則size的大小是ps+cap，ps是一頁，cap也是從新的一頁開始，也就是頁對齊了
        long size = Math.max(1L, (long)cap + (pa ? ps : 0));        //JVM維護(hù)所有直接內(nèi)存的大小，如果已分配的直接內(nèi)存加上本次要分配的大小超過允許分配的直接內(nèi)存的最大值會
        //引起GC，否則允許分配并把已分配的直接內(nèi)存總量加上本次分配的大小。如果GC之后，還是超過所允許的最大值，
        //則throw new OutOfMemoryError("Direct buffer memory");
        Bits.reserveMemory(size, cap);        long base = 0;        try {           //是吧，unsafe可以直接操作底層內(nèi)存
            base = unsafe.allocateMemory(size);
        } catch (OutOfMemoryError x) {、            //沒有分配成功，把剛剛加上的已分配的直接內(nèi)存的大小減去。
            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;
    }

unsafe.allocateMemory()的源碼在openjdk/src/openjdk/hotspot/src/share/vm/prims/unsafe.cpp中。具體的源碼如下：

UNSAFE_ENTRY(jlong, Unsafe_AllocateMemory(JNIEnv *env, jobject unsafe, jlong size))
  UnsafeWrapper("Unsafe_AllocateMemory");  size_t sz = (size_t)size;  if (sz != (julong)size || size < 0) {
    THROW_0(vmSymbols::java_lang_IllegalArgumentException());
  }  if (sz == 0) {    return 0;
  }
  sz = round_to(sz, HeapWordSize);  //最后調(diào)用的是 u_char* ptr = (u_char*)::malloc(size + space_before + space_after)，也就是malloc。
  void* x = os::malloc(sz, mtInternal);  if (x == NULL) {
    THROW_0(vmSymbols::java_lang_OutOfMemoryError());
  }  //Copy::fill_to_words((HeapWord*)x, sz / HeapWordSize);
  return addr_to_java(x);
UNSAFE_END

JVM通過malloc分配得到連續(xù)的緩沖區(qū)，這部分緩沖區(qū)可以直接作為緩沖區(qū)參數(shù)進(jìn)行操作系統(tǒng)調(diào)用。