【Netty】ByteBuf （一）
欢迎关注公众号：【 爱编码】
若是有须要后台回复 2019赠送 1T的学习资料哦！！
简介
全部的网路通讯都涉及字节序列的移动，因此高效易用的数据结构明显是必不可少的。Netty的ByteBuf实现知足并超越了这些需求。html
ByteBuf结构
ByteBuf维护了两个不一样的索引：一个是用于读取，一个用于写入。当你从ByteBuf读取是，它的readerIndex将会被递增已经被读取的字节数。一样地，当你写入ByteBuf时，它的witerIndex也会被递增。api
做为一个容器，源码中的以下。有三块区域
discardable bytes：无效空间(已经读取过的空间)，可丢弃字节的区域，由readerIndex指针控制
readable bytes：内容空间，可读字节的区域，由readerIndex和writerIndex指针控制控制
writable bytes：空闲空间，可写入字节的区域，由writerIndex指针和capacity容量控制数组
* <pre> * +-------------------+------------------+------------------+ * | discardable bytes | readable bytes | writable bytes | * | | (CONTENT) | | * +-------------------+------------------+------------------+ * | | | | * 0 <= readerIndex <= writerIndex <= capacity * </pre>
ByteBuf使用模式
整体分类划分是可根据JVM堆内存来区分的。缓存
1.堆内内存（JVM堆空间内）
2.堆外内存（本机直接内存）
3.复合缓冲区（以上2种缓冲区多个混合）
1.堆内内存
最经常使用的ByteBuf模式是将数据存储在JVM的堆空间中。它能在没有使用池化的状况下提供快速的分配和释放。安全
2.堆外内存
JDK容许JVM实现经过本地调用来分配内存。主要是为了不每次调用本地I/O操做以前（或者以后）将缓冲区的内容复制到一个中间缓冲区（或者从中间缓冲区把内容复制到缓冲区）。
**最大的特色:它的内容将驻留在常规的会被垃圾回收的堆以外。
最大的缺点：相对于堆缓冲区，它的分配和释放都是较为昂贵的。**数据结构
3.复合缓冲区
经常使用类：CompositeByteBuf，它为多个ByteBuf提供一个聚合视图，将多个缓冲区表示为单个合并缓冲区的虚拟表示。
好比：HTTP协议：头部和主体这两部分由应用程序的不一样模块产生。这个时候把这两部分合并的话，选择CompositeByteBuf是比较好的。jvm
ByteBuf分类
主要分为三大类ide
Pooled和Unpooled （池化）
unsafe和非unsafe （）
Heap和Direct （堆内和堆外）
Pooled和Unpooled
Pooled：每次都从预先分配好的内存中去取出一段连续内存封装成一个ByteBuf给应用程序使用
Unpooled：每次分配内存的时候，直接调用系统api，向操做系统申请一
块内存学习
Heap和Direct:
Head：是调用jvm的堆内存进行分配的，须要被gc进行管理
Direct：是调用jdk的api进行内存分配，不受jvm控制，不会参与到gc的过程大数据
Unsafe和非Unsafe
jdk中有Unsafe对象能够直接拿到对象的内存地址，而且基于这个内存地址进行读写操做。那么对应的分类的区别就是是否能够拿到jdk底层的Unsafe进行读写操做了。
内存分配ByteBufAllocator
这个接口实现负责分配缓冲区而且是线程安全的。从下面的接口方法以及注释能够总结出主要是围绕上面的三种ByteBuf内存模式：堆内，堆外以及复合型的内存分配。
/**
* Implementations are responsible to allocate buffers. Implementations of this interface are expected to be
* thread-safe.
*/
public interface ByteBufAllocator {
ByteBufAllocator DEFAULT = ByteBufUtil.DEFAULT_ALLOCATOR;
/**
* Allocate a {@link ByteBuf}. If it is a direct or heap buffer
* depends on the actual implementation.
*/
ByteBuf buffer();
/**
* Allocate a {@link ByteBuf} with the given initial capacity.
* If it is a direct or heap buffer depends on the actual implementation.
*/
ByteBuf buffer(int initialCapacity);
/**
* Allocate a {@link ByteBuf} with the given initial capacity and the given
* maximal capacity. If it is a direct or heap buffer depends on the actual
* implementation.
*/
ByteBuf buffer(int initialCapacity, int maxCapacity);
/**
* Allocate a {@link ByteBuf}, preferably a direct buffer which is suitable for I/O.
*/
ByteBuf ioBuffer();
/**
* Allocate a {@link ByteBuf}, preferably a direct buffer which is suitable for I/O.
*/
ByteBuf ioBuffer(int initialCapacity);
/**
* Allocate a {@link ByteBuf}, preferably a direct buffer which is suitable for I/O.
*/
ByteBuf ioBuffer(int initialCapacity, int maxCapacity);
/**
* Allocate a heap {@link ByteBuf}.
*/
ByteBuf heapBuffer();
/**
* Allocate a heap {@link ByteBuf} with the given initial capacity.
*/
ByteBuf heapBuffer(int initialCapacity);
/**
* Allocate a heap {@link ByteBuf} with the given initial capacity and the given
* maximal capacity.
*/
ByteBuf heapBuffer(int initialCapacity, int maxCapacity);
/**
* Allocate a direct {@link ByteBuf}.
*/
ByteBuf directBuffer();
/**
* Allocate a direct {@link ByteBuf} with the given initial capacity.
*/
ByteBuf directBuffer(int initialCapacity);
/**
* Allocate a direct {@link ByteBuf} with the given initial capacity and the given
* maximal capacity.
*/
ByteBuf directBuffer(int initialCapacity, int maxCapacity);
/**
* Allocate a {@link CompositeByteBuf}.
* If it is a direct or heap buffer depends on the actual implementation.
*/
CompositeByteBuf compositeBuffer();
/**
* Allocate a {@link CompositeByteBuf} with the given maximum number of components that can be stored in it.
* If it is a direct or heap buffer depends on the actual implementation.
*/
CompositeByteBuf compositeBuffer(int maxNumComponents);
/**
* Allocate a heap {@link CompositeByteBuf}.
*/
CompositeByteBuf compositeHeapBuffer();
/**
* Allocate a heap {@link CompositeByteBuf} with the given maximum number of components that can be stored in it.
*/
CompositeByteBuf compositeHeapBuffer(int maxNumComponents);
/**
* Allocate a direct {@link CompositeByteBuf}.
*/
CompositeByteBuf compositeDirectBuffer();
/**
* Allocate a direct {@link CompositeByteBuf} with the given maximum number of components that can be stored in it.
*/
CompositeByteBuf compositeDirectBuffer(int maxNumComponents);
/**
* Returns {@code true} if direct {@link ByteBuf}'s are pooled
*/
boolean isDirectBufferPooled();
/**
* Calculate the new capacity of a {@link ByteBuf} that is used when a {@link ByteBuf} needs to expand by the
* {@code minNewCapacity} with {@code maxCapacity} as upper-bound.
*/
int calculateNewCapacity(int minNewCapacity, int maxCapacity);
}
其中ByteBufAllocator 的具体实现能够查看其子类，以下图
下面来看看各自子类的功能以及区别
UnpooledByteBufAllocator
heap内存分配
入口new InstrumentedUnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
发现分配Unpooled、Unsafe、Heap内存，实际上是分配了一个byte数组，并保存在UnpooledHeapByteBuf#array成员变量中。该内存的初始值容量和最大可扩展容量能够指定。
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);
}
protected byte[] allocateArray(int initialCapacity) {
return new byte[initialCapacity];
}
查看UnpooledHeapByteBuf#getByte()方法，堆内存类型的ByteBuf获取的时候。直接经过下标获取byte数组中的byte
@Override
public byte getByte(int index) {
ensureAccessible();
return _getByte(index);
}
@Override
protected byte _getByte(int index) {
//该array为初始化的时候，实例化的byte[]
return HeapByteBufUtil.getByte(array, index);
}
static byte getByte(byte[] memory, int index) {
//直接拿到一个数组
return memory[index];
}
direct内存分配
入口UnpooledByteBufAllocator#newDirectBuffer() --> UnpooledUnsafeDirectByteBuf
public UnpooledUnsafeDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) {
super(maxCapacity);
if (alloc == null) {
throw new NullPointerException("alloc");
}
checkPositiveOrZero(initialCapacity, "initialCapacity");
checkPositiveOrZero(maxCapacity, "maxCapacity");
if (initialCapacity > maxCapacity) {
throw new IllegalArgumentException(String.format(
"initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity));
}
this.alloc = alloc;
setByteBuffer(allocateDirect(initialCapacity), false);
}
protected ByteBuffer allocateDirect(int initialCapacity) {
return ByteBuffer.allocateDirect(initialCapacity);
}
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();
}
能够发现，Unpooled、Direct类型得内存分配其实是维护了一个底层jdk的一个DirectByteBuffer。分配内存的时候就建立它，并将他保存到buffer成员变量。
跟踪iUnpooledHeapByteBuf#_getByte()，就比较简单了，直接使用jdk的api获取
@Override
protected byte _getByte(int index) {
//使用buffer
return buffer.get(index);
}
更加详细的分析能够查看下面这篇文章
https://www.jianshu.com/p/158...
PooledByteBufAllocator
入口PooledByteBufAllocator#newHeapBuffer()和PooledByteBufAllocator#newDirectBuffer()，堆内内存和堆外内存分配的模式都比较固定。
1.拿到线程局部缓存PoolThreadCache
2.拿到不一样类型的rena
3.使用不一样类型的arena进行内存分配
@Override
protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
//拿到线程局部缓存
PoolThreadCache cache = threadCache.get();
//拿到heapArena
PoolArena<byte[]> heapArena = cache.heapArena;
final ByteBuf buf;
if (heapArena != null) {
//使用heapArena分配内存
buf = heapArena.allocate(cache, initialCapacity, maxCapacity);
} else {
buf = PlatformDependent.hasUnsafe() ?
new UnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) :
new UnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
}
return toLeakAwareBuffer(buf);
}
@Override
protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
//拿到线程局部缓存
PoolThreadCache cache = threadCache.get();
//拿到directArena
PoolArena<ByteBuffer> directArena = cache.directArena;
final ByteBuf buf;
if (directArena != null) {
//使用directArena分配内存
buf = directArena.allocate(cache, initialCapacity, maxCapacity);
} else {
buf = PlatformDependent.hasUnsafe() ?
UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) :
new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
}
return toLeakAwareBuffer(buf);
}
跟踪threadCache.get()调用的是FastThreadLocal#get()方法。那么其实threadCache也是一个FastThreadLocal,能够当作是jdk的ThreadLocal，get方法调用了初始化方法initializel
public final V get() {
InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get();
Object v = threadLocalMap.indexedVariable(index);
if (v != InternalThreadLocalMap.UNSET) {
return (V) v;
}
//调用初始化方法
V value = initialize(threadLocalMap);
registerCleaner(threadLocalMap);
return value;
}
initialValue()方法的逻辑以下
1.从预先准备好的heapArenas和directArenas中获取最少使用的arena
2.使用获取到的arean为参数，实例化一个PoolThreadCache并返回
final class PoolThreadLocalCache extends FastThreadLocal<PoolThreadCache> {
private final boolean useCacheForAllThreads;
PoolThreadLocalCache(boolean useCacheForAllThreads) {
this.useCacheForAllThreads = useCacheForAllThreads;
}
@Override
protected synchronized PoolThreadCache initialValue() {
/**
* arena翻译成竞技场，关于内存非配的逻辑都在这个竞技场中进行分配
*/
//获取heapArena：从heapArenas堆内竞技场中拿出使用最少的一个arena
final PoolArena<byte[]> heapArena = leastUsedArena(heapArenas);
//获取directArena：从directArena堆内竞技场中拿出使用最少的一个arena
final PoolArena<ByteBuffer> directArena = leastUsedArena(directArenas);
Thread current = Thread.currentThread();
if (useCacheForAllThreads || current instanceof FastThreadLocalThread) {
//建立PoolThreadCache：该Cache最终被一个线程使用
//经过heapArena和directArena维护两大块内存：堆和堆外内存
//经过tinyCacheSize，smallCacheSize，normalCacheSize维护ByteBuf缓存列表维护反复使用的内存块
return new PoolThreadCache(
heapArena, directArena, tinyCacheSize, smallCacheSize, normalCacheSize,
DEFAULT_MAX_CACHED_BUFFER_CAPACITY, DEFAULT_CACHE_TRIM_INTERVAL);
}
// No caching so just use 0 as sizes.
return new PoolThreadCache(heapArena, directArena, 0, 0, 0, 0, 0);
}
//省略代码......
}
查看PoolThreadCache其维护了两种类型的内存分配策略，一种是上述经过持有heapArena和directArena，另外一种是经过维护tiny,small,normal对应的缓存列表来维护反复使用的内存。
final class PoolThreadCache {
private static final InternalLogger logger = InternalLoggerFactory.getInstance(PoolThreadCache.class);
//经过arena的方式维护内存
final PoolArena<byte[]> heapArena;
final PoolArena<ByteBuffer> directArena;
//维护了tiny, small, normal三种类型的缓存列表
// Hold the caches for the different size classes, which are tiny, small and normal.
private final MemoryRegionCache<byte[]>[] tinySubPageHeapCaches;
private final MemoryRegionCache<byte[]>[] smallSubPageHeapCaches;
private final MemoryRegionCache<ByteBuffer>[] tinySubPageDirectCaches;
private final MemoryRegionCache<ByteBuffer>[] smallSubPageDirectCaches;
private final MemoryRegionCache<byte[]>[] normalHeapCaches;
private final MemoryRegionCache<ByteBuffer>[] normalDirectCaches;
// Used for bitshifting when calculate the index of normal caches later
private final int numShiftsNormalDirect;
private final int numShiftsNormalHeap;
private final int freeSweepAllocationThreshold;
private final AtomicBoolean freed = new AtomicBoolean();
private int allocations;
// TODO: Test if adding padding helps under contention
//private long pad0, pad1, pad2, pad3, pad4, pad5, pad6, pad7;
PoolThreadCache(PoolArena<byte[]> heapArena, PoolArena<ByteBuffer> directArena,
int tinyCacheSize, int smallCacheSize, int normalCacheSize,
int maxCachedBufferCapacity, int freeSweepAllocationThreshold) {
checkPositiveOrZero(maxCachedBufferCapacity, "maxCachedBufferCapacity");
this.freeSweepAllocationThreshold = freeSweepAllocationThreshold;
//经过持有heapArena和directArena，arena的方式管理内存分配
this.heapArena = heapArena;
this.directArena = directArena;
//经过tinyCacheSize,smallCacheSize,normalCacheSize建立不一样类型的缓存列表并保存到成员变量
if (directArena != null) {
tinySubPageDirectCaches = createSubPageCaches(
tinyCacheSize, PoolArena.numTinySubpagePools, SizeClass.Tiny);
smallSubPageDirectCaches = createSubPageCaches(
smallCacheSize, directArena.numSmallSubpagePools, SizeClass.Small);
numShiftsNormalDirect = log2(directArena.pageSize);
normalDirectCaches = createNormalCaches(
normalCacheSize, maxCachedBufferCapacity, directArena);
directArena.numThreadCaches.getAndIncrement();
} else {
// No directArea is configured so just null out all caches
tinySubPageDirectCaches = null;
smallSubPageDirectCaches = null;
normalDirectCaches = null;
numShiftsNormalDirect = -1;
}
if (heapArena != null) {
// Create the caches for the heap allocations
//建立规格化缓存队列
tinySubPageHeapCaches = createSubPageCaches(
tinyCacheSize, PoolArena.numTinySubpagePools, SizeClass.Tiny);
//建立规格化缓存队列
smallSubPageHeapCaches = createSubPageCaches(
smallCacheSize, heapArena.numSmallSubpagePools, SizeClass.Small);
numShiftsNormalHeap = log2(heapArena.pageSize);
//建立规格化缓存队列
normalHeapCaches = createNormalCaches(
normalCacheSize, maxCachedBufferCapacity, heapArena);
heapArena.numThreadCaches.getAndIncrement();
} else {
// No heapArea is configured so just null out all caches
tinySubPageHeapCaches = null;
smallSubPageHeapCaches = null;
normalHeapCaches = null;
numShiftsNormalHeap = -1;
}
// Only check if there are caches in use.
if ((tinySubPageDirectCaches != null || smallSubPageDirectCaches != null || normalDirectCaches != null
|| tinySubPageHeapCaches != null || smallSubPageHeapCaches != null || normalHeapCaches != null)
&& freeSweepAllocationThreshold < 1) {
throw new IllegalArgumentException("freeSweepAllocationThreshold: "
+ freeSweepAllocationThreshold + " (expected: > 0)");
}
}
private static <T> MemoryRegionCache<T>[] createSubPageCaches(
int cacheSize, int numCaches, SizeClass sizeClass) {
if (cacheSize > 0 && numCaches > 0) {
//MemoryRegionCache 维护缓存的一个对象
@SuppressWarnings("unchecked")
MemoryRegionCache<T>[] cache = new MemoryRegionCache[numCaches];
for (int i = 0; i < cache.length; i++) {
// TODO: maybe use cacheSize / cache.length
//每一种MemoryRegionCache(tiny,small,normal)都表示不一样内存大小（不一样规格）的一个队列
cache[i] = new SubPageMemoryRegionCache<T>(cacheSize, sizeClass);
}
return cache;
} else {
return null;
}
}
private static <T> MemoryRegionCache<T>[] createNormalCaches(
int cacheSize, int maxCachedBufferCapacity, PoolArena<T> area) {
if (cacheSize > 0 && maxCachedBufferCapacity > 0) {
int max = Math.min(area.chunkSize, maxCachedBufferCapacity);
int arraySize = Math.max(1, log2(max / area.pageSize) + 1);
//MemoryRegionCache 维护缓存的一个对象
@SuppressWarnings("unchecked")
MemoryRegionCache<T>[] cache = new MemoryRegionCache[arraySize];
for (int i = 0; i < cache.length; i++) {
//每一种MemoryRegionCache(tiny,small,normal)都表示不一样内存（不一样规格）大小的一个队列
cache[i] = new NormalMemoryRegionCache<T>(cacheSize);
}
return cache;
} else {
return null;
}
}
......
}
更加详细分析可参考如下文章
https://www.jianshu.com/p/1cd...
directArena分配direct内存的流程
上一步拿到PoolThreadCache以后，获取对应的Arena。那么以后就是Arena具体分配内存的步骤。
入口PooledByteBufAllocator#newDirectBuffer()方法种有以下代码
PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) {
//拿到PooledByteBuf对象,仅仅是一个对象
PooledByteBuf<T> buf = newByteBuf(maxCapacity);
//从cache种分配内存，并初始化buf种内存地址相关的属性
allocate(cache, buf, reqCapacity);
return buf;
}
能够看到分配的过程以下：拿到PooledByteBuf对象从cache中分配内存，并重置相关属性.
1.newByteBuf(maxCapacity);拿到PooledByteBuf对象
@Override
protected PooledByteBuf<ByteBuffer> newByteBuf(int maxCapacity) {
if (HAS_UNSAFE) {
//获取一个PooledByteBuf
return PooledUnsafeDirectByteBuf.newInstance(maxCapacity);
} else {
return PooledDirectByteBuf.newInstance(maxCapacity);
}
}
static PooledUnsafeDirectByteBuf newInstance(int maxCapacity) {
//从带有回收特性的对象池RECYCLER获取一个PooledUnsafeDirectByteBuf
PooledUnsafeDirectByteBuf buf = RECYCLER.get();
//buf多是从回收站拿出来的，要进行复用
buf.reuse(maxCapacity);
return buf;
}
2.Recycler是一个基于线程本地堆栈的对象池。Recycler维护了一个ThreadLocal成员变量，用于返回一个stack给回收处理DefaultHandle，该处理器经过维护这个堆栈来维护PooledUnsafeDirectByteBuf缓存。
private static final Recycler<PooledUnsafeDirectByteBuf> RECYCLER = new Recycler<PooledUnsafeDirectByteBuf>() {
@Override
protected PooledUnsafeDirectByteBuf newObject(Handle<PooledUnsafeDirectByteBuf> handle) {
//Recycler负责用回收处理器handler维护PooledUnsafeDirectByteBuf
//handler底层持有一个stack做为对象池，维护对象池，handle同时负责对象回收
//存储handler为成员变量，使用完该ByteBuf能够调用回收方法回收
return new PooledUnsafeDirectByteBuf(handle, 0);
}
};
//维护了一个`ThreadLocal`，`initialValue`方法返回一个堆栈。
private final FastThreadLocal<Stack<T>> threadLocal = new FastThreadLocal<Stack<T>>() {
@Override
protected Stack<T> initialValue() {
return new Stack<T>(Recycler.this, Thread.currentThread(), maxCapacityPerThread, maxSharedCapacityFactor,
ratioMask, maxDelayedQueuesPerThread);
}
@Override
protected void onRemoval(Stack<T> value) {
// Let us remove the WeakOrderQueue from the WeakHashMap directly if its safe to remove some overhead
if (value.threadRef.get() == Thread.currentThread()) {
if (DELAYED_RECYCLED.isSet()) {
DELAYED_RECYCLED.get().remove(value);
}
}
}
};
3.再看Recycler#get()方法
public final T get() {
if (maxCapacityPerThread == 0) {
return newObject((Handle<T>) NOOP_HANDLE);
}
//获取对应的堆栈，至关一个回收站
Stack<T> stack = threadLocal.get();
//从栈顶拿出一个来DefaultHandle（回收处理器）
//DefaultHandle持有一个value，实际上是PooledUnsafeDirectByteBuf
DefaultHandle<T> handle = stack.pop();
//没有回收处理器，说明没有闲置的ByteBuf
if (handle == null) {
//新增一个处理器
handle = stack.newHandle();
//回调，还记得么？该回调返回一个PooledUnsafeDirectByteBuf
//让处理器持有一个新的PooledUnsafeDirectByteBuf
handle.value = newObject(handle);
}
//若是有，则可直接重复使用
return (T) handle.value;
}
public final V get() {
InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get();
Object v = threadLocalMap.indexedVariable(index);
if (v != InternalThreadLocalMap.UNSET) {
return (V) v;
}
//回调initialize
V value = initialize(threadLocalMap);
registerCleaner(threadLocalMap);
return value;
}
private V initialize(InternalThreadLocalMap threadLocalMap) {
V v = null;
try {
//回调
v = initialValue();
} catch (Exception e) {
PlatformDependent.throwException(e);
}
threadLocalMap.setIndexedVariable(index, v);
addToVariablesToRemove(threadLocalMap, this);
return v;
}
DefaultHandle<T> newHandle() {
//实例化一个处理器并而且初四话成员变量，该成员变量stack从threalocal中初始化
return new DefaultHandle<T>(this);
}
DefaultHandle用stack做为缓存池维护PooledUnsafeDirectByteBuf,同理PooledDirectByteBuf也是同样的。只不过实例化的对象的实现不同而已。同时，处理器定义了回收的方法是将兑现存回栈内，使用的时候则是从栈顶取出。
static final class DefaultHandle<T> implements Handle<T> {
private int lastRecycledId;
private int recycleId;
boolean hasBeenRecycled;
//对象缓存池
private Stack<?> stack;
private Object value;
DefaultHandle(Stack<?> stack) {
this.stack = stack;
}
/**
* 定义回收方法，回收对象到stack
* @param object
*/
@Override
public void recycle(Object object) {
if (object != value) {
throw new IllegalArgumentException("object does not belong to handle");
}
Stack<?> stack = this.stack;
if (lastRecycledId != recycleId || stack == null) {
throw new IllegalStateException("recycled already");
}
//回收：将本身存进栈中缓存起来
stack.push(this);
}
}
到这咱们刚刚看完第一步，到第二步重置缓存内指针的时候了 ,获取到PooledUnsafeDirectByteBuf的时候，有多是从缓存中取出来的。所以须要复用.
static PooledUnsafeDirectByteBuf newInstance(int maxCapacity) {
//从带有回收特性的对象池RECYCLER获取一个PooledUnsafeDirectByteBuf
PooledUnsafeDirectByteBuf buf = RECYCLER.get();
//buf多是从回收站拿出来的，要进行复用
buf.reuse(maxCapacity);
return buf;
}
final void reuse(int maxCapacity) {
//重置最大容量
maxCapacity(maxCapacity);
//设置引用
setRefCnt(1);
//重置指针
setIndex0(0, 0);
//重置标记值
discardMarks();
}
到这才刚刚完成分配内存的第一步（拿到PooledByteBuf对象），以上都是仅仅是获取而且用回收站和回收处理器管理这些对象，这些对象仍然只是一个对象，尚未分配实际的内存。
跟踪PoolArena#allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity)
private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity) {
final int normCapacity = normalizeCapacity(reqCapacity);
//不一样的规格大小进行内存分配
/**
* 分配总体逻辑（先判断tiny和small规格的，再判断normal规格的）
* 1. 尝试从缓存上进行内存分配，成功则返回
* 2. 失败则再从内存堆中进行分配内存
*/
if (isTinyOrSmall(normCapacity)) { // capacity < pageSize
int tableIdx;
PoolSubpage<T>[] table;
boolean tiny = isTiny(normCapacity);
//尝试tiny和small规格的缓存内存分配
if (tiny) { // < 512
if (cache.allocateTiny(this, buf, reqCapacity, normCapacity)) {
// was able to allocate out of the cache so move on
return;
}
tableIdx = tinyIdx(normCapacity);
table = tinySubpagePools;
} else {
if (cache.allocateSmall(this, buf, reqCapacity, normCapacity)) {
// was able to allocate out of the cache so move on
return;
}
tableIdx = smallIdx(normCapacity);
table = smallSubpagePools;
}
final PoolSubpage<T> head = table[tableIdx];
/**
* Synchronize on the head. This is needed as {@link PoolChunk#allocateSubpage(int)} and
* {@link PoolChunk#free(long)} may modify the doubly linked list as well.
*/
synchronized (head) {
final PoolSubpage<T> s = head.next;
if (s != head) {
assert s.doNotDestroy && s.elemSize == normCapacity;
long handle = s.allocate();
assert handle >= 0;
s.chunk.initBufWithSubpage(buf, null, handle, reqCapacity);
incTinySmallAllocation(tiny);
return;
}
}
//tiny和small规格的缓存内存分配尝试失败
//从内存堆中分配内存
synchronized (this) {
allocateNormal(buf, reqCapacity, normCapacity);
}
incTinySmallAllocation(tiny);
return;
}
//normal规格
//若是分配处出来的内存大于一个值（chunkSize），则执行allocateHuge
if (normCapacity <= chunkSize) {
//从缓存上进行内存分配
if (cache.allocateNormal(this, buf, reqCapacity, normCapacity)) {
// was able to allocate out of the cache so move on
return;
}
//缓存没有再从内存堆中分配内存
synchronized (this) {
allocateNormal(buf, reqCapacity, normCapacity);
++allocationsNormal;
}
} else {
// Huge allocations are never served via the cache so just call allocateHuge
allocateHuge(buf, reqCapacity);
}
}
其总体分配内存的逻辑是根据不一样规格大小的内存须要来的，显示tiny和small规格的，再是normal规格的。分配也是先尝试从缓存中进行内存分配，若是分配失败再从内存堆中进行内存分配。 固然，分配出来的内存回和第一步拿到的PooledByteBuf进行绑定起来。
总结
主要学习了ByteBuf 的基本结构、使用模式、分类、基本的内存分配。
下次再学习ByteBuf 的命中逻辑以及内存回收。
参考文章
https://www.cnblogs.com/xiang...
https://www.jianshu.com/u/fc9...
最后
若是对 Java、大数据感兴趣请长按二维码关注一波，我会努力带给大家价值。以为对你哪怕有一丁点帮助的请帮忙点个赞或者转发哦。
关注公众号【爱编码】，回复2019有相关资料哦。
相关文章
- 1. Netty ByteBuf
- 2. netty---ByteBuf
- 3. Netty Bytebuf
- 4. Netty Associated -- ByteBuf
- 5. netty中的ByteBuf
- 6. Netty ByteBuf分析
- 7. Netty之ByteBuf
- 8. Netty ByteBuf转String
- 9. Netty 4.x – ByteBuf
- 10. Netty - Bytebuf（1）
- 更多相关文章...
- • 一对一关联查询 - MyBatis教程
- • MySQL唯一约束（UNIQUE KEY） - MySQL教程
- • RxJava操作符（一）Creating Observables
- • Kotlin学习（一）基本语法
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章