Netty ByteBuf与NIO ByteBuffer

╰半橙微兮° 2022-11-27 03:17 154阅读 0赞

Netty中引入了ByteBuf，相较于NIO中的ByteBuffer有了一下改进，使得网络编程更加简便。

### ByteBuffer的缺点 ###

在NIO的Buffer中，我们已经知道Buffer有七种，最常用的是ByteBuffer，而ByteBuffer也有一些局限性：

1.  ByteBuffer的长度是固定的，一旦分配完成就不能进行扩容和收缩，当需要操作的对象大于Buffer的容量时，会发生异常；
2.  在进行读写状态切换时，需要调用flip()或rewind()方法改变指针position的位置，稍有不慎，就不能完成任务；

基于ByteBuffer的这些局限性，Netty的ByteBuf改进了这些问题。

### ByteBuf的工作原理 ###

由于ByteBuf是改进了ByteBuffer的缺点， 这一部分内容主要从两个方面介绍：

### **扩容** ###

ByteBuffer

在对ByteBuffer进行put()操作时，若缓冲区可写入空间不足，会发生BufferOverFlowException异常，因此，ByteBuffer中每次在进行put()操作时，都会对可用空间进行校验，空间不足时，会创建一个新的ByteBuffer，将旧ByteBuffer复制到新的中去，最后释放旧的ByteBuffer，代码如下：

if(
         
          this.buffer.remaining() < needSize) {
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     
         
          int toBeExtSize = needSize < 
         
          128 ? needSize : 
         
          128;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     ByteBuffer tmpBuffer = ByteBuffer.allocate(
         
          this.buffer.capacity() + toBeExtSize);
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     
         
          this.buffer.flip();
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     tmpBuffer.put(
         
          this.buffer);
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     
         
          this.buffer = tmpBuffer;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 }

ByteBuf

问题在于，每次进行put()操作都要进行可写空间校验，会导致代码冗余，不小心操作就可能出现别的问题。基于此，ByteBuf对ByteBuffer进行了封装，利用write操作进行可用空间的校验，当可用空间不足时，ByteBuf会进行自动扩容，对于使用者而言，不需要关心底层实现，操作更简便。

AbstractByteBuf
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
             
         
          @Override
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
             
         
          public ByteBuf writeByte(int value) {
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 ensureWritable(
         
          1);
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 setByte(writerIndex++, value);
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 
         
          return 
         
          this;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
             }
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
             
         
          @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));
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 }
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                
         
          //扩容
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 
         
          int newCapacity = calculateNewCapacity(writerIndex + minWritableBytes);
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 capacity(newCapacity);
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 
         
          return 
         
          this;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
             }

可以看到具体的容量计算在calcculateNewCapacity()方法中：

private int calculateNewCapacity(int minNewCapacity) {
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
             
         
          int maxCapacity = 
         
          this.maxCapacity;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
             
         
          int threshold = 
         
          4194304; 
         
          //4mb阀值
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
             
         
          if(minNewCapacity == 
         
          4194304) {
         
          //如果新容量为阀值，直接返回
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 
         
          return 
         
          4194304;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
             } 
         
          else {
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 
         
          int newCapacity;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 
         
          if(minNewCapacity > 
         
          4194304) {
         
          //如果传入的新容量大于阀值，进行计算
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     newCapacity = minNewCapacity / 
         
          4194304 * 
         
          4194304;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     
         
          if(newCapacity > maxCapacity - 
         
          4194304) {
         
          //如果大于最大容量，新容量为最大容量
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                         newCapacity = maxCapacity;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     } 
         
          else {
         
          //否则新容量 + 阀值 4mb，按照阀值扩容
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                         newCapacity += 
         
          4194304;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     }
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     
         
          return newCapacity;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 } 
         
          else {
         
          //如果小于阀值，则以64为计数倍增，知道倍增的结果>=需要的容量值
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     
         
          for(newCapacity = 
         
          64; newCapacity < minNewCapacity; newCapacity <<= 
         
          1) {
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                         ;
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     }
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                     
         
          return Math.min(newCapacity, maxCapacity);
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
                 }
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
             }
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
         }

注意：

1. 当申请的新空间大于阀值时，采用每次步进4MB的方式进行扩张内存，而不是倍增，因为这会造成内存膨胀和浪费

2. 而但申请的新空间小于阀值时，则以64为基数进行倍增而不是步进，因为当内存比较小的时候，倍增是可以接受的（64 -> 128 和 10Mb -> 20Mb相比）

### **位置指针** ###

ByteBuffer

ByteBuffer只有一个位置指针，当我们需要将读写状态切换时，需要手动调用flip()方法或rewind()方法改变position的值，使用不慎就会造成问题。

//创建缓冲区
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
         ByteBuffer buffer = ByteBuffer.allocate(
         
          1024);
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
         
          //存入数据
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
         buffer.put((
         
          byte)
         
          'A');
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
         buffer.put((
         
          byte)
         
          'B');
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
         
          //翻转
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
         buffer.flip();
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
         
          //读取二次数据
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
         System.out.println((
         
          char)buffer.get());
        
         
       
        
       
       
        
        
         
       
        
       
        
        
         
          
         System.out.println((
         
          char)buffer.get());

ByteBuf

ByteBuf通过两个位置指针来协助缓冲区的读写操作，读操作使用readerIndex，写操作使用writerIndex。

readerIndex和writerIndex的取值一开始都是0，随着数据的写入writerIndex会增加，读取数据会使readerIndex增加，但是它不会超过writerIndex。在读取之后，0～readerIndex的就被视为discard的，调用discardReadBytes方法，可以释放这部分空间，它的作用类似ByteBuffer的compact方法。readerIndex和writerIndex之间的数据是可读取的，等价于ByteBuffer position和limit之间的数据。writerIndex和capacity之间的空间是可写的，等价于ByteBuffer limit和capacity之间的可用空间。

由于写操作不修改readerIndex指针，读操作不修改writerIndex指针，因此读写之间不再需要调整位置指针，这极大地简化了缓冲区的读写操作，避免了由于遗漏或者不熟悉flip()操作导致的功能异常。

初始分配的ByteBuf如图:

![20190320214459136.png][]

写入N个字节之后的ByteBuf如图:

![20190320214459156.png][]

读取M（＜N）个字节之后的ByteBuf如图:

![20190320214459175.png][]

调用discardReadBytes操作之后的ByteBuf如图:

![20190320214459179.png][]

调用clear操作之后的ByteBuf如图:

![20190320214459183.png][]

## **ByteBuf源码思路** ##

由于NIO的Channel读写的参数都是ByteBuffer，因此，Netty的ByteBuf接口必须提供API方便的将ByteBuf转换成ByteBuffer，或者将ByteBuffer包装成ByteBuf。考虑到性能，应该尽量避免缓冲区的复制，内部实现的时候可以考虑聚合一个ByteBuffer的私有指针用来代表ByteBuffer。

从内存分配的角度看，ByteBuf可以分为两类。

（1）**堆内存**（HeapByteBuf）字节缓冲区：特点是内存的分配和回收速度快，可以被JVM自动回收；缺点就是如果进行Socket的I/O读写，需要额外做一次内存复制，将堆内存对应的缓冲区复制到内核Channel中，性能会有一定程度的下降。

（2）**直接内存**（DirectByteBuf）字节缓冲区：非堆内存，它在堆外进行内存分配，相比于堆内存，它的分配和回收速度会慢一些，但是将它写入或者从Socket Channel中读取时，由于少了一次内存复制，速度比堆内存快。

正是因为各有利弊，所以Netty提供了多种ByteBuf供开发者使用，经验表明，ByteBuf的最佳实践是在I/O通信线程的读写缓冲区使用DirectByteBuf，后端业务消息的编解码模块使用HeapByteBuf，这样组合可以达到性能最优。

从内存回收角度看，ByteBuf也分为两类：基于对象池的ByteBuf和普通ByteBuf。

两者的主要区别就是基于对象池的ByteBuf可以重用ByteBuf对象，它自己维护了一个内存池，可以循环利用创建的ByteBuf，提升内存的使用效率，降低由于高负载导致的频繁GC。测试表明使用内存池后的Netty在高负载、大并发的冲击下内存和GC更加平稳。尽管推荐使用基于内存池的ByteBuf，但是内存池的管理和维护更加复杂，使用起来也需要更加谨慎，因此，Netty提供了灵活的策略供使用者来做选择。

### **内存池原理分析**  ###

Arena本身是指一块区域，在内存管理中，Memory Arena是指内存中的一大块连续的区域，PoolArena就是Netty的内存池实现类。

为了集中管理内存的分配和释放，同时提高分配和释放内存时候的性能，很多框架和应用都会通过预先申请一大块内存，然后通过提供相应的分配和释放接口来使用内存。这样一来，对内存的管理就被集中到几个类或者函数中，由于不再频繁使用系统调用来申请和释放内存，应用或者系统的性能也会大大提高。在这种设计思路下，预先申请的那一大块内存就被称为Memory Arena。

不同的框架，Memory Arena的实现不同，Netty的PoolArena是由多个Chunk组成的大块内存区域，而每个Chunk则由一个或者多个Page组成，因此，对内存的组织和管理也就主要集中在如何管理和组织Chunk和Page了。

**PoolChunk**

Chunk主要用来组织和管理多个Page的内存分配和释放，在Netty中，Chunk中的Page被构建成一棵二叉树。假设一个Chunk由16个Page组成，那么这些Page将会被按照下图所示的形式组织起来。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2R1YW5kdWFuX2w_size_16_color_FFFFFF_t_70][]

Page的大小是4个字节，Chunk的大小是64个字节(4×16)。整棵树有5层，第1层（也就是叶子节点所在的层）用来分配所有Page的内存，第4层用来分配2个Page的内存，依次类推。

每个节点都记录了自己在整个Memory Arena中的偏移地址，当一个节点代表的内存区域被分配出去之后，这个节点就会被标记为已分配，自这个节点以下的所有节点在后面的内存分配请求中都会被忽略。举例来说，当我们请求一个16字节的存储区域时，上面这个树中的第3层中的4个节点中的一个就会被标记为已分配，这就表示整个Memroy Arena中有16个字节被分配出去了，新的分配请求只能从剩下的3个节点及其子树中寻找合适的节点。

对树的遍历采用深度优先的算法，但是在选择哪个子节点继续遍历时则是随机的，并不像通常的深度优先算法中那样总是访问左边的子节点。

**PoolSubpage**

对于小于一个Page的内存，Netty在Page中完成分配。每个Page会被切分成大小相等的多个存储块，存储块的大小由第一次申请的内存块大小决定。假如一个Page是8个字节，如果第一次申请的块大小是4个字节，那么这个Page就包含2个存储块；如果第一次申请的是8个字节，那么这个Page就被分成1个存储块。

一个Page只能用于分配与第一次申请时大小相同的内存，比如，一个4字节的Page，如果第一次分配了1字节的内存，那么后面这个Page只能继续分配1字节的内存，如果有一个申请2字节内存的请求，就需要在一个新的Page中进行分配。

Page中存储区域的使用状态通过一个long数组来维护，数组中每个long的每一位表示一个块存储区域的占用情况：0表示未占用，1表示以占用。对于一个4字节的Page来说，如果这个Page用来分配1个字节的存储区域，那么long数组中就只有一个long类型的元素，这个数值的低4位用来指示各个存储区域的占用情况。对于一个128字节的Page来说，如果这个Page也是用来分配1个字节的存储区域，那么long数组中就会包含2个元素，总共128位，每一位代表一个区域的占用情况。

**内存回收策略**

无论是Chunk还是Page，都通过状态位来标识内存是否可用，不同之处是Chunk通过在二叉树上对节点进行标识实现，Page是通过维护块的使用状态标识来实现。

对于使用者来说，不需要关心内存池的实现细节，也不需要与这些类库打交道，只需要按照API说明正常使用即可。

## **辅助类功能介绍** ##

### **ByteBufHolder**  ###

ByteBufHolder是ByteBuf的容器，在Netty中，它非常有用，例如HTTP协议的请求消息和应答消息都可以携带消息体，这个消息体在NIO ByteBuffer中就是个ByteBuffer对象，在Netty中就是ByteBuf对象。由于不同的协议消息体可以包含不同的协议字段和功能，因此，需要对ByteBuf进行包装和抽象，不同的子类可以有不同的实现。为了满足这些定制化的需求，Netty抽象出了ByteBufHolder对象，它包含了一个ByteBuf，另外还提供了一些其他实用的方法，使用者继承ByteBufHolder接口后可以按需封装自己的实现。

### **ByteBufAllocator**  ###

ByteBufAllocator是字节缓冲区分配器，按照Netty的缓冲区实现不同，共有两种不同的分配器：基于内存池的字节缓冲区分配器和普通的字节缓冲区分配器。

### **CompositeByteBuf** ###

CompositeByteBuf允许将多个ByteBuf的实例组装到一起，形成一个统一的视图，有点类似于数据库将多个表的字段组装到一起统一用视图展示。

CompositeByteBuf在一些场景下非常有用，例如某个协议POJO对象包含两部分：消息头和消息体，它们都是ByteBuf对象。当需要对消息进行编码的时候需要进行整合，如果使用JDK的默认能力，有以下两种方式：

（1）将某个ByteBuffer复制到另一个ByteBuffer中，或者创建一个新的ByteBuffer，将两者复制到新建的ByteBuffer中；

（2）通过List或数组等容器，将消息头和消息体放到容器中进行统一维护和处理。

上面的做法非常别扭，实际上我们遇到的问题跟数据库中视图解决的问题一致——缓冲区有多个，但是需要统一展示和处理，必须有存放它们的统一容器。为了解决这个问题，Netty提供了CompositeByteBuf。

它定义了一个Component类型的集合，实际上Component就是ByteBuf的包装实现类，它聚合了ByteBuf对象，维护了在集合中的位置偏移量信息等。

关于使用复合缓冲区实现零拷贝会在下篇博文介绍。

### **ByteBufUtil**  ###

ByteBufUtil是一个非常有用的工具类，它提供了一系列静态方法用于操作ByteBuf对象。其中最有用的方法就是对字符串的编码和解码，具体如下。

（1）encodeString(ByteBufAllocator alloc, CharBuffer src, Charset charset)：对需要编码的字符串src按照指定的字符集charset进行编码，利用指定的ByteBufAllocator生成一个新的ByteBuf；

（2）decodeString(ByteBuffer src, Charset charset)：使用指定的ByteBuffer和charset进行对ByteBuffer进行解码，获取解码后的字符串。

还有一个非常有用的方法就是hexDump，它能够将参数ByteBuf的内容以十六进制字符串的方式打印出来，用于输出日志或者打印码流，方便问题定位，提升系统的可维护性。hexDump包含了一系列的方法，参数不同，输出的结果也不同。

### Unpooled （非池化）缓存 ###

当未引用 ByteBufAllocator 时，上面的方法无法访问到 ByteBuf。对于这个用例 Netty 提供一个实用工具类称为 Unpooled,，它提供了静态辅助方法来创建非池化的 ByteBuf 实例。表列出了最重要的方法

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2R1YW5kdWFuX2w_size_16_color_FFFFFF_t_70 1][]

在 非联网项目，该 Unpooled 类也使得它更容易使用的 ByteBuf API，获得一个高性能的可扩展缓冲 API，而不需要 Netty 的其他部分的。

[20190320214459136.png]: /images/20221124/4e640ea9bfb54752b3f6214fc56fc6c8.png
[20190320214459156.png]: /images/20221124/56edb43cc7ab4b889b32dd6d64fe5165.png
[20190320214459175.png]: /images/20221124/1ad84be2a8104720b673ca5883abc0b2.png
[20190320214459179.png]: /images/20221124/52acdf972dc84ebea31c50c8e2e9bcda.png
[20190320214459183.png]: /images/20221124/8a817c3f102e47eeadac566d1e9a1bdf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2R1YW5kdWFuX2w_size_16_color_FFFFFF_t_70]: /images/20221124/47fd0c106981454690caf61d0a53ee4c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2R1YW5kdWFuX2w_size_16_color_FFFFFF_t_70 1]: /images/20221124/f3dfb25c788842a3a4c1079579bcb3c3.png