Netty中ByteBuf详细介绍
不论我们在前面学习NIO的ByteBuffer，还是现在Netty当中的ByteBuf，其都有使用直接内存的方式。
在Netty当中，我们使用完直接内存，需要去手动进行释放，而不应该等待GC去进行回收，以减少发生内存溢出的风险。
ByteBuf与ByteBuffer的区别
设计理念：
Buffer: Buffer是Java NIO（New Input/Output）中的一个抽象类，用于表示一个字节序列的缓冲区。它是所有缓冲区类的父类，子类包括 ByteBuffer、CharBuffer、IntBuffer 等，分别处理不同类型的数据，提供了对字节数据的读写操作。
ByteBuffer：是Java NIO中的一个核心类，用于处理字节数据。它是一个抽象类，提供了基本的缓冲区操作，如读写、翻转等12。
ByteBuf：是Netty框架中的一个类，专门用于处理字节数据。它提供了比ByteBuffer更灵活和强大的功能，如读写索引分离、动态扩容、内存池化等23。
特性:
Buffer特性:
固定容量：创建时指定容量，不可动态扩展。
读写模式切换：通过 flip()、clear()、compact() 等方法切换读写模式。
直接内存支持：通过 allocateDirect() 分配堆外内存（避免 JVM 堆与本地内存的拷贝）。
局限性：
读写索引共用 position，需频繁切换模式，容易出错。
API 功能较为基础，缺少高级操作（如自动扩容、复合缓冲区）。
ByteBuffer特性:
直接操作字节：提供 get()、put() 等方法读写字节。
堆内存与堆外内存：
allocate()：分配堆内存（JVM 管理）。
allocateDirect()：分配堆外内存（由操作系统管理，减少拷贝开销）。
视图缓冲区：支持生成只读缓冲区或分片（slice()、asReadOnlyBuffer()）。
局限性：
固定容量：无法动态扩容，需手动管理空间。
读写切换繁琐：每次读写后需调用 flip() 或 rewind() 重置指针。
功能受限：缺少复合缓冲区、引用计数等高级特性。
属性说明:
capacity: 缓冲区容量（创建后不可变）
position: 当前读写位置
limit: 可读写边界
mark:标记位置（用于reset）
position 和 limit 的关系：
写模式：position 表示下一个写入位置，limit 表示容量
读模式：position 表示下一个读取位置，limit 表示可读上限
ByteBuf:
读写索引分离：readerIndex 和 writerIndex 分开，无需调用 flip()。
动态扩容：写入数据时自动扩展容量（可配置阈值）。
内存池化：支持基于内存池的分配（PooledByteBuf），减少内存分配开销。
复合缓冲区：通过 CompositeByteBuf 将多个 ByteBuf 逻辑合并，避免内存拷贝。
引用计数：基于 ReferenceCounted 接口实现内存的自动释放（类似 C++ 的智能指针）。
零拷贝支持：通过 slice()、duplicate() 共享数据，避免拷贝。
内存类型：
堆内存（HeapByteBuf）：由 JVM 管理。
堆外内存（DirectByteBuf）：由操作系统管理，适合网络 I/O。
优势场景：
高频读写操作（如网络通信）。
需要高性能内存管理（如高并发服务器）。
区别:
读写模式
Buffer：作为抽象类，定义了缓冲区的通用行为，读写模式的具体实现依赖于其子类。
ByteBuffer：使用单一的position指针来跟踪读写的位置。在读写模式切换时，需要调用flip方法将position重置为0，并将limit设置为之前的position值。在写操作之前，通常需要调用clear方法来准备缓冲区。
ByteBuf：使用两个独立的指针，readerIndex和writerIndex，分别跟踪读和写的位置。这种设计允许读写操作并发进行，无需模式切换。
扩容支持
Buffer：抽象类本身不支持动态扩容。
ByteBuffer：一旦分配完成，容量固定，无法自动扩展。
ByteBuf：支持动态扩容。当数据超出当前容量时，ByteBuf会自动分配一个新的、更大的字节数组，并将旧数组的内容复制到新数组中。
池化机制
Buffer：抽象类本身不包含池化机制。
ByteBuffer：没有内置的内存池化机制。每次分配和释放都需要系统调用。
ByteBuf：内置内存池化机制（如PooledByteBufAllocator），优化了内存分配和回收效率。这有助于减少内存碎片和垃圾回收的开销。
引用计数
Buffer：抽象类本身不支持引用计数。
ByteBuffer：不支持引用计数。生命周期需要开发者手动管理。
ByteBuf：支持引用计数（retain()、release()），可以自动管理缓冲区的生命周期。这有助于避免内存泄漏和重复释放内存的问题。
零拷贝特性
Buffer：抽象类本身不包含零拷贝特性。
ByteBuffer：支持基础的零拷贝操作，如slice()和duplicate()。
ByteBuf：支持更高级的零拷贝功能，如slice和复合缓冲区（CompositeByteBuf）。这些操作可以在不复制数据的情况下有效地处理数据，从而减少了CPU的负担和内存的消耗。
ByteBuf的种类
关于其种类，有很多种，我们根据前面提到的池化机制，将其主要分为两大类，每一类当当中又分为堆内存和直接内存：
UnpooledHeapByteBuf：非池化堆内存ByteBuf，受JVM内存管理，可以等待GC回收。
UnpooledDirectByteBuf：非池化直接内存ByteBuf，不收JVM管理，虽然可以受GC回收，但不是及时的，可能会发生内存溢出，需要手动进行回收。
PooledByteBuf：池化ByteBuf，这种有更复杂的回收范式，后面通过源码分析，具体查看其实现细节。
PooledHeapByteBuf：池化堆内存ByteBuf
PooledDirectByteBuf：池化直接内存ByteBuf
ByteBuf 该如何选择: 一般业务数据的内存分配选用Java堆内存UnpooledHeapByteBuf，其实现简单，回收快，不会出现内存管理问题；对于I/O数据的内存分配一般选用池化的直接内存PooledDirectByteBuf，避免Java堆内存到直接内存的拷贝，但使用池化ByteBuf时切记自己分配的内存一定要在用完后手动释放。
Bytebuf简单总结：
池化 - 可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能
读写指针分离，不需要像 ByteBuffer 一样切换读写模式
可以自动扩容
支持链式调用，使用更流畅
很多地方体现零拷贝，例如 slice、duplicate、CompositeByteBuf
一、ByteBuf的结构
每一个ByteBuf都有一个可容纳的字节上限叫capacity。在ByteBuf中通过两个指针readerIndex和writerIndex将整个个ByteBuf划分成3个部分，分别是已丢弃部分、可读部分和可写部分，示意图如下。
+-------------------+------------------+------------------+
| 已丢弃 | 可读 | 可写 |
| | (CONTENT) | |
+-------------------+------------------+------------------+
| | | |
0 <= readerIndex <= writerIndex <= capacityreaderIndex、writerIndex和capacity之间满足简单地数学关系0 <= readerIndex <= writerIndex <= capcity。
二、ByteBuf使用
ByteBuf的主要分类
ByteBuf的主要实现类图如下图所示，总体上分为Pooled和UnPooled两类，顾名思义为池化的和非池化的，又根据分配的是直接内存还是堆内存分为HeapByteBuf和DirectByteBuf。
通过ByteBufAllocator创建
ByteBufAllocator有两个实现分别为PooledByteBufAllocator和UnpooledByteBufAllocator对应分配出来是的PooledByteBuf和UnpooledByteBuf。
PooledByteBufAllocator将在后续的文章中重点分析，而UnpooledByteBufAllocator则比较简单，不再赘述。
通过包裹ByteBuffer及byte[]创建
使用Unpooled.wrappedBuffer方法可以对jdkButeBuffer及byte[]进行包裹创建出一个UnpooledByteBuf。
ByteBuf的创建方式
Netty通过ByteBufAllocator分配器来创建缓冲区和分配内存空间。Netty提供了ByteBufAllocator的两种实现：PoolByteBufAllocator和UnpooledByteBufAllocator。
PoolByteBufAllocator（池化ByteBuf分配器）将ByteBuf实例放入池中，提高了性能，将内存碎片减少到最小；这个池化分配器采用了jemalloc高效内存分配的策略，该策略被好几种现代操作系统所采用。
UnpooledByteBufAllocator是普通的未池化ByteBuf分配器，它没有把ByteBuf放入池中，每次被调用时，返回一个新的ByteBuf实例；通过Java的垃圾回收机制回收。
在Netty中，默认的分配器为ByteBufAllocator.DEFAULT，可以通过Java系统参数（System Property）的选项io.netty.allocator.type进行配置，配置时使用字符串值："unpooled", "pooled"。
不同的Netty版本，对于分配器的默认使用策略是不一样的。在Netty 4.0版本中，默认的分配器为UnpooledByteBufAllocator。而在Netty 4.1版本中，默认的分配器为PooledByteBufAllocator。现在PooledByteBufAllocator已经广泛使用了一段时间，并且有了增强的缓冲区泄漏追踪机制。因此，可以在Netty程序中设置启动器Bootstrap的时候，将PooledByteBufAllocator设置为默认的分配器。
ServerBootstrap b = new ServerBootstrap()
//....
//4 设置通道的参数
b.option(ChannelOption.SO_KEEPALIVE, true);
b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
b.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
//....分配的方式:
ByteBuf buffer = null;
//创建ByteBuf，类似可能是堆或直接的，依赖具体的实现
//方法一：分配器默认分配初始容量为9，最大容量100的缓冲区
buffer = ByteBufAllocator.DEFAULT.buffer(9, 100);
//方法二：分配器默认分配初始容量为256，最大容量Integer.MAX_VALUE的缓冲区
buffer = ByteBufAllocator.DEFAULT.buffer();
//方法三：非池化分配器，分配基于Java的堆（Heap）结构内存缓冲区
buffer = UnpooledByteBufAllocator.DEFAULT.heapBuffer();
//方法四：池化分配器，分配基于操作系统管理的直接内存缓冲区
buffer = PooledByteBufAllocator.DEFAULT.directBuffer();
//创建堆类型ByteBuf
buffer = ByteBufAllocator.DEFAULT.heapBuffer()
//创建直接内存类型ByteBuf
buffer = ByteBufAllocator.DEFAULT.directBuffer()
//创建复合类型缓冲区
buffer = ByteBufAllocator.DEFAULT.compositeBuffer()
//创建内部为堆类型ByteBuf的复合缓冲区
buffer = ByteBufAllocator.DEFAULT.heapCompositeBuffer()
//创建内部为直接类型ByteBuf的复合缓冲区
buffer = ByteBufAllocator.DEFAULT.directCompositeBuffer()
#创建用于IO的byteBuf
buffer = ByteBufAllocator.DEFAULT.ioBuffer()//创建未使用池技术的堆缓冲ByteBuf
ByteBuf buf1 = Unpooled.buffer(100);
//创建未使用池技术的直接缓冲ByteBuf
ByteBuf buf2 = Unpooled.directBuffer();
//创建包含给定数据的ByteBuf,零拷贝
ByteBuf buf2 = Unpooled.wrappedBuffer();
//复制给定数据然后创建ByteBuf
ByteBuf buf2 = Unpooled.copiedBuffer();
#组合的byteBuf
CompositeByteBuf byteBufs = ByteBufAllocator.DEFAULT.compositeDirectBuffer(2);
byteBufs.addComponent(true, Unpooled.buffer(2).writeShort(buf.readableBytes()));
byteBufs.addComponent(true, buf);
#深拷贝的byteBuf
ByteBuf buf = Unpooled.copiedBuffer("Netty in Action rocks!", utf8);ByteBuf的回收方式
方式一:
在pipline管道的头尾,会自动释放byteBuf, 在DefaultChannelPipeline中,有HeadContext,TailContext一头一尾两个上下文中释放传入的ByteBuf.
protected void onUnhandledInboundMessage(Object msg) {
try {
logger.debug(
"Discarded inbound message {} that reached at the tail of the pipeline. " +
"Please check your pipeline configuration.", msg);
} finally {
ReferenceCountUtil.release(msg);
}
}方式二:
如果在handler中手动创建的byteBuf,遵循谁用完谁释放.
// 手动调用
ReferenceCountUtil.release(msg);
log.info("msg release");
1.1 创建ByteBuf
通常可以使用如下方式创建：
public static void main(String[] args) {
//创建一个容量为10的ByteBuf
ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(10);
System.out.println(buf);
}查看结果：
PooledUnsafeDirectByteBuf(ridx: 0, widx: 0, cap: 10)由结果我们看到初始化了一个池化的直接内存buf，也就是说如果我们使用默认类型创建，会返回一个直接内存。
ridx表示read index，即读取位置是0；
widx表示write index，即写入位置是0；
cap表示容量，是10；
1.1.1 直接内存 和 堆内存
使用如下方式创建一个堆内存：
//创建一个堆内存buf
ByteBuf heapBuffer = ByteBufAllocator.DEFAULT.heapBuffer(10);
关于堆内存是通过jvm进行回收的，所以我们不需要过多的干预。
使用如下方式创建一个直接内存
//创建一个直接内存buf
ByteBuf directBuffer = ByteBufAllocator.DEFAULT.directBuffer(10);
HeapByteBuffer也是我们最常用的方式指向对中内存byte[]的地址，当读取IO数据时先把数据拷贝到直接内存，再拷贝到jvm内存中，两次拷贝
而DirectByteBuffer直接指向直接内存，省去了一步拷贝工作，这种技术也叫零拷贝，读取数据更快
直接内存的特性：
直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用
直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放
对比
那问题就来了，既然直接内存IO速度都很快，为啥我们常用的确实HeapByteBuffer？
相比于堆内存，直接内存的分配时间较长，因为JVM内存是物理内存提前分配好的，属于虚拟机自己的内存分配肯定很快，而堆外内存需要重新向物理内存索要额外空间，肯定需要更长时间
还有一个重要原因：堆外内存不受GC管控，容易造成内存溢出(可以调用system.gc手动GC)
1.1.2 池化 和 非池化
无论在任何使用池的地方，例如数据库连接池，线程池等等。
池化最大的意义莫过于可以重用资源。
在Netty当中，引入池化机制，使得我们可以重用ByteBuf。
那么池化对我们使用ByteBuf提供了哪些优点？
使我们不必在每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵；即使是使用堆内存，也会增加 GC 压力。
可以重用池中 ByteBuf 实例，并且采用了与 jemalloc（http://jemalloc.net/） 类似的内存分配算法提升分配效率。
高并发时，池化功能更节约内存，减少内存溢出的可能
1.1.3 ByteBuf的组成
其源码如下：
public abstract class ByteBuf extends Object implements ReferenceCounted, Comparable<ByteBuf>
如上所示，是一个抽象类，继承自Object，实现了ReferenceCounted，Comparable<ByteBuf>。
ReferenceCounted是用于ByteBuf的回收工作，从其名称就能看出，是引用计数法，refCnt是引用数值；其提供了方法retain()，计数+1；release()方法，计数减1，当最终引用计数是0时，资源将被释放。此处不多做介绍。
Comparable主要用作比较。
就像普通的原始字节数组一样， ByteBuf使用从零开始的索引 。 这意味着第一个字节的索引始终为0 ，最后一个字节的索引始终为capacity - 1 。
ByteBuf提供了两个指针变量来支持顺序读写操作readerIndex用于读操作和writerIndex用于写操作。
如上所示的缓冲区划分方式，相比于之前学习的NIO中的ByteBuffer有一定的优势：
在ByteBuffer中，读写通过position这一个指针，需要通过flip()去切换；而在ByteBuf中，读写指针区分了，使用更加方便。
ByteBuf引入了自动扩容的能力。
读方法
写方法
通用方法
1.2.2 大端写入和小端写入
在前面的图中有一些方法是带有LE的，可以成为小端写入，比如
writeIntLE(int value)
其中的LE是 Little Endian 的缩写，可以理解为小字节序、低字节序。
与之相对应的就是 Big Endian，大字节序或高字节序。高字节序没有特别标注，因为在网络编程中通常使用的都是高字节序。如下面的代码就是高字节序：
writeInt(int value)
举例：写入两个数字888和888L
其二进制表示就是：
| 0000 0000 | 0000 0000 | 0000 0011 | 0111 1000 |
| 0000 0000 | 0000 0000 | 0000 0000 | 0000 0000 | 0000 0000 | 0000 0000 | 0000 0011 | 0111 1000 |
将其转换成字节数组就是
[0,0,3,120]
[0,0,0,0,0,0,3,120]
有如下测试代码：
public static void main(String[] args) {
int a = 888;
//大端写入
ByteBuf byteBuf1 = ByteBufAllocator.DEFAULT.buffer();
byteBuf1.writeInt(a);
byte[] bytes1 = new byte[4];
byteBuf1.readBytes(bytes1);
System.out.println(Arrays.toString(bytes1));
//小端写入
ByteBuf byteBuf2 = ByteBufAllocator.DEFAULT.buffer();
byteBuf2.writeIntLE(a);
byte[] bytes2 = new byte[4];
byteBuf2.readBytes(bytes2);
System.out.println(Arrays.toString(bytes2));
long b = 888L;
//大端写入
ByteBuf byteBuf3 = ByteBufAllocator.DEFAULT.buffer();
byteBuf3.writeLong(b);
byte[] bytes3 = new byte[8];
byteBuf3.readBytes(bytes3);
System.out.println(Arrays.toString(bytes3));
//小端写入
ByteBuf byteBuf4 = ByteBufAllocator.DEFAULT.buffer();
byteBuf4.writeLongLE(b);
byte[] bytes4 = new byte[8];
byteBuf4.readBytes(bytes4);
System.out.println(Arrays.toString(bytes4));
}结果:
[0, 0, 3, 120]
[120, 3, 0, 0]
[0, 0, 0, 0, 0, 0, 3, 120]
[120, 3, 0, 0, 0, 0, 0, 0]结论：int占4个字节，long占8个字节，大端写入是从左向右写；小端写入从右向左写。
网络传输通常使用大端传输（Big Endian）
1.2.3 set方式的写入不改变读写下标
还有一些列以set命名的写入方式，以下面为例：
public abstract ByteBuf setBytes(int index, byte[] src)
这一类方法要求我们指定index，从此index开始写入数据， 此方法不会修改此缓冲区的readerIndex或writerIndex 。
1.2.4 自动扩容
此处我们继续使用在1.2.1小节中的代码，继续增加5个长度：
public static void main(String[] args) {
//申请长度是10的buffer
ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer(10);
//写入5个长度的字节数组
byte[] bytes = new byte[]{1, 2, 3, 4, 5};
byteBuf.writeBytes(bytes);
System.out.println(byteBuf);
//再次写入5个
byteBuf.writeBytes(bytes);
System.out.println(byteBuf);
//再次写入5个
byteBuf.writeBytes(bytes);
System.out.println(byteBuf);
}
结果:
PooledUnsafeDirectByteBuf(ridx: 0, widx: 5, cap: 10)
PooledUnsafeDirectByteBuf(ridx: 0, widx: 10, cap: 10)
PooledUnsafeDirectByteBuf(ridx: 0, widx: 15, cap: 16)如上所示发现最后一条的容量变成了16。可是我们初始给的只有10，此处就是自动进行了扩容操作。
AbstractByteBuf中的ensureWritable0方法就是扩容方法，其扩容代码如下所示：
final void ensureWritable0(int minWritableBytes) {
//获取当前写入下标
final int writerIndex = writerIndex();
//预计写入后的下标
final int targetCapacity = writerIndex + minWritableBytes;
//如果小于初始容量，不扩容
if (targetCapacity <= capacity()) {
ensureAccessible();
return;
}
//超过最大容量，跑出异常
if (checkBounds && targetCapacity > maxCapacity) {
ensureAccessible();
throw new IndexOutOfBoundsException(String.format(
"writerIndex(%d) + minWritableBytes(%d) exceeds maxCapacity(%d): %s",
writerIndex, minWritableBytes, maxCapacity, this));
}
// 将目标容量归一化为2的幂。
//此处用初始化ByteBuf时指定的默认大小，会根据初始赋值计算出2的幂
//会用该默认大小 减去 写入位置的下标，得到能写入的大小的最大值
final int fastWritable = maxFastWritableBytes();
//如果能写入的最大值 大于 需要写入的容量，则将能写入的值 加上 当前写入下标做为新的容量
//否则使用calculateNewCapacity去分配
int newCapacity = fastWritable >= minWritableBytes ? writerIndex + fastWritable
: alloc().calculateNewCapacity(targetCapacity, maxCapacity);
// Adjust to the new capacity.
capacity(newCapacity);
}
maxFastWritableBytes内部会获取一个maxLength，这个值是在allocate缓冲区是计算出来的2的幂，且小于512的，用它减去当前写入index的位置。
public int maxFastWritableBytes() {
return Math.min(maxLength, maxCapacity()) - writerIndex;
}下面看下初始化时如何计算出的这个默认值：
int normalizeCapacity(int reqCapacity) {
checkPositiveOrZero(reqCapacity, "reqCapacity");
// 如果此处大于chunkSize（16777217），则将这个值作为分配的值
if (reqCapacity >= chunkSize) {
return directMemoryCacheAlignment == 0 ? reqCapacity : alignCapacity(reqCapacity);
}
// >= 512
if (!isTiny(reqCapacity)) {
// 翻倍（乘以2）
int normalizedCapacity = reqCapacity;
normalizedCapacity --;
normalizedCapacity |= normalizedCapacity >>> 1;
normalizedCapacity |= normalizedCapacity >>> 2;
normalizedCapacity |= normalizedCapacity >>> 4;
normalizedCapacity |= normalizedCapacity >>> 8;
normalizedCapacity |= normalizedCapacity >>> 16;
normalizedCapacity ++;
if (normalizedCapacity < 0) {
normalizedCapacity >>>= 1;
}
assert directMemoryCacheAlignment == 0 || (normalizedCapacity & directMemoryCacheAlignmentMask) == 0;
return normalizedCapacity;
}
if (directMemoryCacheAlignment > 0) {
return alignCapacity(reqCapacity);
}
// Quantum-spaced
if ((reqCapacity & 15) == 0) {
return reqCapacity;
}
return (reqCapacity & ~15) + 16;
}在上面的代码中，有一个默认值chunkSize，如果分配的值大于chunkSize（16777217），则将这个值作为分配的值，而在后面扩容时ensureWritable0中的判断：
int newCapacity = fastWritable >= minWritableBytes ? writerIndex + fastWritable : this.alloc().calculateNewCapacity(targetCapacity, this.maxCapacity);此时会走重新计算容量的方法：
his.alloc().calculateNewCapacity(targetCapacity, this.maxCapacity);public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
checkPositiveOrZero(minNewCapacity, "minNewCapacity");
//如果需要的容量大于最大容量，则抛出异常
if (minNewCapacity > maxCapacity) {
throw new IllegalArgumentException(String.format(
"minNewCapacity: %d (expected: not greater than maxCapacity(%d)",
minNewCapacity, maxCapacity));
}
//定义一页4M的阈值
final int threshold = CALCULATE_THRESHOLD; // 4 MiB page
//如果需要的容量等于页阈值，就返回该阈值
if (minNewCapacity == threshold) {
return threshold;
}
// 如果需要容量大于 页阈值
if (minNewCapacity > threshold) {
//计算最小需要几个页阈值的大小，并赋予一个新的容量值
int newCapacity = minNewCapacity / threshold * threshold;
//如果新容量，比 最大容量 - 一个分页还要大，即剩余不到一个页，则赋予最大容量
if (newCapacity > maxCapacity - threshold) {
newCapacity = maxCapacity;
} else {
//在赋予的新容量基础上 在加一个页容量
newCapacity += threshold;
}
return newCapacity;
}
// 如果没有超过页阈值，则从64开始，最大增加到4
int newCapacity = 64;
while (newCapacity < minNewCapacity) {
//左移一位，即*2
newCapacity <<= 1;
}
// 取新赋予容量和最大值中的最小值
return Math.min(newCapacity, maxCapacity);
}1.3.2 get方式的读取不改变读下标
1）使用get开头的方法，不会修改index。
2）使用mark：
public static void main(String[] args) {
//申请长度是10的buffer
ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer(10);
//写入5个长度的字节数组
byte[] bytes = new byte[]{1, 2, 3, 4, 5};
byteBuf.writeBytes(bytes);
//设置一个mark
byteBuf.markReaderIndex();
//读取自己数组
System.out.println(byteBuf.readByte());
System.out.println(byteBuf);
//重置到mark
byteBuf.resetReaderIndex();
//读取自己数组
System.out.println(byteBuf.readByte());
System.out.println(byteBuf);
}
结果:
1
PooledUnsafeDirectByteBuf(ridx: 1, widx: 5, cap: 10)
1
PooledUnsafeDirectByteBuf(ridx: 1, widx: 5, cap: 10)二、直接内存回收原理
非池化的ByteBuf可以通过JVM GC自动回收，也推荐手动回收UnpooledDirectByteBuf等使用堆外内存的ByteBuf；
另一类是池化的ByteBuf，包括pooledHeapByteBuf、pooledDirectByteBuf等等，其先申请一块大内存池，在内存池中分配空间，对于这种应用级别的内存二次分配，就需要手动对池化的ByteBuf进行释放，否则就有可能出现内存泄露的问题。
在前面的文章中，我们简单聊到过ByteBuf的结构：
public abstract class ByteBuf implements ReferenceCounted
如上所示，其实现了ReferenceCounted的接口，接口翻译过来叫做“引用计数”。
相信学过jvm GC的同学应该有所了解“引用计数法”，当一个对象有引用时，我们就对计数器加1，反之就减1，但是引用计数法无法处理环形垃圾，所以后面提出了“根可达算法”，简单提一下，需要了解细节的朋友可以看我的专题【JVM】。
此处的引用计数，用于ByteBuf的直接内存回收，我们看下其主要的方法：
public interface ReferenceCounted {
/**
* 返回当前对象的引用计数
*/
int refCnt();
/**
* 将引用计数增加1
*/
ReferenceCounted retain();
/**
* 按指定的increment增加引用计数
*/
ReferenceCounted retain(int increment);
/**
* 将引用计数减少1，并在引用计数达到0解除分配此对象
*/
boolean release();
/**
* 将引用计数减少指定的decrement ，如果引用计数达到0则取消分配此对象。
*/
boolean release(int decrement);
}所有的ByteBuf都会实现这个接口，当一个新的ReferenceCounted被实例化时，它以1的引用计数开始。 retain()增加引用计数，而release()减少引用计数。 如果引用计数减少到0 ，对象将被释放，并且访问释放的对象通常会导致访问冲突。
public static void main(String[] args) {
ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
//打印当前的引用计数
System.out.println("初始化后的引用计数" + byteBuf.refCnt());
//释放引用计数
byteBuf.release();
//打印当前的引用计数
System.out.println("释放后的引用计数" + byteBuf.refCnt());
//调用byteBuf
try {
byteBuf.writeInt(888);
} catch (Exception e) {
System.out.println("释放后调用异常：" + e);
}
//增加引用计数
try {
byteBuf.retain();
} catch (Exception e) {
System.out.println("释放后增加引用计数异常：" + e);
}
// 重新分配
byteBuf = ByteBufAllocator.DEFAULT.buffer();
//调用byteBuf
byteBuf.writeInt(888);
System.out.println("重新分配后的引用计数" + byteBuf.refCnt());
}
结果:
初始化后的引用计数1
释放后的引用计数0
释放后调用异常：io.netty.util.IllegalReferenceCountException: refCnt: 0
释放后增加引用计数异常：io.netty.util.IllegalReferenceCountException: refCnt: 0, increment: 1
重新分配后的引用计数1当引用计数变为0后，整个内存就释放了，再次使用会抛出异常，重新尝试增加引用计数也会跑出异常，只能进行重新分配。
PooledUnsafeDirectByteBuf中内存释放
PooledUnsafeDirectByteBuf中内存释放的入口方法是其父类AbstractReferenceCountedByteBuf中的release方法:
@Override
public boolean release() {
return release0(1);
}这里调用了release0, 跟进去
private boolean release0(int decrement) {
for (;;) {
int refCnt = this.refCnt;
if (refCnt < decrement) {
throw new IllegalReferenceCountException(refCnt, -decrement);
}
if (refCntUpdater.compareAndSet(this, refCnt, refCnt - decrement)) {
if (refCnt == decrement) {
deallocate();
return true;
}
return false;
}
}
}if (refCnt == decrement) 中判断当前byteBuf是否没有被引用了, 如果没有被引用, 则通过deallocate()方法进行释放
因为我们是以PooledUnsafeDirectByteBuf为例, 所以这里会调用其父类PooledByteBuf的deallocate方法:
protected final void deallocate() {
if (handle >= 0) {
final long handle = this.handle;
this.handle = -1;
memory = null;
chunk.arena.free(chunk, handle, maxLength, cache);
recycle();
}
}this.handle = -1表示当前的ByteBuf不再指向任何一块内存
memory = null这里将memory也设置为null
chunk.arena.free(chunk, handle, maxLength, cache)这一步是将ByteBuf的内存进行释放
recycle()是将对象放入的对象回收站, 循环利用
我们首先分析free方法
void free(PoolChunk<T> chunk, long handle, int normCapacity, PoolThreadCache cache) {
//是否为unpooled
if (chunk.unpooled) {
int size = chunk.chunkSize();
destroyChunk(chunk);
activeBytesHuge.add(-size);
deallocationsHuge.increment();
} else {
//那种级别的Size
SizeClass sizeClass = sizeClass(normCapacity);
//加到缓存里
if (cache != null && cache.add(this, chunk, handle, normCapacity, sizeClass)) {
return;
}
//将缓存对象标记为未使用
freeChunk(chunk, handle, sizeClass);
}
}首先判断是不是unpooled, 我们这里是Pooled, 所以会走到else块中:
sizeClass(normCapacity)计算是哪种级别的size, 我们按照tiny级别进行分析
cache.add(this, chunk, handle, normCapacity, sizeClass)是将当前当前ByteBuf进行缓存
再分配ByteBuf时首先在缓存上分配, cache.add这步, 就是将其缓存的过程, 跟进去:
boolean add(PoolArena<?> area, PoolChunk chunk, long handle, int normCapacity, SizeClass sizeClass) {
//拿到MemoryRegionCache节点
MemoryRegionCache<?> cache = cache(area, normCapacity, sizeClass);
if (cache == null) {
return false;
}
//将chunk, 和handle封装成实体加到queue里面
return cache.add(chunk, handle);
}首先根据根据类型拿到相关类型缓存节点, 这里会根据不同的内存规格去找不同的对象, 我们简单回顾一下, 每个缓存对象都包含一个queue, queue中每个节点是entry, 每一个entry中包含一个chunk和handle, 可以指向唯一的连续的内存
我们跟到cache中
private MemoryRegionCache<?> cache(PoolArena<?> area, int normCapacity, SizeClass sizeClass) {
switch (sizeClass) {
case Normal:
return cacheForNormal(area, normCapacity);
case Small:
return cacheForSmall(area, normCapacity);
case Tiny:
return cacheForTiny(area, normCapacity);
default:
throw new Error();
}
}假设我们是tiny类型, 这里就会走到cacheForTiny(area, normCapacity)方法中, 跟进去:
privateMemoryRegionCache<?> cacheForTiny(PoolArena<?> area, intnormCapacity) {
intidx = PoolArena.tinyIdx(normCapacity);
if(area.isDirect()) {
returncache(tinySubPageDirectCaches, idx);
}
returncache(tinySubPageHeapCaches, idx);
}这个方法我们之前剖析过, 就是根据大小找到第几个缓存中的第几个缓存, 拿到下标之后, 通过cache去超相对应的缓存对象:
privatestatic<T> MemoryRegionCache<T> cache(MemoryRegionCache<T>[] cache, intidx) {
if(cache == null|| idx > cache.length - 1) {
returnnull;
}
returncache[idx];
}我们这里看到, 是直接通过下标拿的缓存对象
cache.add
这里的cache对象调用了一个add方法, 这个方法就是将chunk和handle封装成一个entry加到queue里面
我们跟到add方法中:
public final boolean add(PoolChunk<T> chunk, long handle) {
Entry<T> entry = newEntry(chunk, handle);
boolean queued = queue.offer(entry);
if (!queued) {
entry.recycle();
}
return queued;
}从在缓存中分配的时候从queue弹出一个entry, 会放到一个对象池里面, 而这里Entry<T> entry = newEntry(chunk, handle)就是从对象池里去取一个entry对象, 然后将chunk和handle进行赋值
然后通过queue.offer(entry)加到queue中
回到free方法中
void free(PoolChunk<T> chunk, long handle, int normCapacity, PoolThreadCache cache) {
//是否为unpooled
if (chunk.unpooled) {
int size = chunk.chunkSize();
destroyChunk(chunk);
activeBytesHuge.add(-size);
deallocationsHuge.increment();
} else {
//那种级别的Size
SizeClass sizeClass = sizeClass(normCapacity);
//加到缓存里
if (cache != null && cache.add(this, chunk, handle, normCapacity, sizeClass)) {
return;
}
freeChunk(chunk, handle, sizeClass);
}
}这里加到缓存之后, 如果成功, 就会return, 如果不成功, 就会调用freeChunk(chunk, handle, sizeClass)方法, 这个方法的意义是, 将原先给ByteBuf分配的内存区段标记为未使用
跟进freeChunk简单分析下:
void freeChunk(PoolChunk<T> chunk, long handle, SizeClass sizeClass) {
final boolean destroyChunk;
synchronized (this) {
switch (sizeClass) {
case Normal:
++deallocationsNormal;
break;
case Small:
++deallocationsSmall;
break;
case Tiny:
++deallocationsTiny;
break;
default:
throw new Error();
}
destroyChunk = !chunk.parent.free(chunk, handle);
}
if (destroyChunk) {
destroyChunk(chunk);
}
}我们再跟到free方法中:
boolean free(PoolChunk<T> chunk, long handle) {
//通过chunk释放一段连续的内存
chunk.free(handle);
if (chunk.usage() < minUsage) {
remove(chunk);
return move0(chunk);
}
return true;
}chunk.free(handle)的意思是通过chunk释放一段连续的内存
再跟到free方法中:
void free(long handle) {
int memoryMapIdx = memoryMapIdx(handle);
int bitmapIdx = bitmapIdx(handle);
//判断是当前缓冲区分配的级别是Page还是Subpage, 如果是Subpage, 则会找到相关的Subpage将其位图标记为0
//如果不是subpage, 这里通过分配内存的反向标记, 将该内存标记为未使用
if (bitmapIdx != 0) {
PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
assert subpage != null && subpage.doNotDestroy;
PoolSubpage<T> head = arena.findSubpagePoolHead(subpage.elemSize);
synchronized (head) {
if (subpage.free(head, bitmapIdx & 0x3FFFFFFF)) {
return;
}
}
}
freeBytes += runLength(memoryMapIdx);
setValue(memoryMapIdx, depth(memoryMapIdx));
updateParentsFree(memoryMapIdx);
}在java jdk中也有使用直接内存的地方
DirectByteBuffer帮我们简化了直接内存的使用，我们不需要直接操作Unsafe类来进行直接内存的申请与释放，那么其是如何实现的呢？
直接内存的申请：
在DirectByteBuffer实例通过构造方法创建的时候，会通过Unsafe类的allocateMemory方法 帮我们申请直接内存资源。
直接内存的释放：
DirectByteBuffer本身是一个Java对象，其是位于堆内存中的，JDK的GC机制可以自动帮我们回收，但是其申请的直接内存，不再GC范围之内，无法自动回收。好在JDK提供了一种机制，可以为堆内存对象注册一个cleaner钩子函数(其实就是实现Runnable接口的子类)持有自己的幽灵引用，当幽灵引用探测到GC可以释放Java堆内的DirectByteBuffer对象时，通过回调来执行释放堆外内存的逻辑。
Cleaner的实现是可以避免重复释放内存的，换句话说如果您拿到DirectByteBuffer的Cleaner对象，直接调用Cleaner的clean()是可以完成手动释放DirectByteBuffer的内存的，即使当GC时JVM回调再次释放也是不会有问题的。
jdk1.9中 Cleaner被正式修正发布到了java.lang.ref.Cleaner中。
class DirectByteBuffer extends MappedByteBuffer implements DirectBuffer
{
....
//构造方法
DirectByteBuffer(int cap) { // package-private
super(-1, 0, cap, cap);
boolean pa = VM.isDirectMemoryPageAligned();
int ps = Bits.pageSize();
long size = Math.max(1L, (long)cap + (pa ? ps : 0));//对申请的直接内存大小，进行重新计算
Bits.reserveMemory(size, cap);
long base = 0;
try {
base = unsafe.allocateMemory(size); //分配直接内存，base表示的是直接内存的开始地址
} catch (OutOfMemoryError x) {
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;
}
//注册钩子函数，释放直接内存
//当堆内存对象被GC回收的时候，会回调run方法，我们可以在这个方法中执行释放DirectByteBuffer引用的直接内存，即在run方法中调用Unsafe 的freeMemory 方法。
cleaner = Cleaner.create(this, new Deallocator(base, size, cap));
att = null;
}
....
}
#Cleaner.create()注册了一个钩子函数，用于清除直接内存的引用。
第一个参数是一个堆内存对象
第二个参数是一个Runnable任务,表示这个堆内存对象被回收的时候，需要执行的回调方法。private static class Deallocator
implements Runnable
{
private static Unsafe unsafe = Unsafe.getUnsafe();
private long address;
private long size;
private int capacity;
private Deallocator(long address, long size, int capacity) {
assert (address != 0);
this.address = address;
this.size = size;
this.capacity = capacity;
}
public void run() {
if (address == 0) {
// Paranoia
return;
}
unsafe.freeMemory(address);//清除直接内存
address = 0;
Bits.unreserveMemory(size, capacity);
}
}三、内存释放使用方式
3.1 手动释放
前面简单了解了关于内存释放的内容，那么我们应该如何使用呢？是不是可以向我们习惯的java代码一样，在finally当中调用呢？
try {
} finally {
//会有几率造成内存溢出的
byteBuf.release();
}直接给出结论，是不行的。
前面我们介绍时候就说过，会有几率造成内存溢出的，即使不会发生也会造成内存的浪费。
前面的文章当中，我们学习了Pipeline和Handler。通常我们会将一个byteBuf传递给另一个channelHandler去处理，是存在一个传递性的。这里面存在两种情况：
假设一共有5个channelHandler，在第二个当中，将byteBuf转换成了java对象，然后将对象传递给第三个channelHandler，此时byteBuf就没有用了，所以此时就应该释放。
一直以byteBuf传递，直到最后一个channelHandler才进行释放。
总结一句话：最后谁用完了，谁就负责释放。
建议：如果确定这个buf在最后时刻用完了，而又无法确定当前有多少个引用计数，使用如下两种方式释放：
循环调用release()，直到返回true。
通过refCnt()获取当前的引用计数，然后调用release(int refCnt)释放
3.2 tail和head自动释放
还记得前面将Pipeline和Handler时，提到了关于head和tail的概念，除了我们自己添加的Handler以外，会默认有一个头和尾的处理器。
在这两个处理器当中，也会有自动回收内存的保底能力，但是前提是要求我们将byteBuf传递到head或tail当中才行，对于中途就转换类型的，仍然需要我们自己去释放资源。
下面我们简单跟踪下源码，看看是如何实现的内存释放。
我们跟踪pipeline的addLast方法，跟踪到了AbstractChannelHandlerContext这个抽象类，其有两个实现类：
3.2.1 TailContext
首先看tail处理器，实现了ChannelInboundHandler，即入站处理器，进行入站首尾工作。
final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler找到channelRead方法：
public void channelRead(ChannelHandlerContext ctx, Object msg){
DefaultChannelPipeline.this.onUnhandledInboundMessage(ctx, msg);
}继续跟踪onUnhandledInboundMessage
protected void onUnhandledInboundMessage(Object msg){
try{
logger.debug("Discarded inbound message {} that reached at the tail of the pipeline. Please check your pipeline configuration.", msg);
} finally {
ReferenceCountUtil.release(msg);
}
}发现其中的引用计数工具类，调用了release方法：
ReferenceCountUtil.release(msg);
判断msg是否是实现了ReferenceCounted ？是就进行是否，否则返回false。
public static boolean release(Object msg){
return msg instanceof ReferenceCounted ? ((ReferenceCounted)msg).release():false;
}3.2.1 HeadContext
查看HeadContext，实现了ChannelOutboundHandler，即出站处理器，进行出站首尾工作。
final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler找到其write方法：
public void write(ChannelHandlerContext ctx,Object msg,ChannelPromise promise){
this.unsafe.write(msg, promise);
}继续跟踪write：
public final void write(Object msg, ChannelPromise promise) {
this.assertEventLoop();
ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
if (outboundBuffer == null) {
this.safeSetFailure(promise, this.newClosedChannelException(AbstractChannel.this.initialCloseCause));
ReferenceCountUtil.release(msg);
} else {
int size;
try {
msg = AbstractChannel.this.filterOutboundMessage(msg);
size = AbstractChannel.this.pipeline.estimatorHandle().size(msg);
if (size < 0) {
size = 0;
}
} catch (Throwable var6) {
this.safeSetFailure(promise, var6);
ReferenceCountUtil.release(msg);
return;
}
outboundBuffer.addMessage(msg, size, promise);
}
}在上面的代码中，仍然发现了
ReferenceCountUtil.release(msg)
无论是head，还是tail，都需要将buf传递过来，才能进行释放。
一、零拷贝
先声明一点，如果要使用以下方法的话，可能要配合前面文章介绍的retain()方法去增加引用计数，否则原ByteBuf被release()后，则会导致我们拷贝出来的buf使用失败，导致异常。
1.1 slice
“零拷贝”的体现之一，对原始 ByteBuf 进行切片，切分成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制，只是多了引用到分片后的Bytebuf，然而还是使用的原始 ByteBuf 的内存，切片后的 ByteBuf 维护独立的 read，write 指针，修改子分片，会修改原ByteBuf。
readSlice()方法不会更改原始byteBuf
public static void main(String[] args) {
ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer(10);
byteBuf.writeBytes(new byte[]{1,2,3,4,5,6,7,8,9,0});
printBuf(byteBuf);
//分片1
ByteBuf slice1 = byteBuf.slice(0, 5);
printBuf(slice1);
//分片2
ByteBuf slice2 = byteBuf.slice(5, 5);
printBuf(slice2);
//将最后一位0修改成10
slice2.setByte(4,10);
printBuf(slice2);
//打印修改后的byteBuf
printBuf(byteBuf);
}
static void printBuf(ByteBuf byteBuf){
StringBuilder stringBuilder = new StringBuilder();
for (int i = 0; i< byteBuf.writerIndex();i++) {
stringBuilder.append(byteBuf.getByte(i));
}
System.out.println(stringBuilder);
}
结果:
1234567890
12345
67890
678910
12345678910注意：slice后的分片，不能再次写入新的数据，这回影响原ByteBuf。
1.2 duplicate
“零拷贝”的体现之一，拷贝了原始 ByteBuf 所有内容，长度仍然以byteBuf为准，不能写入新数据，也是与原始 ByteBuf 使用同一块底层内存，只是读写指针是独立的。
public class DuplicateTest {
public static void main(String[] args) {
ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer(10);
byteBuf.writeBytes(new byte[]{1,2,3,4,5,6,7,8,9,0});
//拷贝一块buf
ByteBuf duplicate = byteBuf.duplicate();
printBuf(duplicate);
//将最后一位0修改成10，看一下byteBuf,原始的也改了
duplicate.setByte(9,10);
printBuf(byteBuf);
// 写入新数据11，看byteBuf,报错
duplicate.writeByte(11);
printBuf(byteBuf);
}
static void printBuf(ByteBuf byteBuf){
StringBuilder stringBuilder = new StringBuilder();
for (int i = 0; i< byteBuf.writerIndex();i++) {
stringBuilder.append(byteBuf.getByte(i));
}
System.out.println(stringBuilder);
}
}1.3 CompositeByteBuf
“零拷贝”的体现之一，可以将多个 ByteBuf 合并为一个逻辑上的 ByteBuf，避免拷贝。
public class CompositeByteBufTest {
public static void main(String[] args) {
ByteBuf byteBuf1 = ByteBufAllocator.DEFAULT.buffer(5);
byteBuf1.writeBytes(new byte[]{1, 2, 3, 4, 5});
ByteBuf byteBuf2 = ByteBufAllocator.DEFAULT.buffer(5);
byteBuf2.writeBytes(new byte[]{6, 7, 8, 9, 0});
CompositeByteBuf compositeByteBuf = ByteBufAllocator.DEFAULT.compositeBuffer();
// 组合两个byteBuf，主要要使用带有increaseWriteIndex的，否则会失败。
compositeByteBuf.addComponents(true,byteBuf1, byteBuf2);
printBuf(compositeByteBuf);
}
static void printBuf(ByteBuf byteBuf){
StringBuilder stringBuilder = new StringBuilder();
for (int i = 0; i< byteBuf.writerIndex();i++) {
stringBuilder.append(byteBuf.getByte(i));
}
System.out.println(stringBuilder);
}
}1.4 Unpooled的wrappedBuffer方法
Unpooled 是一个工具类，提供了非池化的 ByteBuf 创建、组合、复制等操作。
这里仅介绍其跟“零拷贝”相关的 wrappedBuffer 方法
使用示例：
public static void main(String[] args) {
ByteBuf byteBuf1 = ByteBufAllocator.DEFAULT.buffer(5);
byteBuf1.writeBytes(new byte[]{1, 2, 3, 4, 5});
ByteBuf byteBuf2 = ByteBufAllocator.DEFAULT.buffer(5);
byteBuf2.writeBytes(new byte[]{6, 7, 8, 9, 0});
// CompositeByteBuf compositeByteBuf = ByteBufAllocator.DEFAULT.compositeBuffer();
// // 组合两个byteBuf，主要要使用带有increaseWriteIndex的，否则会失败。
// compositeByteBuf.addComponents(true,byteBuf1, byteBuf2);
// 组合两个byteBuf，底层使用CompositeByteBuf。
ByteBuf wrappedBuffer = Unpooled.wrappedBuffer(byteBuf1, byteBuf2);
printBuf(wrappedBuffer);
}
static void printBuf(ByteBuf byteBuf){
StringBuilder stringBuilder = new StringBuilder();
for (int i = 0; i< byteBuf.writerIndex();i++) {
stringBuilder.append(byteBuf.getByte(i));
}
System.out.println(stringBuilder);
}copiedBuffer复制数据组合多个ByteBuf
copiedBuffer，和wrappedBuffer最大的区别是，该方法会实现数据复制，这种是深拷贝,不影响原先的ByteBuf.
下面代码演示了copiedBuffer和wrappedbuffer的区别，可以看到在case标注的位置中，修改了原始ByteBuf的值，并没有影响到allBb。
public static void main(String[] args) {
ByteBuf header= ByteBufAllocator.DEFAULT.buffer();//可自动扩容
header.writeCharSequence("header", CharsetUtil.UTF_8);
ByteBuf body=ByteBufAllocator.DEFAULT.buffer();
body.writeCharSequence("body", CharsetUtil.UTF_8);
ByteBuf allBb=Unpooled.copiedBuffer(header,body);
log(allBb);
header.setCharSequence(0,"Newer0",CharsetUtil.UTF_8); //case
log(allBb);
}二、深度拷贝
ByteBuf提供了copy方法，这一类方法是真正的拷贝原ByteBuf到新的内存，返回一个新的ByteBuf，与原ByteBuf没有关系。
提供两个拷贝，一个是全量；一个指定位置和长度。
public abstract ByteBuf copy();
public abstract ByteBuf copy(int index, int length);