Netty ByteBuf入门

Posted 2023-03-16

参考技术A ByteBuf由一段地址空间，一个read index和一个write index组成。两个index分别记录读写进度，省去了NIO中ByteBuffer手动调用flip和clear的烦恼。

通过上图可以很好的理解ByteBuf的数据划分。writer index到capacity之间的部分是空闲区域，可以写入数据；reader index到writer index之间是已经写过还未读取的可读数据；0到reader index是已读过可以释放的区域。

三个index之间的关系是：reader index <= writer index <= capacity

ByteBuf根据其数据存储空间不同有可以分为三种：基于JVM堆内的，基于直接内存的和组合的。

堆内受JVM垃圾收集器的管辖，使用上相对安全一些，不用每次手动释放。弊端是GC是会影响性能的；还有就是内存的拷贝带来的性能损耗(JVM进程到Socket)。

直接内存则不受JVM的管辖，省去了向JVM拷贝数据的麻烦。但是坏处就是别忘了释放内存，否则就会发生内存泄露。相比于堆内存，直接内存的的分配速度也比较慢。

最佳实践：在IO通信的线程中的读写Buffer使用DirectBuffer(省去内存拷贝的成本)，在后端业务消息的处理使用HeapBuffer(不用担心内存泄露)。

通过hasArray检查一个ByteBuf heap based还是direct buffer。

ByteBuf提供了两个工具类来创建ByteBuf，分别是支持池化的Pooled和普通的Unpooled。Pooled缓存了ByteBuf的实例，提高性能并且减少内存碎片。它使用Jemalloc来高效的分配内存。

如果在Channel中我们可以通过channel.alloc()来拿到ByteBufAllocator，具体它使用Pool还是Unpool，Directed还是Heap取决于程序的配置。

markReaderIndex和resetReaderIndex是一个成对的操作。markReaderIndex可以打一个标记，调用resetReaderIndex可以把readerIndex重置到原来打标记的位置。

discardReadByte可以把读过的空间释放，这时buffer的readerIndex置为0，可写空间和writerIndex也会相应的改变。discardReadBytes在内存紧张的时候使用用，但是调用该方法会伴随buffer的内存整理的。这是一个expensive的操作。

clear是把readerIndex和writerIndex重置到0。但是，它不会进行内存整理，新写入的内容会覆盖掉原有的内容。

派生操作会产生一个新的ByteBuf实例。这里的新指得是ByteBuf的引用是新的所有的index也是新的。但是它们共用着一套底层存储。派生函数：

如果想要复制一个 全新 的ByteBuffer请使用copy，这会完全的复制一个新的ByteBuf出来。

引用计数记录了当前ByteBuf被引用的次数。新建一个ByteBuf它的refCnt是1，当refCnt == 0时，这个ByteBuf即可被回收。

引用技术主要用于内存泄露的判断，Netty提供了内存泄露检测工具。通过使用参数 -Dio.netty.leakDetectionLevel=$level 可以配置检测级别：

很多时候需要从ByteBuf中查找特定的字符，比如LineBasedFrameDecoder需要在ByteBuf中查找'\r\n'。ByteBuf提供了简单的indexOf这样的函数。同时也可以使用ByteProcesser来查找。

以下 gist 提供了一些example。

Netty入门——ByteBuf

目录

• • 一、ByteBuf的概述
  • 二、ByteBuf的创建
  • • 2.1、创建一个带有初始容量的ByteBuf代码示例
    • 2.2、查看ByteBuf最大容量的代码示例
    • 2.3、查看ByteBuf最大容量及扩容后最大容量的代码示例
  • 三、ByteBuf (直接内存 vs 堆内存)
  • • 3.1、创建池化基于堆的ByteBuf 代码示例
    • 3.2、创建池化基于直接内存的ByteBuf 代码示例
  • 四、ByteBuf (池化 vs 非池化)
  • • 4.1、默认采用的是池化基于直接内存的 代码示例
    • 4.1、设置vm参数采用非池化基于直接内存的 代码示例
  • 五、ByteBuf 组成
  • 六、ByteBuf 写入
  • • 6.1、ByteBuf 写入方法
    • 6.2、ByteBuf 写入的 代码示例
  • 七、ByteBuf 扩容
  • • 7.1、扩容规则
    • 7.2、扩容代码示例
  • 八、 ByteBuf 读取
  • • 8.1、每次读一个字节的代码示例
    • 8.2、重复读取字节的代码示例
  • 九、 ByteBuf 内存释放（retain & release）
  • • 9.1、ByteBuf 内存释放的概述
  • 十、ByteBuf 零拷贝（slice 切片）
  • • 10.1、零拷贝（slice）的概述
    • 10.2、无参数 slice 的示例代码
    • 10.3、有参数 slice 的示例代码
    • 10.4、slice 切片后的 max capacity 被固定为这个区间的大小的示例代码
    • 10.5、slice 切片后slice 的内容发生了更改，原始 ByteBuf 也会受影响的示例代码
  • 十一、ByteBuf 零拷贝（duplicate）
  • • 11.1、零拷贝（duplicate）的概述
  • 十二、ByteBuf 零拷贝（CompositeByteBuf 合并）
  • • 12.1、多个 ByteBuf 合并为一个逻辑上的 ByteBuf ，进行了数据的内存复制的示例代码
    • 12.2、多个 ByteBuf 合并为一个逻辑上的 ByteBuf ，使用compositeBuffer的方式，避免拷贝的示例代码
  • 十三、ByteBuf 零拷贝（Unpooled 创建、组合、复制）
  • • 13.1、ByteBuf 零拷贝（Unpooled）的概述
    • 13.2、多个 ByteBuf 合并为一个逻辑上的 ByteBuf ,使用Unpooled,包装ByteBuf的示例代码
    • 13.3、多个 ByteBuf 合并为一个逻辑上的 ByteBuf ,使用Unpooled,包装普通字节数组的示例代码

一、ByteBuf的概述

• ByteBuf是对字节数据的封装。

二、ByteBuf的创建

2.1、创建一个带有初始容量的ByteBuf代码示例

• 代码示例

  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.ByteBufAllocator;
  import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
  import static io.netty.util.internal.StringUtil.NEWLINE;
  /**
   * @description: ByteBuf(创建)
   * @author: xz
   */
  public class ByteBufTest 
      public static void main(String[] args) 
          //createByteBuf1();
      
  
      /**
       *  创建了一个初始容量是 10 的ByteBuf
       * */
      public static void createByteBuf1()
          //创建了一个默认的 ByteBuf（池化基于直接内存的 ByteBuf），初始容量是 10
          ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(10);
          System.out.println("查看设置的ByteBuf初始容量:"+buf);
      
      /**
       * 输出ByteBuf的工具类
       * */
      public static void log(ByteBuf buffer) 
          int length = buffer.readableBytes();
          int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
          StringBuilder buf = new StringBuilder(rows * 80 * 2)
                  .append("read index:").append(buffer.readerIndex())
                  .append(" write index:").append(buffer.writerIndex())
                  .append(" capacity:").append(buffer.capacity())
                  .append(NEWLINE);
          appendPrettyHexDump(buf, buffer);
          System.out.println(buf.toString());
      
  
  

• 输出结果
  

2.2、查看ByteBuf最大容量的代码示例

• 代码示例

  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.ByteBufAllocator;
  import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
  import static io.netty.util.internal.StringUtil.NEWLINE;
  /**
   * @description: ByteBuf(创建)
   * @author: xz
   */
  public class ByteBufTest 
      public static void main(String[] args) 
          createByteBuf2();
      
       /**
       *  查看ByteBuf最大容量
       * */
      public static void createByteBuf2()
          //创建了一个默认的 ByteBuf（池化基于直接内存的 ByteBuf）
          ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
          //查看ByteBuf最大容量
          System.out.println("ByteBuf最大容量:"+buf.maxCapacity());
      
      /**
       * 输出ByteBuf的工具类
       * */
      public static void log(ByteBuf buffer) 
          int length = buffer.readableBytes();
          int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
          StringBuilder buf = new StringBuilder(rows * 80 * 2)
                  .append("read index:").append(buffer.readerIndex())
                  .append(" write index:").append(buffer.writerIndex())
                  .append(" capacity:").append(buffer.capacity())
                  .append(NEWLINE);
          appendPrettyHexDump(buf, buffer);
          System.out.println(buf.toString());
      
  
  

• 输出结果
  

2.3、查看ByteBuf最大容量及扩容后最大容量的代码示例

• 代码示例

  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.ByteBufAllocator;
  import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
  import static io.netty.util.internal.StringUtil.NEWLINE;
  
  /**
   * @description: ByteBuf(创建)
   * @author: xz
   */
  public class ByteBufTest 
      public static void main(String[] args) 
          createByteBuf3();
      
       /**
       *  查看ByteBuf最大容量及扩容后的最大容量
       * */
      public static void createByteBuf3()
          //1、创建了一个默认的 ByteBuf（池化基于直接内存的 ByteBuf），最大容量256
          ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
          //调用输出ByteBuf的工具类
          log(buf);
  
          //2、创建长度为300的字符串写入ByteBuf，查看扩容后的最大容量
          StringBuilder sb = new StringBuilder();
          for (int i = 0; i < 300; i++) 
              sb.append("a");
          
          buf.writeBytes(sb.toString().getBytes());
          //调用输出ByteBuf的工具类
          log(buf);
      
      /**
       * 输出ByteBuf的工具类
       * */
      public static void log(ByteBuf buffer) 
          int length = buffer.readableBytes();
          int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
          StringBuilder buf = new StringBuilder(rows * 80 * 2)
                  .append("read index:").append(buffer.readerIndex())
                  .append(" write index:").append(buffer.writerIndex())
                  .append(" capacity:").append(buffer.capacity())
                  .append(NEWLINE);
          appendPrettyHexDump(buf, buffer);
          System.out.println(buf.toString());
      
  
  

• 输出结果
  

三、ByteBuf (直接内存 vs 堆内存)

3.1、创建池化基于堆的ByteBuf 代码示例

• 代码示例

  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.ByteBufAllocator;
  import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
  import static io.netty.util.internal.StringUtil.NEWLINE;
  /**
   * @description: ByteBuf (直接内存 vs 堆内存)
   * @author: xz
   */
  public class ByteBufTest2 
      public static void main(String[] args) 
          //创建了一个默认的 ByteBuf（创建池化基于堆的 ByteBuf），初始容量是 10
          ByteBuf buf1 = ByteBufAllocator.DEFAULT.heapBuffer(10);
          //通过输出类信息PooledUnsafeHeapByteBuf可知，采用的是池化基于堆内存
          System.out.println(buf1.getClass());//输出class io.netty.buffer.PooledUnsafeHeapByteBuf
          //输出容量
          log(buf1);
      
  
      /**
       * 输出ByteBuf的工具类
       * */
      public static void log(ByteBuf buffer) 
          int length = buffer.readableBytes();
          int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
          StringBuilder buf = new StringBuilder(rows * 80 * 2)
                  .append("read index:").append(buffer.readerIndex())
                  .append(" write index:").append(buffer.writerIndex())
                  .append(" capacity:").append(buffer.capacity())
                  .append(NEWLINE);
          appendPrettyHexDump(buf, buffer);
          System.out.println(buf.toString());
      
  
  

• 输出结果
  

3.2、创建池化基于直接内存的ByteBuf 代码示例

• 代码示例

  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.ByteBufAllocator;
  import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
  import static io.netty.util.internal.StringUtil.NEWLINE;
  /**
   * @description: ByteBuf (直接内存 vs 堆内存)
   * @author: xz
   */
  public class ByteBufTest2 
      public static void main(String[] args) 
          //创建了一个默认的 ByteBuf（创建池化基于直接内存的 ByteBuf），初始容量是 10
          ByteBuf buf2 = ByteBufAllocator.DEFAULT.directBuffer(10);
          //通过输出类信息PooledUnsafeDirectByteBuf可知，采用的是池化基于直接内存
          System.out.println(buf2.getClass());//输出class io.netty.buffer.PooledUnsafeDirectByteBuf
          //输出容量
          log(buf2);
      
  
      /**
       * 输出ByteBuf的工具类
       * */
      public static void log(ByteBuf buffer) 
          int length = buffer.readableBytes();
          int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
          StringBuilder buf = new StringBuilder(rows * 80 * 2)
                  .append("read index:").append(buffer.readerIndex())
                  .append(" write index:").append(buffer.writerIndex())
                  .append(" capacity:").append(buffer.capacity())
                  .append(NEWLINE);
          appendPrettyHexDump(buf, buffer);
          System.out.println(buf.toString());
      
  
  

• 输出结果
  

四、ByteBuf (池化 vs 非池化)

4.1、默认采用的是池化基于直接内存的 代码示例

• 代码示例

  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.ByteBufAllocator;
  /**
   * @description: ByteBuf (池化 vs 非池化)
   * @author: xz
   */
  public class ByteBufTest3 
      public static void main(String[] args) 
          getPooledByteBuf();
      
      /**
       * 默认采用的是池化基于直接内存
       * */
      public static void getPooledByteBuf()
          //创建了一个默认的 ByteBuf（创建池化基于直接内存的 ByteBuf）
          ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer();
          //通过输出类信息PooledUnsafeDirectByteBuf可知，采用的是池化基于直接内存
          System.out.println(buf1.getClass());//输出class io.netty.buffer.PooledUnsafeDirectByteBuf
      
  
  

• 输出结果
  

4.1、设置vm参数采用非池化基于直接内存的 代码示例

• 代码示例

  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.ByteBufAllocator;
  /**
   * @description: ByteBuf (池化 vs 非池化)
   * @author: xz
   */
  public class ByteBufTest3 
      public static void main(String[] args) 
          getUnPooledByteBuf();
      
  
      /**
       * 通过设置vm参数 -Dio.netty.allocator.type=unpooled之后，
       * 采用的是非池化基于直接内存
       * */
      public static void getUnPooledByteBuf()
          //创建了一个默认的 ByteBuf（创建非池化基于直接内存的 ByteBuf）
          ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer();
          //通过输出类信息UnpooledByteBufAllocator可知，采用的是非池化基于直接内存
          System.out.println(buf1.getClass());//输出class io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeNoCleanerDirectByteBuf
      
  
  

• 通过设置vm参数 -Dio.netty.allocator.type=unpooled之后，采用非池化基于直接内存。
  

• 输出结果
  

五、ByteBuf 组成

• ByteBuf 由四部分组成，最开始读写指针都在 0 位置。
  

六、ByteBuf 写入

6.1、ByteBuf 写入方法

• ByteBuf 写入方法列表，省略一些不重要的方法

  方法签名	含义	备注
  writeBoolean(boolean value)	写入 boolean 值	用一字节 01|00 代表 true|false
  writeByte(int value)	写入 byte 值
  writeShort(int value)	写入 short 值
  writeInt(int value)	写入 int 值	Big Endian，即 0x250，写入后 00 00 02 50，先写高位在写低位
  writeIntLE(int value)	写入 int 值	Little Endian，即 0x250，写入后 50 02 00 00，先写低位在写高位
  writeLong(long value)	写入 long 值
  writeChar(int value)	写入 char 值
  writeFloat(float value)	写入 float 值
  writeDouble(double value)	写入 double 值
  writeBytes(ByteBuf src)	写入 netty 的 ByteBuf
  writeBytes(byte[] src)	写入 byte[]
  writeBytes(ByteBuffer src)	写入 nio 的 ByteBuffer
  int writeCharSequence(CharSequence sequence, Charset charset)	写入字符串

6.2、ByteBuf 写入的 代码示例

• 代码示例

  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.ByteBufAllocator;
  import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
  import static io.netty.util.internal.StringUtil.NEWLINE;
  /**
   * @description: ByteBuf (写入)
   * @author: xz
   */
  public class ByteBufTest4 
      public static void main(String[] args) 
          getWriteBytes();
      
      public static void getWriteBytes()
          //创建了一个默认的 ByteBuf（池化基于直接内存的 ByteBuf），初始容量是 10
          ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(10);
          //1、先写入 4 个字节
          buf1.writeBytes(new byte[]1, 2, 3, 4);
          log(buf1);
          //2、再写入一个 int 整数，也是 4 个字节
          buf1.writeInt(5);
          log(buf1);
      
  
      /**
       * 输出ByteBuf的工具类
       * */
      public static void log(ByteBuf buffer) 
          int length = buffer.readableBytes();
          int rows 以上是关于Netty ByteBuf入门的主要内容，如果未能解决你的问题，请参考以下文章
   
   Netty系列之：netty中的ByteBuf详解
   netty里的ByteBuf扩容源码分析
   Netty之ByteBuf原理解析及应用
   ByteBuf in Netty
   ByteBuf in Netty
   ByteBuf in Netty