体系结构

1563803296721.png

从图可以看出来，ByteBuf分类主要是三个维度：

• Unpooled和Pooled
• Unsafe和非Unsafe
• Heap和Direct

访问索引

从图示中可以容易理解ByteBuf的读写区域

1563801077089.png

内存分配

策略

1563805578302.png 1563805646415.png 1563805753773.png

可以看到在接口方法里面已经区分了是direct还是heap，在继承体系上区分了是Pooled还是Unpooled

而在具体的逻辑中，再根据系统底层是否支持Unsafe来决定是否要返回Unsafe的实现。

UnpooledByteBufAllocator

heap内存的分配

1563806190062.png

protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
    // 根据是否支持Unsafe与否来返回具体的UnpooledHeapByteBuf
    return PlatformDependent.hasUnsafe() ? new UnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity)
            : new UnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
}

初始化

UnpooledUnsafeHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
    // 直接去调用父类UnpooledHeapByteBuf的构造函数
    // 说明底层的内存分配是一样的，只不过数据操作的方式不一样，会用Unsafe的方式，更快，更直接。
    super(alloc, initialCapacity, maxCapacity);
}

protected UnpooledHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
    // 直接新建byte数组，长度为initialCapacity，readindex和writeindex为0
    this(alloc, new byte[initialCapacity], 0, 0, maxCapacity);
}

private UnpooledHeapByteBuf(
        ByteBufAllocator alloc, byte[] initialArray, int readerIndex, int writerIndex, int maxCapacity) {

    super(maxCapacity);

    ...

    this.alloc = alloc;
    // 保存上面的byte数组作为ByteBuf的存储
    setArray(initialArray);
    // read和write index归零
    setIndex(readerIndex, writerIndex);
}

getByte

Unsafe

// Unsafe的方式操作数据
protected byte _getByte(int index) {
    return UnsafeByteBufUtil.getByte(array, index);
}

static byte getByte(byte[] data, int index) {
    // 可以看到，这里直接通过byte数组的内存偏移量加上数组的index来获取数据
    return UNSAFE.getByte(data, BYTE_ARRAY_BASE_OFFSET + index);
}

非Unsafe

// 非Unsafe的方式操作数据
protected byte _getByte(int index) {
    return HeapByteBufUtil.getByte(array, index);
}

// 通过数组下标的方式来操作数组
static byte getByte(byte[] memory, int index) {
    return memory[index];
}

direct内存分配

1563807731297.png

protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
    // 根据底层对Unsafe的支持与否来决定最终的UnpooledDirectByteBuf
    ByteBuf buf = PlatformDependent.hasUnsafe() ?
            UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) :
            new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);

    return disableLeakDetector ? buf : toLeakAwareBuffer(buf);
}

初始化

static UnpooledUnsafeDirectByteBuf newUnsafeDirectByteBuf(
        ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
    if (PlatformDependent.useDirectBufferNoCleaner()) {
        return new UnpooledUnsafeNoCleanerDirectByteBuf(alloc, initialCapacity, maxCapacity);
    }
    return new UnpooledUnsafeDirectByteBuf(alloc, initialCapacity, maxCapacity);
}

非Unsafe

protected UnpooledDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
    super(maxCapacity);
    ...

    this.alloc = alloc;
    // NIO的方法，这里直接去堆外分配内存
    setByteBuffer(ByteBuffer.allocateDirect(initialCapacity));
}

private void setByteBuffer(ByteBuffer buffer) {
    ByteBuffer oldBuffer = this.buffer;
    // 而这里对老的buffer进行回收，TODO
    if (oldBuffer != null) {
        if (doNotFree) {
            doNotFree = false;
        } else {
            freeDirect(oldBuffer);
        }
    }
    // 将前面创建的堆外内存保存下来
    this.buffer = buffer;
    tmpNioBuf = null;
    capacity = buffer.remaining();
}

Unsafe

protected UnpooledUnsafeDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
    super(maxCapacity);
    ...

    this.alloc = alloc;
    // 直接堆外内存分配
    setByteBuffer(allocateDirect(initialCapacity), false);
}

// 跟非Unsafe类似
final void setByteBuffer(ByteBuffer buffer, boolean tryFree) {
    if (tryFree) {
        ByteBuffer oldBuffer = this.buffer;
        if (oldBuffer != null) {
            if (doNotFree) {
                doNotFree = false;
            } else {
                freeDirect(oldBuffer);
            }
        }
    }
    this.buffer = buffer;
    // Unsafe不一样的地方在于，需要知道buffer的内存偏移量，而这个偏移量在allocateDirect会生成
    memoryAddress = PlatformDependent.directBufferAddress(buffer);
    tmpNioBuf = null;
    capacity = buffer.remaining();
}

static long directBufferAddress(ByteBuffer buffer) {
    // 可以看到这个偏移量会去找ADDRESS_FIELD_OFFSET
    // final Field field = Buffer.class.getDeclaredField("address");
    // 而堆外内存的偏移量早在allocateDirect的时候就已经自动赋给buffer的address了，直接反射去拿就好
    // 这个address只代表堆外内存的偏移量，
    return getLong(buffer, ADDRESS_FIELD_OFFSET);
}

getByte

Unsafe

protected byte _getByte(int index) {
    return UnsafeByteBufUtil.getByte(addr(index));
}

long addr(int index) {
    // 根据上面保存下来的偏移量来去拿到index的地址
    return memoryAddress + index;
}

非Unsafe

protected byte _getByte(int index) {
    // 直接通过api来对buffer进行读取
    return buffer.get(index);
}

PooledByteBufAllocator

direct内存分配

protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
    // 拿到当前线程绑定的PoolThreadCache
    PoolThreadCache cache = threadCache.get();
    // 进而拿到PoolThreadCache的堆外内存
    PoolArena<ByteBuffer> directArena = cache.directArena;

    ByteBuf buf;
    if (directArena != null) {
        // 在当前线程绑定的堆外内存区域请求一块空间
        buf = directArena.allocate(cache, initialCapacity, maxCapacity);
    } else {
        if (PlatformDependent.hasUnsafe()) {
            buf = UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
        } else {
            buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
        }
    }

    return toLeakAwareBuffer(buf);
}