[netty核心类]--缓冲区ByteBuf

小咪咪 2022-07-13 00:08 235阅读 0赞

> 本文主要包括以下内容：  
> 1）ByteBuf的三种类型：heapBuffer(堆缓冲区)、directBuffer(直接缓冲区)以及Composite Buffer(复合缓冲区)。
> 
> 2）ByteBuf的工作原理。
> 
> 3）ByteBuf与JDK中ByteBuffer的区别以及对比
> 
> 4）ByteBuf的引用计数器实现类AbstractReferenceCountedByteBuf分析。
> 
> 5）UnpooledHeapByteBuf 基于堆内存缓冲器的源码分析
> 
> 6）PooledDirectByteBuf 源码分析

# 1. netty中ByteBuf的优势 #

缓冲区是不同的通道之间传递数据的中介，JDK中的ByteBuffer操作复杂，而且没有经过优化，所以在netty中实现了一个更加强大的缓冲区 **ByteBuf** 用于表示字节序列。ByteBuf在netty中是通过Channel传输数据的，新的设计解决了JDK中ByteBuffer中的一些问题。

netty中ByteBuf的缓冲区的优势：  
（1）可以自定义缓冲区的类型；

（2）通过内置的复合缓冲类型实现零拷贝；

（3）不需要调用flip()函数切换读/写模式

（4）读取和写入的索引分开了，不像JDK中使用一个索引

（5）引用计数（referenceCounting的实现原理？）

（6） Pooling池

# 2.netty中的ByteBuf的类型： #

[JDK中的Buffer的类型][JDK_Buffer] 有heapBuffer和directBuffer两种类型，但是在netty中除了heap和direct类型外，还有composite Buffer(复合缓冲区类型)。

### (1)Heap Buffer 堆缓冲区 ###

这是最常用的类型，ByteBuf将数据存储在JVM的堆空间，通过将数据存储在数组中实现的。  
1）堆缓冲的优点是：由于数据存储在JVM的堆中可以快速创建和快速释放，并且提供了数组的直接快速访问的方法。

2）堆缓冲缺点是：每次读写数据都要先将数据拷贝到直接缓冲区再进行传递。

### (2)Direct Buffer 直接缓冲区 ###

Direct Buffer在堆之外直接分配内存，直接缓冲区不会占用堆的容量。

（1）Direct Buffer的优点是：在使用Socket传递数据时性能很好，由于数据直接在内存中，不存在从JVM拷贝数据到直接缓冲区的过程，性能好。

（2）缺点是：因为Direct Buffer是直接在内存中，所以分配内存空间和释放内存比堆缓冲区更复杂和慢。

虽然netty的Direct Buffer有这个缺点，但是netty通过内存池来解决这个问题。直接缓冲池不支持数组访问数据，但可以通过间接的方式访问数据数组：

ByteBuf directBuf = Unpooled.directBuffer(16);   
    if(!directBuf.hasArray()){   
        int len = directBuf.readableBytes();   
        byte[] arr = new byte[len];   
        directBuf.getBytes(0, arr);   
    }

但是上面的操作太过复杂，所以在使用时，建议一般是用heap buffer。  
  
不过对于一些IO通信线程中读写缓冲时建议使用DirectByteBuffer，因为这涉及到大量的IO数据读写。对于后端的业务消息的编解码模块使用HeapByteBuffer。

### (3)Composite Buffer 复合缓冲区 ###

这个是netty特有的缓冲类型。复合缓冲区就类似于一个ByteBuf的组合视图，在这个视图里面我们可以创建不同的ByteBuf(可以是不同类型的)。 这样，复合缓冲区就类似于一个列表，我们可以动态的往里面添加和删除其中的ByteBuf，JDK里面的ByteBuffer就没有这样的功能。

Netty提供了Composite ByteBuf来处理复合缓冲区。例如：一条消息由Header和Body组成，将header和body组装成一条消息发送出去。下图显示了Composite ByteBuf组成header和body：  
![这里写图片描述][SouthEast]

如果使用的是JDK的ByteBuffer就不能简单的实现，只能通过创建数组或则新的ByteBuffer，再将里面的内容复制到新的ByteBuffer中，下面给出了一个CompositeByteBuf的使用示例：

//组合缓冲区
    CompositeByteBuf compBuf = Unpooled.compositeBuffer();   
    //堆缓冲区
    ByteBuf heapBuf = Unpooled.buffer(8);   
    //直接缓冲区
    ByteBuf directBuf = Unpooled.directBuffer(16);   
    //添加ByteBuf到CompositeByteBuf 
    compBuf.addComponents(heapBuf, directBuf);   
    //删除第一个ByteBuf 
    compBuf.removeComponent(0);   
    Iterator<ByteBuf> iter = compBuf.iterator();   
    while(iter.hasNext()){   
        System.out.println(iter.next().toString());   
    }   
    
    //使用数组访问数据 
    if(!compBuf.hasArray()){   
        int len = compBuf.readableBytes();   
        byte[] arr = new byte[len];   
        compBuf.getBytes(0, arr);   
    }

# 3.ByteBuf是如何工作的： #

ByteBuf是一个抽象类，内部全部是抽象的函数接口，AbstractByteBuf这个抽象类基本实现了ByteBuf，下面我们通过分析AbstractByteBuf里面的实现来分析ByteBuf的工作原理。

ByteBuf都是基于字节序列的，类似于一个字节数组。在AbstractByteBuf里面定义了下面5个变量：

//源码
    int readerIndex; //读索引
    int writerIndex; //写索引
    private int markedReaderIndex;//标记读索引
    private int markedWriterIndex;//标记写索引
    private int maxCapacity;//缓冲区的最大容量

ByteBuf 与JDK中的 ByteBuffer 的最大区别之一就是：  
（1）netty的ByteBuf采用了读/写索引分离，一个初始化的ByteBuf的readerIndex和writerIndex都处于0位置。  
  
（2）当读索引和写索引处于同一位置时，如果我们继续读取，就会抛出异常IndexOutOfBoundsException。  
  
（3）对于ByteBuf的任何读写操作都会分别单独的维护读索引和写索引。maxCapacity最大容量默认的限制就是Integer.MAX\_VALUE。

## 1.随机访问索引 ##

ByteBuf提供读/写索引，从0开始的索引，第一个字节索引是0，最后一个字节的索引是capacity-1，下面给出一个示例遍历ByteBuf的字节：

public static void main(String[] args) {
        //创建一个16字节的buffer,这里默认是创建heap buffer
        ByteBuf buf = Unpooled.buffer(16);
        //写数据到buffer
        for(int i=0; i<16; i++){
            buf.writeByte(i+1);
        }
        //读数据
        for(int i=0; i<buf.capacity(); i++){
            System.out.print(buf.getByte(i)+", ");
        }
    }
    
    /***output: 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, */

这里有一点需要注意的是：通过索引访问byte时不会改变真实的读索引和写索引，我们可以通过ByteBuf的readerIndex()或则writerIndex()函数来分别推进读索引和写索引。

下面分别看看writeByte()和readByte()的源码

@Override
    public ByteBuf writeByte(int value) {
        ensureAccessible();//检验是否可以写入
        ensureWritable0(1);
        _setByte(writerIndex++, value);//这里写索引自增了
        return this;
    }
    @Override
    public byte readByte() {
        checkReadableBytes0(1);
        int i = readerIndex;
        byte b = _getByte(i);
        readerIndex = i + 1;//这里读索引自增了
        return b;
    }

## 读写索引的分区 ##

ByteBuf提供了两个索引指针变量来支持读写操作，读操作使用的是readerIndex()，写操作使用的是writerIndex()。这与JDK中的ByteBuffer有很大不同，ByteBuffer只有一个方法来设置索引，而且需要使用flip()方法来切换读写操作，这就很麻烦了。

ByteBuf一定满足的是：0<=readerIndex<=writerIndex<=capacity  
下图显示了一个ByteBuf中可以被划分为三个区域：  
![这里写图片描述][SouthEast 1]

## Discardable bytes 回收字节 ##

对于已经读过的字节，我们需要回收，通过调用ByteBuf.discardReadBytes()来回收已经读取过的字节，discardReadBytes()将回收从索引0到readerIndex之间的字节。调用discardReadBytes()方法之后会变成如下图所示;  
![这里写图片描述][SouthEast 2]

很明显discardReadBytes()函数很可能会导致内存的复制，它需要移动ByteBuf中可读字节到开始位置，所以该操作会导致时间开销。说白了也就是时间换空间。

## 可读可写字节 ##

当我们读取字节的时候，一般要先判断buffer中是否有字节可读，这时候可以调用isReadable()函数来判断：源码如下：

@Override
    public boolean isReadable() {
        return writerIndex > readerIndex;
    }

其实也就是判断 读索引是否小于写索引 来判断是否还可以读取字节。在判断是否可写时也是判断写索引是否小于最大容量来判断。

@Override
    public boolean isWritable() {
        return capacity() > writerIndex;
    }

## 清除缓冲区 ##

清除ByteBuf来说，有两种形式，第一种是clear()函数：源码如下：

@Override
    public ByteBuf clear() {
        readerIndex = writerIndex = 0;
        return this;
    }

很明显这种方式并没有真实的清除缓冲区中的数据，而只是把读/写索引值重新都置为0了，这与discardReadBytes()方法有很大的区别。

## 标记Mark和重置reset ##

从源码可知，每个ByteBuf有两个标注索引，

private int markedReaderIndex;//标记读索引
    private int markedWriterIndex;//标记写索引

可以通过重置方法返回上次标记的索引的位置。

## 衍生的缓冲区 ##

调用duplicate()、slice()、slice(int index, int length)等方法可以创建一个现有缓冲区的视图（现有缓冲区与原有缓冲区是指向相同内存）。衍生的缓冲区有独立的readerIndex和writerIndex和标记索引。如果需要现有的缓冲区的全新副本，可以使用copy()获得。

# 4. 创建ByteBuf的方法 #

前面我们也讲过了，ByteBuf主要有三种类型，heap、direct和composite类型，下面介绍创建这三种Buffer的方法：  
（1）通过ByteBufAllocator这个接口来创建ByteBuf，这个接口可以创建上面的三种Buffer，一般都是通过channel的alloc()接口获取。

（2）通过Unpooled类里面的静态方法，创建Buffer

CompositeByteBuf compBuf = Unpooled.compositeBuffer();           
    ByteBuf heapBuf = Unpooled.buffer(8);            
    ByteBuf directBuf = Unpooled.directBuffer(16);

还有一点就是，ByteBuf里面的数据都是保存在字节数组里面的：

byte[] array;

# 5. ByteBuf与ByteBuffer的对比： #

先来说说ByteBuffer的缺点：  
（1）下面是NIO中ByteBuffer存储字节的字节数组的定义，我们可以知道ByteBuffer的字节数组是被定义成final的，也就是长度固定。一旦分配完成就不能扩容和收缩，灵活性低，而且当待存储的对象字节很大可能出现数组越界，用户使用起来稍不小心就可能出现异常。如果要避免越界，在存储之前就要只要需求字节大小，如果buffer的空间不够就创建一个更大的新的ByteBuffer，再将之前的Buffer中数据复制过去，这样的效率是奇低的。

final byte[] hb;// Non-null only for heap buffers

（2）ByteBuffer只用了一个position指针来标识位置，读写模式切换时需要调用flip()函数和rewind()函数，使用起来需要非常小心，不然很容易出错误。

下面说说对应的ByteBuf的优点：  
（1）ByteBuf是吸取ByteBuffer的缺点之后重新设计，存储字节的数组是动态的，最大是Integer.MAX\_VALUE。这里的动态性存在write操作中，write时得知buffer不够时，会自动扩容。

（2） ByteBuf的读写索引分离，使用起来十分方便。此外ByteBuf还新增了很多方便实用的功能。

# 6. ByteBuf的引用计数类AbstractReferenceCountedByteBuf分析 #

看类名我们就可以知道，该类主要是对引用进行计数，有点类似于JVM中判断对象是否可回收的引用计数算法。这个类主要是根据ByteBuf的引用次数判断ByteBuf是否可被自动回收。下面来看看源码：

成员变量

private static final AtomicIntegerFieldUpdater<AbstractReferenceCountedByteBuf> refCntUpdater;
    
    //静态代码段初始化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;

首先我们能看到refCntUpdater这个变量，这是一个原子变量类AtomicIntegerFieldUpdater，她是一个静态变量，而且是在static代码段里面实例化的，这说明这个类是单例的。这个类的主要作用是以原子的方式对成员变量进行更新操作以实现线程安全（这里线程安全的保证也就是CAS+volatile）。

然后是定义了refCnt变量，用于跟踪对象的引用次数，使用volatile修饰解决原子变量可视性问题。

对象引用计数器  
那么，对对象的引用计数与释放是怎么实现的呢？核心就是两个函数：

//计数加1
    retain();
    
    //计数减一
    release();

下面分析这两个函数源码：  
每调用一次retain()函数一次，引用计数器就会加一，由于可能存在多线程并发使用的情景，所以必须保证累加操作是线程安全的，那么是怎么保证的呢？我们来看一下源码：

public ByteBuf retain() {
        return retain0(1);
    }
    
    public ByteBuf retain(int increment) {
        return retain0(checkPositive(increment, "increment"));
    }
    
    /** 最后都是调用这个函数。 */
    private ByteBuf retain0(int increment) {
        for (;;) {
            int refCnt = this.refCnt;
            final int nextCnt = refCnt + increment;
    
            // Ensure we not resurrect (which means the refCnt was 0) and also that we encountered an overflow.
            if (nextCnt <= increment) {
                throw new IllegalReferenceCountException(refCnt, increment);
            }
            if (refCntUpdater.compareAndSet(this, refCnt, nextCnt)) {
                break;
            }
        }
        return this;
    }

在retain0()函数中， 通过for(;;)来实现了自旋锁。通过自旋来对引用计数器refCnt执行加1操作。这里的加一操作是通过原子变量refCntUpdater的compareAndSet(this, refCnt, nextCnt)方法实现的，这个通过硬件级别的CAS保证了原子性，如果修改失败了就会不停的自旋，直到修改成功为止。

下面再看看释放的过程：release()函数：

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;
            }
        }
    }

这里基本和retain()函数一样，也是通过自旋和CAS保证执行的正确的将计数器减一。这里需要注意的是当`refCnt == decrement` 也就是引用对象不可达时，就需要调用deallocate();方法来释放ByteBuf对象。

# 7. UnpooledHeapByteBuf源码分析 #

从类名就可以知道UnpooledHeapByteBuf 是基于堆内存的字节缓冲区，没有基于对象池实现，这意味着每次的IO读写都会创建一个UnpooledHeapByteBuf对象，会造成一定的性能影响，但是也不容易出现内存管理的问题。

![这里写图片描述][SouthEast 3]  
成员变量  
有三个成员变量，各自的含义见注释。

//缓冲区分配器，用于UnpooledHeapByteBuf的内存分配。在UnpooledHeapByteBuf构造器中实例化
    private final ByteBufAllocator alloc;
    //字节数组作为缓冲区
    byte[] array;
    //实现ByteBuf与NIO中ByteBuffer的转换
    private ByteBuffer tmpNioBuf;

动态扩展缓冲区  
在说道AbstractByteBuf的时候，ByteBuf是可以自动扩展缓冲区大小的，这里我们分析一下在UnpooledHeapByteBuf中是怎么实现的。

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;
    }

里面的实现并不复杂：  
（1）首先获取原本的容量oldCapacity；

（2）如果新需求容量大于oldCapacity，以新的容量newCapacity创建字节数组，将原来的字节数组内容通过调用System.arraycopy(array, 0, newArray, 0, array.length);复制过去，并将新的字节数组设为ByteBuf的字节数组。

（3）如果新需求容量小于oldCapacity就不需要动态扩展，但是需要截取出一段新缓冲区。

# 8. PooledDirectByteBuf 内存池原理分析 #

PooledDirectByteBuf基于内存池实现的，具体的内存池的实现原理，比较复杂，我没分析清楚，具体的只知道，内存池就是一片提前申请的内存，当需要ByteBuf的时候，就从内存池中申请一片内存，这样效率比较高。

PooledDirectByteBuf和UnPooledDirectByteBuf基本一样，唯一不同的就是内存分配策略。

创建字节缓冲区实例  
  
由于PooledDirectByteBuf基于内存池实现的，所以不能通过new关键字直接实例化一个对象，而是直接从内存池中获取，然后设置引用计数器的值。看下源码：

static PooledDirectByteBuf newInstance(int maxCapacity) {
        PooledDirectByteBuf buf = RECYCLER.get();
        buf.reuse(maxCapacity);
        return buf;
    }

通过RECYCLER对象的get()函数从内存池获取PooledDirectByteBuf对象。然后在`buf.reuse(maxCapacity);` 函数里面设置引用计数器为1。

[JDK_Buffer]: http://blog.csdn.net/u010853261/article/details/53464322
[SouthEast]: /images/20220713/5974297fdef247968b7a0826499c3cfb.png
[SouthEast 1]: /images/20220713/ba4fb9135f304748897fa7b8cca6f3db.png
[SouthEast 2]: /images/20220713/e1942a46e5c84fafa7e11a2b0f6825ce.png
[SouthEast 3]: /images/20220713/eeaefecac1fa4371a2ec060171bd0e3b.png