Netty（8）netty组件：ByteBuf

约定不等于承诺〃 2024-04-08 10:40 165阅读 0赞

概述

ByteBuf是对字节数据的封装

ByteBuf的容量是可以自动扩容的

创建

ByteBuf 如果不指定容量的话，默认是256

  1. //指定初始容量为10
  2. ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(10);

使用

  1. ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
  2. System.out.println(buf);
  3. StringBuilder sb = new StringBuilder();
  4. for (int i = 0; i < 320; i++) {
  6.     sb.append("a");
  7. }
  8. buf.writeBytes(sb.toString().getBytes());
  9. System.out.println(buf);

运行结果

  1. PooledUnsafeDirectByteBuf(ridx: 0, widx: 0, cap: 256)
  2. PooledUnsafeDirectByteBuf(ridx: 0, widx: 320, cap: 512)

调试的工具类

  1. import io.netty.buffer.ByteBuf;
  2. import io.netty.buffer.ByteBufAllocator;
  4. import static io.netty.buffer.ByteBufUtil.appendPrettyHexDump;
  5. import static io.netty.util.internal.StringUtil.NEWLINE;
  7. public static void log(ByteBuf buffer) {
  9.     int length = buffer.readableBytes();
  10.     int rows = length / 16 + (length % 15 == 0 ? 0 : 1) + 4;
  11.     StringBuilder buf = new StringBuilder(rows * 80 * 2)
  12.             .append("read index:").append(buffer.readerIndex())
  13.             .append(" write index:").append(buffer.writerIndex())
  14.             .append(" capacity:").append(buffer.capacity())
  15.             .append(NEWLINE);
  16.     appendPrettyHexDump(buf, buffer);
  17.     System.out.println(buf.toString());
  18. }

使用上面的方法打印出来的数据如下（for循环的次数改为了32）：
在这里插入图片描述

ByteBuf支持的内存分配

直接内存

分配效率较低，但是读写效果高

netty默认情况下，使用的是直接内存作为ByteBuf的内存（也可以选择）

下面的代码来创建池化基于直接内存的 ByteBuf

  1. ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);
  2. //不指定的话，默认使用的是直接内存
  • 直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用
  • 直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放

堆内存

分配效率较高，但是读写效率比较低

下面的代码来创建池化基于堆的 ByteBuf

  1. ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10);

支持池化的管理

池化管理：对于创建比较慢的一些资源，可以使用池化管理的思想进行优化

池化 vs 非池化

池化的 最大意义在于可以重用 ByteBuf，优点有

  • 没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加 GC 压力
  • 有了池化，则可以重用池中 ByteBuf 实例，并且采用了与 jemalloc 类似的内存分配算法提升分配效率
  • 高并发时，池化功能更节约内存，减少内存溢出的可能

开启池化功能

池化功能是否开启，可以通过下面的系统环境变量来设置

  1. -Dio.netty.allocator.type={
  2.     unpooled|pooled}
  • 4.1 以后，非 Android 平台默认启用池化实现，Android 平台启用非池化实现
  • 4.1 之前，池化功能还不成熟，默认是非池化实现

查看当前的实现是什么
在这里插入图片描述

在IDEA进行设置

在这里插入图片描述

ByteBuf组成

ByteBuf 由四部分组成

  1. 最大容量：整数的最大值，21
  2. 容量：容量与最大容量之间为可扩容区域
  3. 读指针：读指针到写指针的位置，称为可读部分
  4. 写指针：写指针到可写位置，称为可写部分
  5. 已经读过的这些数据，称为废弃部分

在这里插入图片描述
最开始读写指针都在 0 位置

写入

方法列表，省略一些不重要的方法












































































方法签名 含义 备注
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) 写入字符串

注意

  • 这些方法的未指明返回值的，其返回值都是 ByteBuf，意味着可以链式调用
  • 网络传输，默认习惯是 Big Endian

大端：一个整数写入时，是先写高位字节还是低位字节，大端就是先写高位的
小端：先写低位的，再写高位的

还有一类方法是 set 开头的一系列方法，也可以写入数据，但不会改变写指针位置

writeBytes

先写入 4 个字节

  1. buffer.writeBytes(new byte[]{
  2.     1, 2, 3, 4});
  3. log(buffer);

结果

  1. read index:0 write index:4 capacity:10
  2.          +-------------------------------------------------+
  3.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  4. +--------+-------------------------------------------------+----------------+
  5. |00000000| 01 02 03 04                                     |....            |
  6. +--------+-------------------------------------------------+----------------+

writeInt

没有带LE，是大端写入，从高位写

在上面的基础上，再写四个字节

  1. buffer.writeInt(5);
  2. log(buffer);

结果是

  1. read index:0 write index:8 capacity:10
  2.          +-------------------------------------------------+
  3.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  4. +--------+-------------------------------------------------+----------------+
  5. |00000000| 01 02 03 04 00 00 00 05                         |........        |
  6. +--------+-------------------------------------------------+----------------+

扩容

如果在上面的基础上，再进行希尔一个int整数时，便会触发扩容

扩容规则

如何写入后数据大小未超过 512，则选择下一个 16 的整数倍，例如写入后大小为 12 ，则扩容后 capacity 是 16

如果写入后数据大小超过 512，则选择下一个2^n，例如写入后大小为 513，则扩容后 capacity 是 2^10=1024
（2^9=512 已经不够了）

扩容不能超过 max capacity 会报错

再次写入一个整数

  1. buffer.writeInt(6);
  2. log(buffer);

结果是

  1. read index:0 write index:12 capacity:16
  2.          +-------------------------------------------------+
  3.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  4. +--------+-------------------------------------------------+----------------+
  5. |00000000| 01 02 03 04 00 00 00 05 00 00 00 06             |............    |
  6. +--------+-------------------------------------------------+----------------+

读取

在写入这个章节里面，我们进行了三次写入，每次写入一个整数（四字节）

现在，在之前的基础上，进行读，如下：读四次，每次读一个字节

  1. System.out.println(buffer.readByte());
  2. System.out.println(buffer.readByte());
  3. System.out.println(buffer.readByte());
  4. System.out.println(buffer.readByte());
  5. log(buffer);

运行结果

  1. 1
  2. 2
  3. 3
  4. 4
  5. read index:4 write index:12 capacity:16
  6.          +-------------------------------------------------+
  7.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  8. +--------+-------------------------------------------------+----------------+
  9. |00000000| 00 00 00 05 00 00 00 06                         |........        |
  10. +--------+-------------------------------------------------+----------------+

现在，读过的内容，就属于废弃部分了，再读只能读那些尚未读取的部分

重复读取

如果需要重复读取 int 整数 5，可以在 read 前先做个标记 mark

  1. buffer.markReaderIndex();
  2. System.out.println(buffer.readInt());
  3. log(buffer);

运行结果：

  1. 5
  2. read index:8 write index:12 capacity:16
  3.          +-------------------------------------------------+
  4.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  5. +--------+-------------------------------------------------+----------------+
  6. |00000000| 00 00 00 06                                     |....            |
  7. +--------+-------------------------------------------------+----------------+

现在，进行重复读取

要重复读取的话，重置到标记位置 reset，

将其还原到之前的位置，使用reset

  1. buffer.resetReaderIndex();
  2. log(buffer);

结果：还原成功

  1. read index:4 write index:12 capacity:16
  2.          +-------------------------------------------------+
  3.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  4. +--------+-------------------------------------------------+----------------+
  5. |00000000| 00 00 00 05 00 00 00 06                         |........        |
  6. +--------+-------------------------------------------------+----------------+

其他方法

采用 get 开头的一系列方法（通过索引去获取），这些方法不会改变 read index

ByteBuf的内存回收

ByteBuf回收方法

Netty 中有堆外内存的 ByteBuf 实现，堆外内存最好是手动来释放，而不是等 GC 垃圾回收。

  • UnpooledHeapByteBuf使用的是 JVM 内存，只需等 GC 回收内存即可
  • UnpooledDirectByteBuf 使用的就是直接内存了，需要特殊的方法来回收内存
  • PooledByteBuf 和它的子类使用了池化机制，需要更复杂的规则来回收内存

netty提供的统一接口进行回收方法

Netty 这里**采用了引用计数法**来控制回收内存，每个 ByteBuf 都实现了 ReferenceCounted 接口

  • 每个 ByteBuf 对象的初始计数为 1
  • 调用 release 方法计数减 1，如果计数为 0，ByteBuf 内存被回收
  • 调用 retain 方法计数加 1，表示调用者没用完之前，其它 handler 即使调用了 release 也不会造成回收
  • 当 计数为 0 时，底层内存会被回收，这时即使 ByteBuf 对象还在，其各个方法均无法正常使用

release在何处进行

因为 pipeline 的存在，一般需要将 ByteBuf 传递给下一个 ChannelHandler，如果在 finally 中 release 了，就失去了传递性（当然，如果在这个 ChannelHandler 内这个 ByteBuf 已完成了它的使命，那么便无须再传递）

基本规则是，谁是最后使用者，谁负责 release，详细分析如下

谁最后使用ByteBuf，谁就去使用release进行处理

  • 起点，对于 NIO 实现来讲，在 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#read 方法中首次创建 ByteBuf 放入 pipeline（line 163 pipeline.fireChannelRead(byteBuf)）

  • 入站 ByteBuf 处理原则

    • 对原始 ByteBuf 不做处理，调用 ctx.fireChannelRead(msg) 向后传递，这时无须 release
    • 将原始 ByteBuf 转换为其它类型的 Java 对象，这时 ByteBuf 就没用了，必须 release
    • 如果不调用 ctx.fireChannelRead(msg) 向后传递，那么也必须 release
    • 注意各种异常，如果 ByteBuf 没有成功传递到下一个 ChannelHandler，必须 release
    • 假设消息一直向后传，那么 TailContext 会负责释放未处理消息（原始的 ByteBuf）

  • 出站 ByteBuf 处理原则

    • 出站消息最终都会转为 ByteBuf 输出，一直向前传，由 HeadContext flush 后 release

  • 异常处理原则

    • 有时候不清楚 ByteBuf 被引用了多少次，但又必须彻底释放，可以循环调用 release 直到返回 true

slice【零拷贝】

这里的零拷贝是指：减少数据复制

slice作用：【零拷贝】的体现之一，对原始 ByteBuf 进行切片成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制，还是使用原始 ByteBuf 的内存，切片后的 ByteBuf 维护独立的 read，write 指针
在这里插入图片描述

  1. ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(10);
  2. buf.writeBytes(new byte[]{
  3.     'a','b','c','d','e','f','g','h','i','j'});
  4. log(buf);
  6. // 在切片过程中，没有发生数据复制
  7. ByteBuf f1 = buf.slice(0, 5);//从a到e
  8. ByteBuf f2 = buf.slice(5, 5);//从f到j
  9. log(f1);
  10. log(f2);
  11. System.out.println("===========演示是否是同一个内存=============");
  12. //修改其中的一个数据，查看buf和f1是否都改变了
  13. f1.setByte(0, 'b');
  14. log(f1);
  15. log(buf);
  16. //通过查看日志可知，二者都发生了变化，所以是操作的同一个内容
  17. //原始 ByteBuf 也会受影响，因为底层都是同一块内存

运行结果：

  1. read index:0 write index:10 capacity:10
  2.          +-------------------------------------------------+
  3.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  4. +--------+-------------------------------------------------+----------------+
  5. |00000000| 61 62 63 64 65 66 67 68 69 6a                   |abcdefghij      |
  6. +--------+-------------------------------------------------+----------------+
  7. read index:0 write index:5 capacity:5
  8.          +-------------------------------------------------+
  9.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  10. +--------+-------------------------------------------------+----------------+
  11. |00000000| 61 62 63 64 65                                  |abcde           |
  12. +--------+-------------------------------------------------+----------------+
  13. read index:0 write index:5 capacity:5
  14.          +-------------------------------------------------+
  15.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  16. +--------+-------------------------------------------------+----------------+
  17. |00000000| 66 67 68 69 6a                                  |fghij           |
  18. +--------+-------------------------------------------------+----------------+
  19. ========================
  20. read index:0 write index:5 capacity:5
  21.          +-------------------------------------------------+
  22.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  23. +--------+-------------------------------------------------+----------------+
  24. |00000000| 62 62 63 64 65                                  |bbcde           |
  25. +--------+-------------------------------------------------+----------------+
  26. read index:0 write index:10 capacity:10
  27.          +-------------------------------------------------+
  28.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  29. +--------+-------------------------------------------------+----------------+
  30. |00000000| 62 62 63 64 65 66 67 68 69 6a                   |bbcdefghij      |
  31. +--------+-------------------------------------------------+----------------+
  33. Process finished with exit code 0

注意点

注意点1：切片后区间被固定

注意点1：调用 slice 进行切片，无参 slice 是从原始 ByteBuf 的 read index 到 write index 之间的内容进行切片，切片后的 max capacity 被固定为这个区间的大小，因此不能追加 write

如果执行，会报 IndexOutOfBoundsException 异常

  1. ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(10);
  2. buf.writeBytes(new byte[]{
  3.     'a','b','c','d','e','f','g','h','i','j'});
  4. log(buf);
  6. // 在切片过程中，没有发生数据复制
  7. ByteBuf f1 = buf.slice(0, 5);
  8. // 如果想再写入一个x数据到f1中：'a','b','c','d','e', 'x'
  9. ByteBuf f2 = buf.slice(5, 5);
  10. f1.writeByte('x');
  11. log(f1);
  12. log(f2);

运行结果

  1. read index:0 write index:10 capacity:10
  2.          +-------------------------------------------------+
  3.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  4. +--------+-------------------------------------------------+----------------+
  5. |00000000| 61 62 63 64 65 66 67 68 69 6a                   |abcdefghij      |
  6. +--------+-------------------------------------------------+----------------+
  7. Exception in thread "main" java.lang.IndexOutOfBoundsException: writerIndex(5) + minWritableBytes(1) exceeds maxCapacity(5): UnpooledSlicedByteBuf(ridx: 0, widx: 5, cap: 5/5, unwrapped: PooledUnsafeDirectByteBuf(ridx: 0, widx: 10, cap: 10))
注意点2：release

注意点2：如果原始的ByteBuf被release（释放内存），由该ByteBuf切片的内容便不能再使用

  1. ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(10);
  2. buf.writeBytes(new byte[]{
  3.     'a','b','c','d','e','f','g','h','i','j'});
  4. log(buf);
  6. // 在切片过程中，没有发生数据复制
  7. ByteBuf f1 = buf.slice(0, 5);
  8. ByteBuf f2 = buf.slice(5, 5);
  10. System.out.println("释放原有 byteBuf 内存");
  11. buf.release();
  12. log(f1);

运行结果

  1. read index:0 write index:10 capacity:10
  2.          +-------------------------------------------------+
  3.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  4. +--------+-------------------------------------------------+----------------+
  5. |00000000| 61 62 63 64 65 66 67 68 69 6a                   |abcdefghij      |
  6. +--------+-------------------------------------------------+----------------+
  7. 释放原有 byteBuf 内存
  8. Exception in thread "main" io.netty.util.IllegalReferenceCountException: refCnt: 0

解决方法：使用retain（retain：+1；release：-1）方法

  1. ByteBuf buf= ByteBufAllocator.DEFAULT.buffer(10);
  2. buf.writeBytes(new byte[]{
  3.     'a','b','c','d','e','f','g','h','i','j'});
  4. log(buf);
  6. // 在切片过程中，没有发生数据复制
  7. ByteBuf f1 = buf.slice(0, 5);
  8. //使buf，release后不影响f1
  9. f1.retain();
  10. ByteBuf f2 = buf.slice(5, 5);
  11. //使buf，release后不影响f2
  12. f2.retain();
  13. log(f1);
  14. log(f2);
  16. System.out.println("释放原有 byteBuf 内存");
  17. buf.release();
  18. log(f1);
  19. //f1和f2使用完了，自己去release
  20. f1.release();
  21. f2.release();

duplicate【复制，零拷贝】

作用：【零拷贝】的体现之一，就好比截取了原始 ByteBuf 所有内容，并且没有 max capacity 的限制，也是与原始 ByteBuf 使用同一块底层内存，只是读写指针是独立的

在这里插入图片描述

copy

会将底层内存数据进行深拷贝，因此无论读写，都与原始 ByteBuf 无关

CompositeByteBuf【零拷贝】

作用：【零拷贝】的体现之一，可以将多个 ByteBuf 合并为一个逻辑上的 ByteBuf，避免拷贝

CompositeByteBuf 是一个组合的 ByteBuf，它内部维护了一个 Component 数组，每个 Component 管理一个 ByteBuf，记录了这个 ByteBuf 相对于整体偏移量等信息，代表着整体中某一段的数据。

  • 优点，对外是一个虚拟视图，组合这些 ByteBuf 不会产生内存复制

  • 缺点，复杂了很多，多次操作会带来性能的损耗（如每次都要去维护写指针的位置）

    ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer();
    buf1.writeBytes(new byte[]{

    1. 1, 2, 3, 4, 5});

    ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer();
    buf2.writeBytes(new byte[]{

    1. 6, 7, 8, 9, 10});

    //这种方法进行了多次数据复制
    /ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
    buffer.writeBytes(buf1).writeBytes(buf2);
    log(buffer);    /

    //把多个小的ByteBuf组成到了一起
    CompositeByteBuf buffer = ByteBufAllocator.DEFAULT.compositeBuffer();
    /*
    addComponents：
    参数1：自动增长写指针，如果不填true，

    1.    addComponents/addComponent都默认不会去调整写指针的位置；
    2.    所以数据不会

    参数2和参数3：这些是可变参数，需要添加的buf
    */
    buffer.addComponents(true, buf1, buf2);
    log(buffer);

Unpooled

Unpooled 是一个工具类，类如其名，提供了非池化的 ByteBuf 创建、组合、复制等操作

这里仅介绍其跟【零拷贝】相关的 wrappedBuffer 方法，可以用来包装 ByteBuf

  1. ByteBuf buf1 = ByteBufAllocator.DEFAULT.buffer(5);
  2. buf1.writeBytes(new byte[]{
  3.     1, 2, 3, 4, 5});
  4. ByteBuf buf2 = ByteBufAllocator.DEFAULT.buffer(5);
  5. buf2.writeBytes(new byte[]{
  6.     6, 7, 8, 9, 10});
  8. // 当包装 ByteBuf 个数超过一个时, 底层使用了 CompositeByteBuf
  9. ByteBuf buf3 = Unpooled.wrappedBuffer(buf1, buf2);
  10. System.out.println(ByteBufUtil.prettyHexDump(buf3));

输出

  1. +-------------------------------------------------+
  2.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  3. +--------+-------------------------------------------------+----------------+
  4. |00000000| 01 02 03 04 05 06 07 08 09 0a                   |..........      |
  5. +--------+-------------------------------------------------+----------------+

也可以用来包装普通字节数组，底层也不会有拷贝操作

  1. ByteBuf buf4 = Unpooled.wrappedBuffer(new byte[]{
  2.     1, 2, 3}, new byte[]{
  3.     4, 5, 6});
  4. System.out.println(buf4.getClass());
  5. System.out.println(ByteBufUtil.prettyHexDump(buf4));

输出

  1. class io.netty.buffer.CompositeByteBuf
  2.          +-------------------------------------------------+
  3.          |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
  4. +--------+-------------------------------------------------+----------------+
  5. |00000000| 01 02 03 04 05 06                               |......          |
  6. +--------+-------------------------------------------------+----------------+

ByteBuf的优势总结

池化：可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能

读写指针分离，不需要像 ByteBuffer 一样切换读写模式

可以自动扩容

支持链式调用，使用更流畅

很多地方体现零拷贝，例如 slice、duplicate、CompositeByteBuf

发表评论

表情：
评论列表 （有 0 条评论，165人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Netty bytebuf 内存泄漏

    客户端发送消息，一直处于高峰，每秒几千条。或曲线状态。此时客户端内存不断飙升 分析原因：客户端发送存在大量缓存，因服务端接收能力有限。 客户端注掉： bootstrap.

    蔚落蔚落/ 0 赞/ 272 阅读

    相关 Netty之ByteBuf

    什么是ByteBuf ByteBuf是Netty中非常重要的一个组件，他就像物流公司的运输工具：卡车，火车，甚至是飞机。而物流公司靠什么盈利，就是靠运输货物，可想而知By

    红太狼红太狼/ 0 赞/ 283 阅读

    相关 Netty（ByteBuf）

        网络数据的基本单位总是字节。Java  NIO提供了ByteBuffer作为它的字节容器，但是这个类使用起来过于复杂，而且也有些繁琐。Netty的ByteBuffer替

    我就是我我就是我/ 0 赞/ 344 阅读