写自己的netty源码分析--ByteBuf

绪论三个问题内存的类别有哪些？如何减少多线程内存分配之间的竞争？不同大小的内存是如何进行分配的？主要内容内存与内存管理器的抽象不同规格大小和不同类别的内存的分配策略内存的回收过程ByteBuf结构与重要API内存结构该结构其实是一个数组，有两个索引 readerIndex 和writerIndex，支持索引访问。*+------------...

灯塔先生

440人浏览 · 2019-07-04 16:59:04

灯塔先生 · 2019-07-04 16:59:04 发布

绪论

三个问题

内存的类别有哪些？
如何减少多线程内存分配之间的竞争？
不同大小的内存是如何进行分配的？

主要内容

内存与内存管理器的抽象
不同规格大小和不同类别的内存的分配策略
内存的回收过程

ByteBuf结构与重要API

内存结构

该结构其实是一个数组，有两个索引 readerIndex 和writerIndex，支持索引访问。

*      +-------------------+------------------+------------------+
*      | discardable bytes |  readable bytes  |  writable bytes  |
*      |                   |     (CONTENT)    |                  |
*      +-------------------+------------------+------------------+
*      |                   |                  |                  |
*      0      <=      readerIndex   <=   writerIndex    <=    capacity

上述结构是在执行discardReadBytes()之前，而在执行discardReadBytes()之后结构变为：

*  AFTER discardReadBytes()
*
*      +------------------+--------------------------------------+
*      |  readable bytes  |    writable bytes (got more space)   |
*      +------------------+--------------------------------------+
*      |                  |                                      |
* readerIndex (0) <= writerIndex (decreased)        <=        capacity

在执行clear()之后变为：

*  AFTER clear()
*      +---------------------------------------------------------+
*      |             writable bytes (got more space)             |
*      +---------------------------------------------------------+
*      |                                                         |
*      0 = readerIndex = writerIndex            <=            capacity

重要API

public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
    public abstract int capacity();
    public abstract int maxCapacity();
    public abstract boolean isDirect();

    public abstract ByteBuf readerIndex(int readerIndex); 读当前位置索引
    public abstract ByteBuf writerIndex(int writerIndex); 写当前位置索引

    public abstract ByteBuf markReaderIndex(); 标记当前读索引
    public abstract ByteBuf resetReaderIndex(); 重置到之前标记的读索引

    public abstract ByteBuf markWriterIndex(); 标记当前写索引
    public abstract ByteBuf resetWriterIndex(); 重置到之前标记的写索引
    ......
}

ByteBuf分类

Pooled和Unpooled
Unsafe和非Unsafe
Heap和Direct

ByteBuf (io.netty.buffer)
    AbstractByteBuf (io.netty.buffer)
        AbstractDerivedByteBuf (io.netty.buffer)
            ReadOnlyByteBuf (io.netty.buffer)
            DuplicatedByteBuf (io.netty.buffer)
                UnpooledDuplicatedByteBuf (io.netty.buffer)
                    PooledNonRetainedDuplicateByteBuf in AbstractPooledDerivedByteBuf (io.netty.buffer)
            AbstractUnpooledSlicedByteBuf (io.netty.buffer)
                UnpooledSlicedByteBuf (io.netty.buffer)
                    PooledNonRetainedSlicedByteBuf in AbstractPooledDerivedByteBuf (io.netty.buffer)
                SlicedByteBuf (io.netty.buffer)
        AbstractReferenceCountedByteBuf (io.netty.buffer)
            AbstractPooledDerivedByteBuf (io.netty.buffer)
                PooledDuplicatedByteBuf (io.netty.buffer)
                PooledSlicedByteBuf (io.netty.buffer)
            CompositeByteBuf (io.netty.buffer)
                WrappedCompositeByteBuf (io.netty.buffer)
                    SimpleLeakAwareCompositeByteBuf (io.netty.buffer)
                        AdvancedLeakAwareCompositeByteBuf (io.netty.buffer)
            ReadOnlyByteBufferBuf (io.netty.buffer)
                ReadOnlyUnsafeDirectByteBuf (io.netty.buffer)
            UnpooledUnsafeDirectByteBuf (io.netty.buffer)
                UnpooledUnsafeNoCleanerDirectByteBuf (io.netty.buffer)
                    InstrumentedUnpooledUnsafeNoCleanerDirectByteBuf in UnpooledByteBufAllocator (io.netty.buffer)
                WrappedUnpooledUnsafeDirectByteBuf (io.netty.buffer)
                ThreadLocalUnsafeDirectByteBuf in ByteBufUtil (io.netty.buffer)
                InstrumentedUnpooledUnsafeDirectByteBuf in UnpooledByteBufAllocator (io.netty.buffer)
            FixedCompositeByteBuf (io.netty.buffer)
            PooledByteBuf (io.netty.buffer)
                PooledUnsafeDirectByteBuf (io.netty.buffer)
                PooledHeapByteBuf (io.netty.buffer)
                    PooledUnsafeHeapByteBuf (io.netty.buffer)
                PooledDirectByteBuf (io.netty.buffer)
            UnpooledDirectByteBuf (io.netty.buffer)
                ThreadLocalDirectByteBuf in ByteBufUtil (io.netty.buffer)
                InstrumentedUnpooledDirectByteBuf in UnpooledByteBufAllocator (io.netty.buffer)
            UnpooledHeapByteBuf (io.netty.buffer)
                UnpooledUnsafeHeapByteBuf (io.netty.buffer)
                    InstrumentedUnpooledUnsafeHeapByteBuf in UnpooledByteBufAllocator (io.netty.buffer)
                InstrumentedUnpooledHeapByteBuf in UnpooledByteBufAllocator (io.netty.buffer)

Pooled和Unpooled

pooled：从预先分配好的一块内存里取一段连续内存封装成ByteBuf分配给应用程序；

unpooled：内存分配时直接调用系统API直接向操作系统申请内存；

Unsafe和非Unsafe

JDK底层有Unsafe对象，可直接拿到对象的内存地址进行一些读写操作；

非Unsafe：不依赖于JDK底层Unsafe对象；

Unsafe例子：

进入类PooledUnsafeHeapByteBuf中的_getByte()方法：
protected byte _getByte(int index) {
        return UnsafeByteBufUtil.getByte(memory, idx(index));
    }
static byte getByte(byte[] array, int index) {
        return PlatformDependent.getByte(array, index);
    }
public static byte getByte(byte[] data, int index) {
        return PlatformDependent0.getByte(data, index);
    }
static byte getByte(byte[] data, int index) {
        return UNSAFE.getByte(data, BYTE_ARRAY_BASE_OFFSET + index);
    }

非Unsafe例子：

进入类PooledHeapByteBuf中的_getByte()方法：
protected byte _getByte(int index) {
        return HeapByteBufUtil.getByte(memory, idx(index));
    }
static byte getByte(byte[] memory, int index) {
        return memory[index];
    }

Heap和Direct

Heap：直接在heap上进行内存分配；需要由GC 管理，内存自动释放；

Direct：直接调用JDK的API进行内存分配；不受GUN控制，不会参与GC 的过程，内存需要手动释放；

heap例子：

进入类UnpooledHeapByteBuf中的_getByte()方法：
protected byte _getByte(int index) {
        return HeapByteBufUtil.getByte(array, index);
    }
static byte getByte(byte[] memory, int index) {
        return memory[index];
    }

Direct例子：

进入类UnpooledDirectByteBuf中：

private ByteBuffer buffer;
基于buffer,而ByteBuffer指定是DirectByteBuffer

进入getByte()方法：
public byte getByte(int index) {
    ensureAccessible();
    return _getByte(index);
}
protected byte _getByte(int index) {
    return buffer.get(index);
}

接下来查看buffer是如何初始化的，netty中有一个内存管理工具--Unpooled类，接下来查看一下它的源码

进入类Unpooled中的directBuffer()方法：
public static ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
        return ALLOC.directBuffer(initialCapacity, maxCapacity);
    }
进入类AbstractByteBufAllocator中：
public ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
        if (initialCapacity == 0 && maxCapacity == 0) {
            return emptyBuf;
        }
        validate(initialCapacity, maxCapacity);
        return newDirectBuffer(initialCapacity, maxCapacity);
    }
进入newDirectBuffer()方法：
protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
        final ByteBuf buf;
        if (PlatformDependent.hasUnsafe()) {
            buf = noCleaner ? new InstrumentedUnpooledUnsafeNoCleanerDirectByteBuf(this, initialCapacity, maxCapacity) :
                    new InstrumentedUnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
        } else {
            buf = new InstrumentedUnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
        }
        return disableLeakDetector ? buf : toLeakAwareBuffer(buf);
    }
进入InstrumentedUnpooledDirectByteBuf()方法：
private static final class InstrumentedUnpooledDirectByteBuf extends UnpooledDirectByteBuf {
        InstrumentedUnpooledDirectByteBuf(
                UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            super(alloc, initialCapacity, maxCapacity);
        }
进入super()方法：
public UnpooledDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
        super(maxCapacity);
        if (alloc == null) {
            throw new NullPointerException("alloc");
        }
        if (initialCapacity < 0) {
            throw new IllegalArgumentException("initialCapacity: " + initialCapacity);
        }
        if (maxCapacity < 0) {
            throw new IllegalArgumentException("maxCapacity: " + maxCapacity);
        }
        if (initialCapacity > maxCapacity) {
            throw new IllegalArgumentException(String.format(
                    "initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
        }

        this.alloc = alloc;
        setByteBuffer(allocateDirect(initialCapacity));
    }
进入setByteBuffer()方法：
private void setByteBuffer(ByteBuffer buffer) {
        ByteBuffer oldBuffer = this.buffer; 可以看到在这里将allocateDirect分配到内存保存到底层的buffer对象
        if (oldBuffer != null) {
            if (doNotFree) {
                doNotFree = false;
            } else {
                freeDirect(oldBuffer);
            }
        }

        this.buffer = buffer;
        tmpNioBuf = null;
        capacity = buffer.remaining();
    }
进入allocateDirect()方法：
protected ByteBuffer allocateDirect(int initialCapacity) {
        return ByteBuffer.allocateDirect(initialCapacity);
    }
public static ByteBuffer allocateDirect(int capacity) {
        return new DirectByteBuffer(capacity);  最终调用了DirectByteBuffer这个API进行内存分配
    }

ByteBufAllocator分析

内存分配器

进入类AbstractByteBufAllocator中：
public ByteBuf buffer(int initialCapacity, int maxCapacity) {
        if (directByDefault) {  默认是堆外内存就分配堆外内存
            return directBuffer(initialCapacity, maxCapacity);
        }
        return heapBuffer(initialCapacity, maxCapacity);  否则分配堆内内存
    }

类AbstractByteBufAllocator通过暴露两个抽象方法实现扩展，让子类实现创建堆内和堆外内存逻辑
protected abstract ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity);
protected abstract ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity);

netty自动判别unsafe和非unsafe，如果JDK底层有unsafe对象，直接分配一个unsafe类型的ByteBuf，否则分配一个非unsafe类型的ByteBuf。

进入类UnpooledByteBufAllocator中：
protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
        return PlatformDependent.hasUnsafe() ?  判断能不能拿到unsafe对象，来分配相应类型的内存
                new InstrumentedUnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
                new InstrumentedUnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
    }

UnpooledByteBufAllocator分析

heap内存的分配
direct内存大分配

heap内存的分配

分析入口
protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
        return PlatformDependent.hasUnsafe() ? 以下两个类都是UnpooledByteBufAllocator类的内部包装类
                new InstrumentedUnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
                new InstrumentedUnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
    }

InstrumentedUnpooledUnsafeHeapByteBuf类

InstrumentedUnpooledUnsafeHeapByteBuf类
InstrumentedUnpooledUnsafeHeapByteBuf(UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            super(alloc, initialCapacity, maxCapacity);
        }
进入super()方法：
UnpooledUnsafeHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
        super(alloc, initialCapacity, maxCapacity);
}
进入super()方法：注意此时进入的是UnpooledHeapByteBuf类的UnpooledHeapByteBuf()构造方法
public UnpooledHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
        super(maxCapacity);

        checkNotNull(alloc, "alloc");

        if (initialCapacity > maxCapacity) {
            throw new IllegalArgumentException(String.format(
                    "initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
        }

        this.alloc = alloc;
        setArray(allocateArray(initialCapacity));
        setIndex(0, 0);
}
进入setArray()方法：
protected byte[] allocateArray(int initialCapacity) {
        return new byte[initialCapacity];
    }
private void setArray(byte[] initialArray) {
        array = initialArray;
        tmpNioBuf = null;
    }
进入setIndex()方法：
public ByteBuf setIndex(int readerIndex, int writerIndex) {
        if (checkBounds) {
            checkIndexBounds(readerIndex, writerIndex, capacity());
        }
        setIndex0(readerIndex, writerIndex);
        return this;
    }
final void setIndex0(int readerIndex, int writerIndex) {
        this.readerIndex = readerIndex;
        this.writerIndex = writerIndex;
    }

InstrumentedUnpooledHeapByteBuf类

InstrumentedUnpooledHeapByteBuf类
InstrumentedUnpooledHeapByteBuf(UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            super(alloc, initialCapacity, maxCapacity);
        }
进入super()方法：此时和上面的unsafe类中的一样
public UnpooledHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
        super(maxCapacity);

        checkNotNull(alloc, "alloc");

        if (initialCapacity > maxCapacity) {
            throw new IllegalArgumentException(String.format(
                    "initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
        }

        this.alloc = alloc;
        setArray(allocateArray(initialCapacity));
        setIndex(0, 0);
    }

从以上可以看出，unsafe和safe类中的构造方法一样，但是不同之处在于各自类中访问内存的方式。

InstrumentedUnpooledUnsafeHeapByteBuf类的构造方法中调用了父类UnpooledUnsafeHeapByteBuf类的构造方法，
而UnpooledUnsafeHeapByteBuf类的父类是UnpooledHeapByteBuf类
UnpooledUnsafeHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
        super(alloc, initialCapacity, maxCapacity);
    }
在此类中_getbyte()的使用是不同的
 protected byte _getByte(int index) {
        return UnsafeByteBufUtil.getByte(array, index);
    }
static byte getByte(byte[] array, int index) {
        return PlatformDependent.getByte(array, index);
    }
public static byte getByte(byte[] data, int index) {
        return PlatformDependent0.getByte(data, index);
    }
static byte getByte(byte[] data, int index) {
        return UNSAFE.getByte(data, BYTE_ARRAY_BASE_OFFSET + index);
    }

InstrumentedUnpooledHeapByteBuf类的构造方法中调用了父类UnpooledHeapByteBuf类的构造方法
InstrumentedUnpooledHeapByteBuf(UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            super(alloc, initialCapacity, maxCapacity);
        }
在此类中的_getbyte()方法如下：
protected byte _getByte(int index) {
        return HeapByteBufUtil.getByte(array, index);
    }
static byte getByte(byte[] memory, int index) {
        return memory[index];
    }

direct内存的分配

入口
protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
        final ByteBuf buf;
        if (PlatformDependent.hasUnsafe()) {  此时暂时不研究noCleaner的问题
            buf = noCleaner ? new InstrumentedUnpooledUnsafeNoCleanerDirectByteBuf(this, initialCapacity, maxCapacity) :
                    new InstrumentedUnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
        } else {
            buf = new InstrumentedUnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
        }
        return disableLeakDetector ? buf : toLeakAwareBuffer(buf);
    }

InstrumentedUnpooledUnsafeDirectByteBuf包装类中构造方法中调用了父类UnpooledUnsafeDirectByteBuf的构造方法
UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            super(alloc, initialCapacity, maxCapacity);
        }
public UnpooledUnsafeDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
        super(maxCapacity);
        if (alloc == null) {
            throw new NullPointerException("alloc");
        }
        if (initialCapacity < 0) {
            throw new IllegalArgumentException("initialCapacity: " + initialCapacity);
        }
        if (maxCapacity < 0) {
            throw new IllegalArgumentException("maxCapacity: " + maxCapacity);
        }
        if (initialCapacity > maxCapacity) {
            throw new IllegalArgumentException(String.format(
                    "initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
        }

        this.alloc = alloc;
        setByteBuffer(allocateDirect(initialCapacity), false);
    }
进入allocateDirect()方法：
protected ByteBuffer allocateDirect(int initialCapacity) {
        return ByteBuffer.allocateDirect(initialCapacity);
    }
进入allocateDirect()方法：此时调用API新建的内存，这是unpool的性质
public static ByteBuffer allocateDirect(int capacity) {
        return new DirectByteBuffer(capacity);
    }
进入setByteBuffer()方法：  这里直接操作的是内存地址
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;
        memoryAddress = PlatformDependent.directBufferAddress(buffer); 内存地址
        tmpNioBuf = null;
        capacity = buffer.remaining();
    }
进入directBufferAddress()方法：
public static long directBufferAddress(ByteBuffer buffer) {
        return PlatformDependent0.directBufferAddress(buffer);
    }
static long directBufferAddress(ByteBuffer buffer) {
        return getLong(buffer, ADDRESS_FIELD_OFFSET);
    }
private static long getLong(Object object, long fieldOffset) {
        return UNSAFE.getLong(object, fieldOffset);  使用UNSAFE获取内存地址
    }

InstrumentedUnpooledDirectByteBuf类的构造方法调用了父类UnpooledDirectByteBuf类的构造方法
InstrumentedUnpooledDirectByteBuf(
                UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
            super(alloc, initialCapacity, maxCapacity);
        }
public UnpooledDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
        super(maxCapacity);
        if (alloc == null) {
            throw new NullPointerException("alloc");
        }
        if (initialCapacity < 0) {
            throw new IllegalArgumentException("initialCapacity: " + initialCapacity);
        }
        if (maxCapacity < 0) {
            throw new IllegalArgumentException("maxCapacity: " + maxCapacity);
        }
        if (initialCapacity > maxCapacity) {
            throw new IllegalArgumentException(String.format(
                    "initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
        }

        this.alloc = alloc;
        setByteBuffer(allocateDirect(initialCapacity));
    }
进入allocateDirect()方法：
protected ByteBuffer allocateDirect(int initialCapacity) {
        return ByteBuffer.allocateDirect(initialCapacity);
    }
public static ByteBuffer allocateDirect(int capacity) {
        return new DirectByteBuffer(capacity); 调用API新建内存
    }
进入setByteBuffer()方法： 对API进行操作
private void setByteBuffer(ByteBuffer buffer) {
        ByteBuffer oldBuffer = this.buffer;  直接对API进行更新
        if (oldBuffer != null) {
            if (doNotFree) {
                doNotFree = false;
            } else {
                freeDirect(oldBuffer);
            }
        }

        this.buffer = buffer;
        tmpNioBuf = null;
        capacity = buffer.remaining();
    }

类UnpooledDirectByteBuf获取内存的方式
protected byte _getByte(int index) {
        return buffer.get(index);
    }
类UnpooledUnsafeDirectByteBuf获取内存的方式
protected byte _getByte(int index) {
        return UnsafeByteBufUtil.getByte(addr(index));
    }
long addr(int index) {
        return memoryAddress + index;  通过内存地址加偏移量的方式获取内存
    }
static byte getByte(long address) {
        return PlatformDependent.getByte(address);
    }
public static byte getByte(long address) {
        return PlatformDependent0.getByte(address);
    }
static byte getByte(long address) {
        return UNSAFE.getByte(address);
    }