Netty编程（五）—— ByteBuf

发表于

这是我参与11月更文挑战的第23天，活动详情查看：2021最后一次更文挑战

创建

ByteBuf 可以通过ByteBufAllocator选择allocator并调用对应的 buffer( )方法来创建的，默认使用直接内存作为ByteBuf，容量为256个字节，可以指定初始容量的大小。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
java复制代码public class ByteBufStudy {
    public static void main(String[] args) {
        // 创建ByteBuf
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);
        ByteBufUtil.log(buffer);

        // 向buffer中写入数据
        StringBuilder sb = new StringBuilder();
        for(int i = 0; i < 20; i++) {
            sb.append("a");
        }
        buffer.writeBytes(sb.toString().getBytes(StandardCharsets.UTF_8));

        // 查看写入结果
        ByteBufUtil.log(buffer);
    }
}

结果可视化：

在这里插入图片描述

  • 当ByteBuf的容量无法容纳所有数据时，ByteBuf会进行扩容操作
  • 如果在handler中创建ByteBuf，建议使用 ChannelHandlerContext ctx.alloc().buffer() 来创建

直接内存与堆内存

通过下面这种方法创建的ByteBuf，使用的是基于直接内存的ByteBuf

1
java复制代码ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);

此外，还可以使用下面的代码来创建池化基于堆的 ByteBuf

1
java复制代码ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(16);

也可以使用下面的代码来创建池化基于直接内存的 ByteBuf

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

池化与非池化

类似于线程池的思想，预先把昂贵的资源创建好，省去创建的时间和步骤，用完后需要归还。池化的最大意义在于可以重用 ByteBuf，他的优点有：

  • 没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加 GC 压力
  • 有了池化，则可以重用池中 ByteBuf 实例，并且采用了与 jemalloc 类似的内存分配算法提升分配效率
  • 高并发时，池化功能更节约内存，减少内存溢出的可能
1
2
3
4
5
6
7
8
9
10
11
12
java复制代码public class ByteBufStudy {
    public static void main(String[] args) {
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);
        System.out.println(buffer.getClass());

        buffer = ByteBufAllocator.DEFAULT.heapBuffer(16);
        System.out.println(buffer.getClass());

        buffer = ByteBufAllocator.DEFAULT.directBuffer(16);
        System.out.println(buffer.getClass());
    }
}
1
2
3
4
5
6
7
8
arduino复制代码// 使用池化的直接内存
class io.netty.buffer.PooledUnsafeDirectByteBuf
    
// 使用池化的堆内存    
class io.netty.buffer.PooledUnsafeHeapByteBuf
    
// 使用池化的直接内存    
class io.netty.buffer.PooledUnsafeDirectByteBuf

组成

ByteBuf主要有以下几个组成部分

  • 最大容量与当前容量
+ 在构造ByteBuf时，可传入两个参数，分别代表初始容量和最大容量，若未传入第二个参数（最大容量），最大容量默认为Integer.MAX\_VALUE
+ 当ByteBuf容量无法容纳所有数据时，会进行扩容操作，若**超出最大容量**，会抛出`java.lang.IndexOutOfBoundsException`异常
  • 对于读写操作与ByteBuffer只用position进行控制不同的是， ByteBuf分别由读指针和写指针两个指针控制
+ 进行读写操作时，无需进行模式的切换


    - 读指针前的部分被称为废弃部分，是已经读过的内容
    - 读指针与写指针之间的空间称为可读部分
    - 写指针与当前容量之间的空间称为可写部分

在这里插入图片描述

写入

常用方法如下

方法签名 含义 备注
writeBoolean(boolean value) 写入 boolean 值 **用一字节 01
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) 写入字符串 CharSequence为字符串类的父类，第二个参数为对应的字符集

注意

  • 这些方法的未指明返回值的，其返回值都是 ByteBuf，意味着可以链式调用来写入不同的数据
  • 网络传输中，默认习惯是 Big Endian，使用 writeInt(int value)
  • CharSequence 是String、StringBuffer、StringBuilder的父类

下面这个例子分别写入了字节类型，int类型以及long类型，值得注意的是，写long类型会进行扩容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
java复制代码public class test {

    public static void main(String[] args) {
        // 创建ByteBuf
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16, 20);

        // 向buffer中写入byte类型，1字节大小
        buffer.writeBytes(new byte[]{1, 2, 3, 4});
		// 向buffer中写入int类型，4字节大小
        buffer.writeInt(5);
		// 向buffer中小端写入int类型，4字节大小
        buffer.writeIntLE(6);
		// 向buffer中写入long类型，8字节大小
        buffer.writeLong(7);
        log(buffer);
    }

    private static void log(ByteBuf buffer) {
        //ByteBuf 可视化
        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中的容量无法容纳写入的数据时，会进行扩容操作，借用上面说的例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
perl复制代码buffer.writeLong(7);
log(buffer);

// 扩容前
read index:0 write index:12 capacity:16
...

// 扩容后
read index:0 write index:20 capacity:20
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 00 00 00 05 06 00 00 00 00 00 00 00 |................|
|00000010| 00 00 00 07                                     |....            |
+--------+-------------------------------------------------+----------------+

扩容规则

  • 如何写入后数据大小未超过 512 字节，则选择下一个 16 的整数倍进行扩容，例如写入后大小为 12 字节，则扩容后 capacity 是 16 字节
  • 如果写入后数据大小超过 512 字节，则选择下一个2n2^n2n ，例如写入后大小为 513 字节，则扩容后 capacity 是 210=1024 字节（292^929=512 已经不够了）
  • 扩容不能超过 maxCapacity，否则会抛出java.lang.IndexOutOfBoundsException异常

读取

读取主要是通过一系列read方法进行读取，读取时会根据读取数据的字节数移动读指针，此外Netty支持重复读取，如果需要重复读取，需要调用buffer.markReaderIndex()对读指针进行标记，并通过buffer.resetReaderIndex()将读指针恢复到mark标记的位置。具体可以看下面的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
java复制代码public static void main(String[] args) {
        // 创建ByteBuf
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16, 20);

        // 向buffer中写入数据
        buffer.writeBytes(new byte[]{1, 2, 3, 4});
        buffer.writeInt(5);
        ByteBufferUtil.log(buffer);

        // 读取4个字节
        System.out.println(buffer.readByte());
        System.out.println(buffer.readByte());
        System.out.println(buffer.readByte());
        System.out.println(buffer.readByte());
        ByteBufferUtil.log(buffer);

        // 通过mark与reset实现重复读取
        buffer.markReaderIndex();
        System.out.println(buffer.readInt());
        ByteBufferUtil.log(buffer);

        // 恢复到mark标记处
        buffer.resetReaderIndex();
        ByteBufferUtil.log(buffer);
    }

在这里插入图片描述

释放

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

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

释放规则

因为 pipeline 的存在，一般需要将 ByteBuf 传递给下一个 ChannelHandler，如果在每个 ChannelHandler 中都去调用 release ，就失去了传递性（如果在这个 ChannelHandler 内这个 ByteBuf 已完成了它的使命，那么便无须再传递），所以基本规则是，谁是最后使用者，谁负责 release

  • 入站 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 :**




1
java复制代码while (!buffer.release()) {}

当ByteBuf被传到了pipeline的head与tail时，ByteBuf会被其中的方法彻底释放。

切片Slice

之前在NIO网络编程（十）—— 零拷贝技术 - 掘金 (juejin.cn)中介绍了一下NIO中的零拷贝问题，这里介绍一下Netty中使用Slice方法实现零拷贝的。

ByteBuf切片是零拷贝的体现之一，对原始 ByteBuf 进行切片成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制，还是使用原始 ByteBuf 的内存，切片后的 ByteBuf 维护独立的 read，write 指针。

  • 得到分片后的buffer后，要调用其retain方法，使其内部的引用计数加一。避免原ByteBuf释放，导致切片buffer无法使用
  • 修改原ByteBuf中的值，也会影响切片后得到的ByteBuf
  • 切片不能扩容，即不能向切片写入内容
  • 一般使用slice，就需要retain，使用完后使用 release

img

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
java复制代码public static void main(String[] args) {
        // 创建ByteBuf
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16, 20);

        // 向buffer中写入数据
        buffer.writeBytes(new byte[]{1, 2, 3, 4, 5, 6, 7, 8, 9, 10});

        // 将buffer分成两部分，参数：start，length，在切片过程中没有发生数据复制，同一块内存
        ByteBuf slice1 = buffer.slice(0, 5);
        ByteBuf slice2 = buffer.slice(5, 5);

        // 需要让分片的buffer引用计数加一
        // 避免原Buffer释放导致分片buffer无法使用
        slice1.retain();
        slice2.retain();

        ByteBufferUtil.log(slice1);
        ByteBufferUtil.log(slice2);

        // 更改原始buffer中的值
        System.out.println("===========修改原buffer中的值===========");
        buffer.setByte(0,5);

        System.out.println("===========打印slice1===========");
        ByteBufferUtil.log(slice1);
    }

上方的代码的执行结果：

在这里插入图片描述

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%