Netty源码分析：AbstractByteBuf
Posted HelloWorld_EE
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Netty源码分析：AbstractByteBuf相关的知识，希望对你有一定的参考价值。
Netty源码分析：AbstractByteBuf
ByteBuf与Java NIO ByteBuffer类似，由于ByteBuffer存在一定的缺陷，具体缺陷如下：
1）ByteBuffer长度固定，一旦分配，则容量不能动态扩展和收缩
2）ByteBuffer只有一个标识位置的指针，读写的时候需要手动的调用flip()方法来进行从写到读模式的切换，否则读出来的内容就是错误的。
而Netty实现的ByteBuf类似ByteBuffer，但其克服了以上两个缺点，其支持动态扩容和收缩，利用两个标识位置的指针来协助读写操作。
先看一个简单的例子
public class TestByteBuf
public static void main(String[] args)
//1 分配一个ByteBuf，然后写入数据到此Buf
ByteBuf byteBuf = Unpooled.buffer(256);
String content = "wojiushimogui";
byteBuf.writeBytes(content.getBytes());
System.out.println("1、readerIndex:"+byteBuf.readerIndex());
System.out.println("1、writerIndex:"+byteBuf.writerIndex());
//2 开始读数据
int len = 5;
byte[] readContent = new byte[len];
byteBuf.readBytes(readContent);
System.out.println("2、readContent:"+new String(readContent));
System.out.println("2、readerIndex:"+byteBuf.readerIndex());
System.out.println("2、writerIndex:"+byteBuf.writerIndex());
//3 重用缓存区
byteBuf.discardReadBytes();
System.out.println("3、readerIndex:"+byteBuf.readerIndex());
System.out.println("3、writerIndex:"+byteBuf.writerIndex());
/*输出
1、readerIndex:0
1、writerIndex:13
2、readContent:wojiu
2、readerIndex:5
2、writerIndex:13
3、readerIndex:0
3、writerIndex:8
*
* */上面就是关于ByteBuf的一个简单例子，其中readerIndex是读索引，二writerIndex为写索引，ByteBuf就是利用这两个索引来进行缓存的读写的。
ByteBuf的主要类继承关系
ByteBuf的主要类继承关系如下：
从内存分配的角度看，ByteBuf可以分为两类：
（1）堆内存字节缓存区
优点：内存的分配和回收速度快，可以被JVM自动回收。
缺点：如果是进行Socket的I／O读写，则需要额外做一次内存复制，即将堆内存对应的缓冲区复制到内核Channel中，因此性能会有一定程度的下降。
（2）直接内存字节缓存区
其优点和缺点刚好与“堆内存字节缓存区”相反，即优点：如果进行Socket的I／O读写，则不需要进行复制，而由于其是在堆外内存分配，因此相比堆内存其分配和回收就慢一些。
从内存回收的角度看，ByteBuf可以分为两类：
（1）基于对象池的ByteBuf
优点：可以重用ByteBuf对象。即可以提升内存的使用效率。
缺点：内存池的管理和维护添加了复杂性。
（2）普通ByteBuf
其优缺点与ByteBuf相反。
以上两种类型进行两两组合就构成了多种多样的ByteBuf，供用户选择使用。
本篇博文主要介绍AbstractByteBuf这个抽象类，这个类是后面其他子类的基础。
AbstractByteBuf
ByteBuf这个抽象类定义了许多方法，是标准。首先先看第一个子类：AbstractByteBuf，这个类定义了一些公共属性（如下），这些公共属性包括：读索引、写索引、mark以及最大容量等。
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;注意这里的读、写索引是netty实现的ByteBuf对Java NIO ByteBuffer的position改善，即使用读、写两个标识位置的指针来支持缓存的读写，而不像Java NIO ByteBuffer这个类只有position这个标识，我们经常要通过调用flip()函数来进行写模式到读模式的切换否则就读出来的数据是错误的。具体关于ByteBuffer的解析可以看这篇博文[Java源码分析》：Java NIO 之 Buffer]
(http://blog.csdn.net/u010412719/article/details/52775637).
在AbstractByteBuf中并没有定义ByteBuf的缓冲区实现，这是因为AbstractByteBuf并不清楚子类到底基于何种实现，可能是基于byte数组或者是DirectBytebBuffer。
AbstractByteBuf中定义了读写操作方法，在本博文的开始的demo中，就有用到，例如：byteBuf.writeBytes(content.getBytes());byteBuf.readBytes(readContent);下面进行介绍。
1） writeBytes(byte[] src)
下面就以这个函数为例来介绍writeBytes方法。
@Override
public ByteBuf writeBytes(byte[] src)
writeBytes(src, 0, src.length);
return this;
这个方法调用了如下的重载方法，这个名为writeByte还有很多重载方法，这里不进行介绍。
@Override
public ByteBuf writeBytes(byte[] src, int srcIndex, int length)
ensureAccessible();
ensureWritable(length);
setBytes(writerIndex, src, srcIndex, length);
writerIndex += length;
return this;
首先通过调用ensureAccessible()方法来确保该ByteBuf实例是存活的，是可以访问的。
protected final void ensureAccessible()
//根据refCnt()的源码注释得到：refCnt方法返回的是ByteBuf对象的引用计数，如果返回值为零，则说明该对象被摧毁。
if (refCnt() == 0)
throw new IllegalReferenceCountException(0);
然后通过调用ensureWritable方法来确保可写，该函数的源码如下：
@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 = calculateNewCapacity(writerIndex + minWritableBytes);
// Adjust to the new capacity.
capacity(newCapacity);
return this;
该函数是如何来判断是否可写呢？进行了如下判断：
如果可写容量大于等于minWritableBytes，则说明可写，直接返回。如果小于，则首先判断是否超过了最大可以容量，如果超过则跑异常，如果没有则扩容。其中首先调用如下的calculateNewCapacity函数来计算自动扩容后的容量，入参为满足要求的最小容量。
private int calculateNewCapacity(int minNewCapacity)
final int maxCapacity = this.maxCapacity;
final int threshold = 1048576 * 4; // 4 MiB page
if (minNewCapacity == threshold)
return threshold;
// If over threshold, do not double but just increase by threshold.
if (minNewCapacity > threshold)
int newCapacity = minNewCapacity / threshold * threshold;
if (newCapacity > maxCapacity - threshold)
newCapacity = maxCapacity;
else
newCapacity += threshold;
return newCapacity;
// Not over threshold. Double up to 4 MiB, starting from 64.
int newCapacity = 64;
while (newCapacity < minNewCapacity)
newCapacity <<= 1;
return Math.min(newCapacity, maxCapacity);
在calculateNewCapacity函数中， 有一个阈值：4MB，然后进行了如下的判断来求新的容量：
（1）如果需要的新容量刚好等于此阈值，则直接使用此阈值作为新容量然后返回。如果不等于，则会出现两种情况，第一种是大于，第二种是小于。 这两种对应了两种分配内存的策略。
（2） 如果是大于阈值，则采用的是以阈值threshold为基础歩进的方式来得到新容量，其中与最大容量进行了比较，以防超过最大容量。
（3）如果是小于阈值，则是以64为基础来进行倍增，即64–>128–>256…,直到满足最小容量要求的容量作为新容量。
这两种分配内存的策略结合值得我们学习，很经典，为什么这么说呢？先倍增后步进带来的好处为：当内存比较小的时候倍增操作并不会带来太多的内存浪费，当时在内存增大到一定阈值之后，如果继续倍增则可能会带来额外的内存浪费。
在调用 calculateNewCapacity 计算完目标容量之后，则需要重新创建新的缓冲区，并将原缓冲区的内容复制到新创建的ByteBuf中。这些都是调用capacity(newCapacity)中来完成的。由于不同的子类会对应不同的复制操作，所以AbstractByteBuf类中的该方法是一个抽象方法，留给子类自己来实现。
这里我们看下UnpooledHeapByteBuf子类的该方法的实现：
下面的代码比较简单，就是简单的new的一个新的数组，然后进行了复制。
@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;
到这里，在writeBytes方法中就调用ensureWritable(length)方法确保了ByteBuf可写，继续看writeBytes方法，该方法调用了setBytes(writerIndex, src, srcIndex, length);来完成数据的写入，具体是如何写入的，在AbstractByteBuf类中也没有进行实现，这也是因为由于不同的子类会对应不同的复制操作，所以AbstractByteBuf类中的该方法是一个抽象方法，留给子类自己来实现。
同样这里看下UnpooledHeapByteBuf子类的该方法的实现：
该方法首先首先调用checkSrcIndex方法来检查目标数组的可读空间是否够，也再一次检查是否ByteBuf可写容量是否够，如果不够则跑异常。然后调用System.arraycopy将原数组的数据写入到ByteBuf中。
@Override
public ByteBuf setBytes(int index, byte[] src, int srcIndex, int length)
checkSrcIndex(index, length, srcIndex, src.length);
System.arraycopy(src, srcIndex, array, index, length);
return this;
2）readBytes(byte[] dst)
上面分析了writeBytes(byte[] src)方法，这一节就来分析下从缓存中读取数据的方法：readBytes(byte[] dst)，其源代码如下：
@Override
public ByteBuf readBytes(byte[] dst)
readBytes(dst, 0, dst.length);
return this;
@Override
public ByteBuf readBytes(byte[] dst, int dstIndex, int length)
checkReadableBytes(length);
getBytes(readerIndex, dst, dstIndex, length);
readerIndex += length;
return this;
首先调用 checkReadableBytes方法检查ByteBuf是否还有长度为minimumReadableBytes的内容可读。
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));
校验通过之后，readBytes方法调用getBytes方法来从当前的读索引开始，复制length个字节到目标byte数组中。由于不同的子类会对应不同的复制操作，所以AbstractByteBuf类中的该方法是一个抽象方法，留给子类自己来实现。
这里还是看下UnpooledHeapByteBuf子类的该方法的实现：
该方法首先调用checkDstIndex方法来检查目标数组的存储空间是否够用，也再一次检查了ByteBuf的可读内容是否够。然后调用 System.arraycopy完成从ByteBuf中读取内容到源数组中。
@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;
3） discardReadBytes()
该方法是用来重用缓冲区的。即将0～readerIndex-1之间这个区域重用起来，具体做法就是将[readerIndex，writerIndex）之间的数据拷贝到[0,writerIndex-readerIndex)的位置。具体代码如下：
@Override
public ByteBuf discardReadBytes()
ensureAccessible();
if (readerIndex == 0)
return this;
if (readerIndex != writerIndex)
setBytes(0, this, readerIndex, writerIndex - readerIndex);
writerIndex -= readerIndex;
adjustMarkers(readerIndex);
readerIndex = 0;
else
adjustMarkers(readerIndex);
writerIndex = readerIndex = 0;
return this;
由于是想将[readerIndex，writerIndex）之间的数据拷贝到[0,writerIndex-readerIndex)的位置，就有如下几种情况：
1）首先判断readerIndex是否等于0，如果等于0，则说明没有可以重用的空间，直接返回。否则则说明有可以重用的缓冲区。继续进行如下的判断。
2）如果读索引不等于0且不等于写索引，则说明此缓冲区中既有已经读取过的被丢弃的缓冲区也有还没有读取的可读缓冲区，则调用setBytes方法将[readerIndex，writerIndex）之间的数据拷贝到[0,writerIndex-readerIndex)的位置，然后调整读写索引以及mark。
3）如果读索引不等于0但等于写索引，则说明缓冲区中则没有可读的数据了，则不需要进行数据的拷贝，只需要调整读写索引以及mark即可。
其中，上面所用到的调整mark的函数的代码如下：
protected final void adjustMarkers(int decrement)
int markedReaderIndex = this.markedReaderIndex;
if (markedReaderIndex <= decrement)
this.markedReaderIndex = 0;
int markedWriterIndex = this.markedWriterIndex;
if (markedWriterIndex <= decrement)
this.markedWriterIndex = 0;
else
this.markedWriterIndex = markedWriterIndex - decrement;
else
this.markedReaderIndex = markedReaderIndex - decrement;
markedWriterIndex -= decrement;
adjustMarkers函数中调整markedReaderIndex、markedWriterIndex的逻辑如下：由于markedReaderIndex、markedWriterIndex都不能小于0，因此首先判断markedReaderIndex、markedWriterIndex是否小于需要减少的decrement，如果小于等于，则设置markedReaderIndex、markedWriterIndex为零，如果不小于，则设置其为当前值减去decrement之后的新值。
对于setBytes方法，由于不同的子类会对应不同的复制操作，所以AbstractByteBuf类中的该方法是一个抽象方法，留给子类自己来实现。 下面是UnpooledHeapByteBuf子类的该方法的实现：
@Override
public ByteBuf setBytes(int index, ByteBuf src, int srcIndex, int length)
checkSrcIndex(index, length, srcIndex, src.capacity());
if (src.hasMemoryAddress())
PlatformDependent.copyMemory(src.memoryAddress() + srcIndex, array, index, length);
else if (src.hasArray())
setBytes(index, src.array(), src.arrayOffset() + srcIndex, length);
else
src.getBytes(srcIndex, array, index, length);
return this;
小结
本篇博文就从一个Demo中用到的三个方法（writeBytes、readBytes、discardReadBytes）入手，简单分析了下AbstractByteBuf这个类，后面将会对其他子类进行分析。
参考资料
1、《Netty权威指南》
以上是关于Netty源码分析：AbstractByteBuf的主要内容，如果未能解决你的问题，请参考以下文章