Netty 权威指南笔记(五):ByteBuf 源码解读
功能介绍
Java 本身提供了 ByteBuffer 类,为什么 Netty 还要搞一个 ByteBuf 类呢?因为 ByteBuffer 类有着许多缺点:
- ByteBuffer 长度固定,无法动态伸缩。
- ByteBuffer 只有一个位置指针 position,读写的时候需要手工调用 flip 和 rewind 方法进行模式转换,操作繁琐,容易出错。
- 功能太少,缺少一些高级特性。
为了弥补这些不足,Netty 提供了自己的缓冲区类实现 ByteBuf。有什么特点呢?
- 两个位置指针协助缓冲区的读写操作:readerIndex、writerIndex,读写之间不需要调整指针位置,大大简化了读写操作。
- 可以动态扩容。
- 当有部分内容已经读取完成时,可以通过 discard 操作对缓冲区进行整理,在不重新申请内存的情况下,增大可写字节数目。
- 支持标记和回滚的功能。
- 支持在 ByteBuf 中查找某个字符串。
- 派生出另一个 ByteBuf:duplicate、copy、slice。
- 转化成标准的 ByteBuffer,这是因为在使用 NIO 进行网络读写时,操作的对象还是 JDK 标准的 ByteBuffer。
- 随机读写。
源码分析
继承关系
ByteBuf 的主要功能类继承关系如下图所示:
从内存分配的角度看,ByteBuf 可以分为两类:
- 堆内存字节缓冲区 HeapByteBuf:优点是内存分配和回收速度快,可以被 JVM 自动回收。缺点是,如果进行 Socket 的 I/O 读写,需要额外做一次内存复制,在堆内存缓冲区和内核 Channel 之间进行复制,性能会有一定程度下降。
- 直接内存字节缓冲区 DirectByteBuf:非堆内存,直接在堆外进行分配。相比于堆内存,内存分配和回收稍慢,但是可以减少复制,提升性能。
两种内存,各有利弊。Netty 最佳实践表明:在 I/O 通信线程的读写缓冲区使用 DirectByteBuf,后端业务消息的编解码模块使用 HeapByteBuf,这样组合可以达到性能最优。
从内存回收的角度看,ByteBuf 也分为两类:基于对象池的 ByteBuf 和普通 ByteBuf。两者区别在于基于对象池的 ByteBuf 可以重用 ByteBuf 对象,它自己维护了一个内存池,可以循环利用创建的 ByteBuf,提升内存的使用效率,降低由于高负载导致的频繁 GC。内存池的缺点是管理和维护比较复杂,使用时需要更加谨慎。
下面我们对一些关键类进行分析和解读。
AbstractByteBuf
AbstractByteBuf 都做了哪些事儿呢?我们先看一下其主要的成员变量:
- 读写指针。
- 用于标记回滚的 marked 读写指针。
- 最大容量 maxCapacity,用于进行内存保护。
- 与本 ByteBuf 大小端属性相反的 ByteBuf:SwappedByteBuf。
我们发现这里没有真正存储数据的数据结构,例如 byte 数组或 DirectByteBuffer,原因是这里还不知道子类是要基于堆内存还是直接内存。
public abstract class AbstractByteBuf extends ByteBuf {
static final ResourceLeakDetector<ByteBuf> leakDetector = new ResourceLeakDetector<ByteBuf>(ByteBuf.class);
int readerIndex;
int writerIndex;
private int markedReaderIndex;
private int markedWriterIndex;
private int maxCapacity;
private SwappedByteBuf swappedBuf;
}
接下来我们看看读操作 readBytes 方法,AbstractByteBuf 类做了什么呢?
- 在 checkReadableBytes 方法中,检查入参有效性。
- 修改读指针 readerIndex。
@Override
public ByteBuf readBytes(byte[] dst, int dstIndex, int length) {
checkReadableBytes(length);
// getBytes 方法未在 AbstractByteBuf 中实现
getBytes(readerIndex, dst, dstIndex, length);
readerIndex += length;
return this;
}
protected final void checkReadableBytes(int minimumReadableBytes) {
ensureAccessible();
if (minimumReadableBytes < 0) {
throw new IllegalArgumentException("minimumReadableBytes: " + minimumReadableBytes + " (expected: >= 0)");
}
if (readerIndex > writerIndex - minimumReadableBytes) {
throw new IndexOutOfBoundsException(String.format(
"readerIndex(%d) + length(%d) exceeds writerIndex(%d): %s",
readerIndex, minimumReadableBytes, writerIndex, this));
}
}
从当前 ByteBuf 中复制数据到 dst 是在 getBytes 方法中,该方法未在 AbstractByteBuf 中实现,也是因为此时具体如何存储数据尚不确定。
下面我们看一下写操作 writeBytes 方法,AbstractByteBuf 负责实现了哪些操作呢?
- 有效性检查,如果引用计数 refCnt 为 0,表示该 ByteBuf 已经被回收,不能再写入。
- 输入参数有效性检查:要写入的数据量不能小于 0,写入之后总数据量也不能大于最大容量。
- 当容量不足时,如果尚未超过最大容量,则进行扩容。
- 修改写指针。
@Override
public ByteBuf writeBytes(byte[] src, int srcIndex, int length) {
ensureAccessible();
ensureWritable(length);
// setBytes 交给子类实现。
setBytes(writerIndex, src, srcIndex, length);
writerIndex += length;
return this;
}
protected final void ensureAccessible() {
if (refCnt() == 0) {
throw new IllegalReferenceCountException(0);
}
}
@Override
public ByteBuf ensureWritable(int minWritableBytes) {
if (minWritableBytes < 0) {
throw new IllegalArgumentException(String.format(
"minWritableBytes: %d (expected: >= 0)", minWritableBytes));
}
if (minWritableBytes <= writableBytes()) {
return this;
}
if (minWritableBytes > maxCapacity - writerIndex) {
throw new IndexOutOfBoundsException(String.format(
"writerIndex(%d) + minWritableBytes(%d) exceeds maxCapacity(%d): %s",
writerIndex, minWritableBytes, maxCapacity, this));
}
// Normalize the current capacity to the power of 2.
int newCapacity = alloc().calculateNewCapacity(writerIndex + minWritableBytes, maxCapacity);
// 具体扩容操作由子类实现。
capacity(newCapacity);
return this;
}
在读写操作中,AbstractByteBuf 主要负责参数校验、读写指针修改,以及写操作时的扩容计算。
除此之外,AbstractByteBuf 还提供了以下功能:
- 操作索引:修改读写指针、mark & reset。
- 重用缓冲区:discardReadBytes。
- 丢弃部分数据:skipBytes。因为丢弃时,只需要修改读指针即可,与数据具体如何存储无关。
总结:在 AbstractByteBuf 中实现的是各个子类中通用的功能。
AbstractReferenceCountedByteBuf
从类名可以看出来,该类主要提供引用计数的功能,类似于 JVM 内存回收的对象引用计数器,用于跟踪对象的分配和回收,实现手动控制内存回收。
首先,我们看一下其成员变量:
- refCnt:记录对象引用次数。
- refCntUpdater:用于对 refCnt 进行原子更新。
private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> refCntUpdater;
static {
AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> updater =
PlatformDependent.newAtomicIntegerFieldUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");
if (updater == null) {
updater = AtomicIntegerFieldUpdater.newUpdater(AbstractReferenceCountedByteBuf.class, "refCnt");
}
refCntUpdater = updater;
}
private volatile int refCnt = 1;
接下来,我们看一下增加引用计数的 retain 方法。该方法是用 CAS 操作对 refCnt 进行加 1。另外,refCnt 值为 0 或 Integer.MAX_VALUE 值不能再操作,会抛出异常。
@Override
public ByteBuf retain() {
for (;;) {
int refCnt = this.refCnt;
if (refCnt == 0) {
throw new IllegalReferenceCountException(0, 1);
}
if (refCnt == Integer.MAX_VALUE) {
throw new IllegalReferenceCountException(Integer.MAX_VALUE, 1);
}
if (refCntUpdater.compareAndSet(this, refCnt, refCnt + 1)) {
break;
}
}
return this;
}
另一个 release 方法表示释放资源,会将引用计数 refCnt 减 1,如果当前 refCnt 等于 1,减 1 之后等于 0,表示对象已经没有被引用,可以被回收了,会调用 deallocate 方法释放内存。
@Override
public final boolean release() {
for (;;) {
int refCnt = this.refCnt;
if (refCnt == 0) {
throw new IllegalReferenceCountException(0, -1);
}
if (refCntUpdater.compareAndSet(this, refCnt, refCnt - 1)) {
if (refCnt == 1) {
deallocate();
return true;
}
return false;
}
}
}
在 UnpooledHeapByteBuf 中,释放内存仅仅是把 array 数组置为 null,剩下的内存回收工作交由 JVM 来完成。
// in UnpooledHeapByteBuf.java
private byte[] array;
@Override
protected void deallocate() {
array = null;
}
在 UnpooledDirectByteBuf 中,则是调用 PlatformDependent.freeDirectBuffer 来释放直接内存。
// in UnpooledDirectByteBuf.java
@Override
protected void deallocate() {
ByteBuffer buffer = this.buffer;
if (buffer == null) {
return;
}
this.buffer = null;
if (!doNotFree) {
freeDirect(buffer);
}
}
protected void freeDirect(ByteBuffer buffer) {
PlatformDependent.freeDirectBuffer(buffer);
}
UnpooledHeapByteBuf
UnpooledHeapByteBuf 是基于堆内存进行内存分配的字节缓冲区,它没有基于对象池实现,意味着每次 I/O 读写都会创建一个新的 UnpooledHeapByteBuf 对象,频繁进行内存的分配和释放对性能会有一定的影响,但是相对堆外内存的申请和释放,成本稍低。
相比于 PooledHeapByteBuf,不需要自己管理内存池,不容易出现内存管理方面的问题,更容易使用和维护。因此,在满足性能的情况下,推荐使用 UnpooledHeapByteBuf。
首先看一下 UnpooledHeapByteBuf 的成员变量:
- 负责内存分配的 ByteBufAllocator。
- 缓冲区实现 byte 数组。
- 从 ByteBuf 到 NIO 的 ByteBuffer 的转换对象 tmpNioBuf。
private final ByteBufAllocator alloc;
private byte[] array;
private ByteBuffer tmpNioBuf;
在将 AbstractByteBuf 的时候,我们提到 getBytes、capacity 等方法是由子类来实现的,这里我们先看看 getBytes 的实现,从代码中可以看出来,是直接调用 System.arraycopy 进行的数组复制。
@Override
public ByteBuf getBytes(int index, byte[] dst, int dstIndex, int length) {
checkDstIndex(index, length, dstIndex, dst.length);
System.arraycopy(array, index, dst, dstIndex, length);
return this;
}
接下来看一下动态伸缩的 capacity 方法,主要做了以下几件事:
- 参数校验,newCapacity 不能小于 0,大于 maxCapacity。
- 如果 maxCapacity 大于 oldCapacity 表示扩容,直接申请新的 byte 数组,进行内存复制即可。
- 如果 maxCapacity 小于 oldCapacity 就是缩容了,同样申请 byte 数组。不同的是,需要根据读指针 readerIndex 与 newCapacity 的大小来决定是否需要进行内存复制。当 readerIndex 小于 newCapacity 时,需要复制内存,否则不需要。
- 设置合适的读写指针位置。
- 更新缓冲区字节数组引用 array 的值。
@Override
public ByteBuf capacity(int newCapacity) {
ensureAccessible();
if (newCapacity < 0 || newCapacity > maxCapacity()) {
throw new IllegalArgumentException("newCapacity: " + newCapacity);
}
int oldCapacity = array.length;
if (newCapacity > oldCapacity) {
byte[] newArray = new byte[newCapacity];
System.arraycopy(array, 0, newArray, 0, array.length);
setArray(newArray);
} else if (newCapacity < oldCapacity) {
byte[] newArray = new byte[newCapacity];
int readerIndex = readerIndex();
if (readerIndex < newCapacity) {
int writerIndex = writerIndex();
if (writerIndex > newCapacity) {
writerIndex(writerIndex = newCapacity);
}
System.arraycopy(array, readerIndex, newArray, readerIndex, writerIndex - readerIndex);
} else {
setIndex(newCapacity, newCapacity);
}
setArray(newArray);
}
return this;
}
private void setArray(byte[] initialArray) {
array = initialArray;
tmpNioBuf = null;
}
public ByteBuf setIndex(int readerIndex, int writerIndex) {
if (readerIndex < 0 || readerIndex > writerIndex || writerIndex > capacity()) {
throw new IndexOutOfBoundsException(String.format(
"readerIndex: %d, writerIndex: %d (expected: 0 <= readerIndex <= writerIndex <= capacity(%d))",
readerIndex, writerIndex, capacity()));
}
this.readerIndex = readerIndex;
this.writerIndex = writerIndex;
return this;
}
从 ByteBuf 到 ByteBuffer 的转换,主要是使用了 ByteBuffer 的 wrap 方法:
@Override
public ByteBuffer nioBuffer(int index, int length) {
ensureAccessible();
return ByteBuffer.wrap(array, index, length).slice();
}
PooledByteBuf
PooledByteBuf 是 ByteBuf 的内存池实现,应用自己实现的内存池管理策略,一般和操作系统的内存管理策略差不多,往往会更简单些。PooledByteBuf 内存池的分配和释放,主要通过 PoolArena 来实现。比如在 capacity 方法中,最终会使用 arena 的 reallocate 方法来重新分配内存。
public final ByteBuf capacity(int newCapacity) {
ensureAccessible();
// If the request capacity does not require reallocation, just update the length of the memory.
if (chunk.unpooled) {
if (newCapacity == length) {
return this;
}
} else {
if (newCapacity > length) {
if (newCapacity <= maxLength) {
length = newCapacity;
return this;
}
} else if (newCapacity < length) {
if (newCapacity > maxLength >>> 1) {
if (maxLength <= 512) {
if (newCapacity > maxLength - 16) {
length = newCapacity;
setIndex(Math.min(readerIndex(), newCapacity), Math.min(writerIndex(), newCapacity));
return this;
}
} else { // > 512 (i.e. >= 1024)
length = newCapacity;
setIndex(Math.min(readerIndex(), newCapacity), Math.min(writerIndex(), newCapacity));
return this;
}
}
} else {
return this;
}
}
// 最终使用 arena 的 reallocate 方法来重新分配内存。
chunk.arena.reallocate(this, newCapacity, true);
return this;
}
PoolArena 是由多个 PoolChunk 组成的大块内存区域。
abstract class PoolArena<T> {
static final int numTinySubpagePools = 512 >>> 4;
final PooledByteBufAllocator parent;
private final int maxOrder;
final int pageSize;
final int pageShifts;
final int chunkSize;
final int subpageOverflowMask;
final int numSmallSubpagePools;
private final PoolSubpage<T>[] tinySubpagePools;
private final PoolSubpage<T>[] smallSubpagePools;
private final PoolChunkList<T> q050;
private final PoolChunkList<T> q025;
private final PoolChunkList<T> q000;
private final PoolChunkList<T> qInit;
private final PoolChunkList<T> q075;
private final PoolChunkList<T> q100;
}
// PoolChunkList 是 PoolChunk 组成的链表
final class PoolChunkList<T> {
private final PoolArena<T> arena;
private final PoolChunkList<T> nextList;
PoolChunkList<T> prevList;
private final int minUsage;
private final int maxUsage;
private PoolChunk<T> head;
}
每个 PoolChunk 由多个 PoolSubpage 组成。
final class PoolChunk<T> {
final PoolArena<T> arena;
final T memory;
final boolean unpooled;
private final byte[] memoryMap;
private final byte[] depthMap;
private final PoolSubpage<T>[] subpages;
/** Used to determine if the requested capacity is equal to or greater than pageSize. */
private final int subpageOverflowMask;
private final int pageSize;
private final int pageShifts;
private final int maxOrder;
private final int chunkSize;
private final int log2ChunkSize;
private final int maxSubpageAllocs;
/** Used to mark memory as unusable */
private final byte unusable;
private int freeBytes; // 当前 chunk 空闲字节数目
PoolChunkList<T> parent; // 父节点
PoolChunk<T> prev; // 链表前一个节点
PoolChunk<T> next; // 链表后一个节点
}
PoolSubpage 负责管理一个 Page 的内存,通过 bitmap 中的每一位来标记每一块儿内存的占用状态。
final class PoolSubpage<T> {
final PoolChunk<T> chunk;
private final int memoryMapIdx; // 当前page在chunk中的id
private final int runOffset; // 当前page在chunk.memory的偏移量
private final int pageSize; // page大小
private final long[] bitmap; // 通过对每一个二进制位的标记来修改一段内存的占用状态
PoolSubpage<T> prev;
PoolSubpage<T> next;
boolean doNotDestroy;
int elemSize;
private int maxNumElems;
private int bitmapLength;
private int nextAvail;
private int numAvail;
}
PooledDirectByteBuf
PooledDirectByteBuf 基于内存池实现,与 UnPooledDirectByteBuf 的唯一区别就是,缓冲区的分配和销毁策略不同。不仅缓冲区所需内存使用内存池分配管理,PooledDirectByteBuf 对象本身,也使用 Recycler 管理。 比如 PooledDirectByteBuf 创建示例调用的是 Recycler 的 get 方法。
final class PooledDirectByteBuf extends PooledByteBuf<ByteBuffer> {
private static final Recycler<PooledDirectByteBuf> RECYCLER = new Recycler<PooledDirectByteBuf>() {
@Override
protected PooledDirectByteBuf newObject(Handle<PooledDirectByteBuf> handle) {
return new PooledDirectByteBuf(handle, 0);
}
};
static PooledDirectByteBuf newInstance(int maxCapacity) {
PooledDirectByteBuf buf = RECYCLER.get();
buf.setRefCnt(1);
buf.maxCapacity(maxCapacity);
return buf;
}
}
Recycler 是一个轻量级的对象池,一个对象池最核心的方法是从池中获取对象和回收对象到池中,分别对应其 get 和 recycle 方法。
public abstract class Recycler<T> {
public final T get() {
Stack<T> stack = threadLocal.get();
DefaultHandle<T> handle = stack.pop();
if (handle == null) {
handle = stack.newHandle();
handle.value = newObject(handle);
}
return (T) handle.value;
}
public final boolean recycle(T o, Handle<T> handle) {
DefaultHandle<T> h = (DefaultHandle<T>) handle;
if (h.stack.parent != this) {
return false;
}
h.recycle(o);
return true;
}
}
PooledDirectByteBuf 中的 copy 方法用于复制一个新的字节缓冲区实例,该方法首先调用 PooledByteBufAllocator 的 directBuffer 来生成新的 ByteBuf,然后复制数据。
@Override
public ByteBuf copy(int index, int length) {
checkIndex(index, length);
ByteBuf copy = alloc().directBuffer(length, maxCapacity());
copy.writeBytes(this, index, length);
return copy;
}
directBuffer 是在抽象类 AbstractByteBufAllocator 中实现的,进行参数校验之后调用 newDirectBuffer 来获取 ByteBuf,该方法由子类来实现。
@Override
public ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
if (initialCapacity == 0 && maxCapacity == 0) {
return emptyBuf;
}
validate(initialCapacity, maxCapacity);
return newDirectBuffer(initialCapacity, maxCapacity);
}
在内存池版本 PooledByteBufAllocator 的实现中,判断如果内存池 directArena 可用,则从中获取,否则自行 new 一个。
@Override
protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
PoolThreadCache cache = threadCache.get();
PoolArena<ByteBuffer> directArena = cache.directArena;
ByteBuf buf;
if (directArena != null) {
buf = directArena.allocate(cache, initialCapacity, maxCapacity);
} else {
if (PlatformDependent.hasUnsafe()) {
buf = new UnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
} else {
buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
}
}
return toLeakAwareBuffer(buf);
}
而在非内存池版本 UnpooledByteBufAllocator 中,则是直接 new 一个。
@Override
protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
ByteBuf buf;
if (PlatformDependent.hasUnsafe()) {
buf = new UnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
} else {
buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
}
return toLeakAwareBuffer(buf);
}
辅助类
- 内存分配相关: ByteBufAllocator 及其子类 UnpooledByteBufAllocator、PooledByteBufAllocator。
- 组合视图:CompositeByteBuf。
- 工具类:ByteBufUtil。