Netty实战（五）

一、什么是ByteBuf

我们前面说过，网络数据的基本单位总是字节。Java NIO 提供了 ByteBuffer 作为它的字节容器，但是这个类使用起来过于复杂，而且也有些繁琐。ByteBuffer 替代品是 ByteBuf，一个强大的实现，既解决了 JDK API 的局限性，又为网络应用程序的开发者提供了更好的 API。

下面我们将会说明和 JDK 的 ByteBuffer 相比，ByteBuf 的卓越功能性和灵活性

二、 ByteBuf 的 API

Netty 的数据处理 API 通过两个组件暴露——abstract class ByteBuf 和 interface ByteBufHolder。

下面是一些 ByteBuf API 的优点：

• 它可以被用户自定义的缓冲区类型扩展；
• 通过内置的复合缓冲区类型实现了透明的零拷贝；
• 容量可以按需增长（类似于 JDK 的 StringBuilder）；
• 在读和写这两种模式之间切换不需要调用 ByteBuffer 的 flip()方法；
• 读和写使用了不同的索引；
• 支持方法的链式调用；
• 支持引用计数；
• 支持池化。

其他类可用于管理 ByteBuf 实例的分配，以及执行各种针对于数据容器本身和它所持有的数据的操作。

三、ByteBuf 类——Netty 的数据容器

因为所有的网络通信都涉及字节序列的移动，所以高效易用的数据结构明显是必不可少的。Netty 的 ByteBuf 实现满足并超越了这些需求。

3.1 ByteBuf如何工作？

ByteBuf 维护了两个不同的索引：一个用于读取，一个用于写入。

当你从 ByteBuf 读取时，它的 readerIndex 将会被递增已经被读取的字节数。
当你写入 ByteBuf 时，它的writerIndex 也会被递增。

ByteBuf 的布局结构和状态如下：

那么这两个索引之间有什么关系？如果读取字节直到 readerIndex 达到和 writerIndex 同样的值时会发生什么？

在那时，我们将会到达“可以读取的”数据的末尾。就如同试图读取超出数组末尾的数据一样，试图读取超出该点的数据将会触发IndexOutOfBoundsException。

名称以 read 或者 write 开头的 ByteBuf 方法，将会推进其对应的索引，而名称以 set 或者 get 开头的操作则不会。

后面的这些方法将在作为一个参数传入的一个相对索引上执行操作。可以指定 ByteBuf 的最大容量。试图移动写索引（即 writerIndex）超过这个值将会触发一个异常。（默认的限制是Integer.MAX_VALUE。）

PS：也就是说用户直接或者间接使 capacity(int)或者 ensureWritable(int)方法来增加超过该最大容量时抛出异常。

3.2 ByteBuf 的使用模式

首先我们记住ByteBuf：一个由不同的索引分别控制读访问和写访问的字节数组。

3.2.1 堆缓冲区

最常用的 ByteBuf 模式是将数据存储在 JVM 的堆空间中。这种模式被称为支撑数组（backing array），它能在没有使用池化的情况下提供快速的分配和释放。这种方式，非常适合于有遗留的数据需要处理的情况。

代码展示：

ByteBuf heapBuf = ...;//检查ByteBuf是否有一个支撑数组if (heapBuf.hasArray()) {//如果有，则获取对该数组的引用byte[] array = heapBuf.array();//计算第一个字节的偏移量int offset = heapBuf.arrayOffset() + heapBuf.readerIndex();//获取可读字节数int length = heapBuf.readableBytes();//使用数组、偏移量和长度作为参数调用你的方法handleArray(array, offset, length);}

PS：当 hasArray()方法返回 false 时，尝试访问支撑数组将触发一个 UnsupportedOperationException。这个模式类似于 JDK 的 ByteBuffer 的用法。

3.2.2 直接缓冲区

直接缓冲区是另外一种 ByteBuf 模式。

我们都希望用于对象创建的内存分配永远都来自于堆中。但这并不是必须的，因为ByteBuffer 类允许 JVM 实现通过本地调用来分配内存。

这主要是为了避免在每次调用本地 I/O 操作之前（或者之后）将缓冲区的内容复制到一个中间缓冲区（或者从中间缓冲区把内容复制到缓冲区）。

直接缓冲区的内容将驻留在常规的会被垃圾回收的堆之外（它的内容可能不会被堆回收）。如果你的数据包含在一个在堆上分配的缓冲区中，那么在通过套接字发送它之前，JVM将会在内部把你的缓冲区复制到一个直接缓冲区中。

直接缓冲区的主要缺点是：
1、相对于基于堆的缓冲区，它们的分配和释放都较为昂贵。
2、因为数据不是在堆上，所以你不得不进行一次复制，如代码所示。

ByteBuf directBuf = ...;//检查 ByteBuf 是否由数组支撑。如果不是，则这是一个直接缓冲区if (!directBuf.hasArray()) {//获取字节可读数int length = directBuf.readableBytes();//分配一个新的数组来保存具有该长度的字节数据byte[] array = new byte[length];//将字节复制到该数组directBuf.getBytes(directBuf.readerIndex(), array);//开始调用handleArray(array, 0, length);}

显然，与使用支撑数组相比，这涉及的工作更多。因此，如果事先知道容器中的数据将会被作为数组来访问，你可能更愿意使用堆内存。

3.2.3 复合缓冲区

第三种也是最后一种模式使用的是复合缓冲区，它为多个 ByteBuf 提供一个聚合视图。

在这里你可以根据需要添加或者删除 ByteBuf 实例，这是一个 JDK 的 ByteBuffer 实现完全缺失的特性。Netty 通过一个 ByteBuf 子类——CompositeByteBuf——实现了这个模式，它提供了一个将多个缓冲区表示为单个合并缓冲区的虚拟表示。

注意：CompositeByteBuf 中的 ByteBuf 实例可能同时包含直接内存分配和非直接内存分配。如果其中只有一个实例，那么对 CompositeByteBuf 上的 hasArray()方法的调用将返回该组件上的 hasArray()方法的值；否则它将返回 false

举个粒子：
假设我们有一条由头部和主体两部分组成并通过 HTTP 协议传输的消息。且这两部分分别由应用程序的不同模块产生，然后在消息被发送的时候组装。该应用程序可以选择为多个消息重用相同的消息主体。当这种情况发生时，对于每个消息都将会创建一个新的头部。

如果我们不想为每个消息都重新分配这两个缓冲区，那么使用 CompositeByteBuf 是一个完美的选择。

如图为持有一个头部和主体的 CompositeByteBuf：

JDK 的 ByteBuffer针对我们这个需求实现代码会像下面这样：

// 使用数组保存消息部分ByteBuffer[] message = new ByteBuffer[] { header, body };// 创建一个新的ByteBuffer，并使用copy合并头和正文ByteBuffer message2 =ByteBuffer.allocate(header.remaining() + body.remaining());message2.put(header);message2.put(body);message2.flip();

使用 CompositeByteBuf 的复合缓冲区模式则是这样：

CompositeByteBuf messageBuf = Unpooled.compositeBuffer();ByteBuf headerBuf = ...; // 可以是堆缓冲或直接缓冲ByteBuf bodyBuf = ...; // 可以是堆缓冲或直接缓冲//将 ByteBuf 实例追加到 CompositeByteBufmessageBuf.addComponents(headerBuf, bodyBuf);.....//删除位于索引位置为 0（第一个组件）的 ByteBufmessageBuf.removeComponent(0); // 移除头部//循环遍历所有的 ByteBuf 实例for (ByteBuf buf : messageBuf) {System.out.println(buf.toString());}

CompositeByteBuf 不支持访问其支撑数组，因此访问 CompositeByteBuf 中的数据类似于（访问）直接缓冲区的模式，像下面这样：

CompositeByteBuf compBuf = Unpooled.compositeBuffer();//获得可读字节数int length = compBuf.readableBytes();//分配一个具有可读字节数长度的新数组byte[] array = new byte[length];//将数组读到该数组中compBuf.getBytes(compBuf.readerIndex(), array);//调用使用handleArray(array, 0, array.length);

Netty使用了CompositeByteBuf来优化套接字的I/O操作，尽可能地消除了由JDK的缓冲区实现所导致的性能以及内存使用率的惩罚。
这种优化发生在Netty的核心代码中，因此不会被暴露出来，但是我们应该知道它所带来的影响。

四、字节级操作

ByteBuf 提供了许多超出基本读、写操作的方法用于修改它的数据。

4.1 随机访问索引

如同在普通的 Java 字节数组中一样，ByteBuf 的索引是从零开始的：第一个字节的索引是0，最后一个字节的索引总是 capacity() – 1。对存储机制的封装使得遍历 ByteBuf 的内容非常简单。

像这样的：

ByteBuf buffer = ...;for (int i = 0; i < buffer.capacity(); i++) {byte b = buffer.getByte(i);System.out.println((char)b);}

需要注意的是，使用那些需要一个索引值参数的方法（的其中）之一来访问数据既不会改变readerIndex 也不会改变 writerIndex。如果有需要，也可以通过调用 readerIndex(index)或者 writerIndex(index)来手动移动这两者。

4.2 顺序访问索引

虽然 ByteBuf 同时具有读索引和写索引，但是 JDK 的 ByteBuffer 却只有一个索引，这也就是为什么必须调用 flip()方法来在读模式和写模式之间进行切换的原因。

以下展示了ByteBuf 是如何被它的两个索引划分成 3 个区域的：

4.3 可丢弃字节

可丢弃字节的分段包含了已经被读过的字节。通过调用 discardReadBytes()方法，可以丢弃它们并回收空间。这个分段的初始大小为 0，存储在 readerIndex 中，会随着 read 操作的执行而增加（get*操作不会移动 readerIndex）。

像图中所展示的缓冲区上调用discardReadBytes()方法后的结果。可以看到，可丢弃字节分段中的空间已经变为可写的了。注意，在调用discardReadBytes()之后，对可写分段的内容并没有任何的保证 （因为只是移动了可以读取的字节以及 writerIndex，而没有对所有可写入的字节进行擦除写）。

有的同学可能会倾向于频繁地调用 discardReadBytes()方法以确保可写分段的最大化，但是请注意，这将极有可能会导致内存复制，因为可读字节（图中标记为 CONTENT 的部分）必须被移动到缓冲区的开始位置。建议只在有真正需要的时候才这样做，例如，当内存非常宝贵的时候。

4.4 可读字节

ByteBuf 的可读字节分段存储了实际数据。新分配的、包装的或者复制的缓冲区的默认的readerIndex 值为 0。任何名称以 read 或者 skip 开头的操作都将检索或者跳过位于当前readerIndex 的数据，并且将它增加已读字节数。

如果被调用的方法需要一个 ByteBuf 参数作为写入的目标，并且没有指定目标索引参数，那么该目标缓冲区的 writerIndex 也将被增加。例如：

readBytes(ByteBuf dest);

如果尝试在缓冲区的可读字节数已经耗尽时从中读取数据，那么将会引发一个 IndexOutOfBoundsException。

下面是一个读取所有数据的代码示例：

ByteBuf buffer = ...;while (buffer.isReadable()) {System.out.println(buffer.readByte());}

4.5 可写字节

可写字节分段是指一个拥有未定义内容的、写入就绪的内存区域。新分配的缓冲区的writerIndex 的默认值为 0。任何名称以 write 开头的操作都将从当前的 writerIndex 处开始写数据，并将它增加已经写入的字节数。

如果写操作的目标也是 ByteBuf，并且没有指定源索引的值，则源缓冲区的 readerIndex 也同样会被增加相同的大小。这个调用如下所示：

writeBytes(ByteBuf dest);

如果尝试往目标写入超过目标容量的数据，将会引发一个IndexOutOfBoundException（PS：在往 ByteBuf 中写入数据时，其将首先确保目标 ByteBuf 具有足够的可写入空间来容纳当前要写入的数据，如果没有，则将检查当前的写索引以及最大容量是否可以在扩展后容纳该数据，可以则会分配并调整容量，否则就会抛出该异常。）

下面代码是一个用随机整数值填充缓冲区，直到它空间不足为止的例子。writeableBytes()方法在这里被用来确定该缓冲区中是否还有足够的空间。

// 用随机整数填充缓冲区的可写字节。ByteBuf buffer = ...;while (buffer.writableBytes() >= 4) {buffer.writeInt(random.nextInt());}

4.6 索引管理

JDK 的 InputStream 定义了 mark(int readlimit)和 reset()方法，这些方法分别被用来将流中的当前位置标记为指定的值，以及将流重置到该位置。（说白了就是给指定位置写值，比如有四个位置可以指定第一个位置写 “作”，第二个位置写“者”，第三个位置写“很”，第四个位置写“帅”）。

也可以通过调用 readerIndex(int)或者 writerIndex(int)来将索引移动到指定位置。试图将任何一个索引设置到一个无效的位置都将导致一个 IndexOutOfBoundsException。

可以通过调用 clear()方法来将 readerIndex 和 writerIndex 都设置为 0。注意，这并不会清除内存中的内容。

clear()未被调用：

clear()调用之后：

调用 clear()比调用 discardReadBytes()轻量得多，因为它将只是重置索引而不会复制任何的内存。

4.7 查找操作

在ByteBuf中有多种可以用来确定指定值的索引的方法。最简单的是使用indexOf()方法。
较复杂的查找可以通过那些需要一个io.netty.util.ByteProcessor，这个接口只定义了一个方法：

boolean process(byte value)

ByteProcessor针对一些常见的值定义了许多便利的方法。假设你的应用程序需要和所谓的包含有以NULL结尾的内容的Flash套接字作为参数的方法达成。可以调用它将检查输入值是否是正在查找的值。

forEachByte(ByteBufProcessor.FIND_NUL)

展示一个查找回车符（\r）的例子：

ByteBuf buffer = ...;int index = buffer.forEachByte(ByteBufProcessor.FIND_CR);

4.8 派生缓冲区

派生缓冲区为 ByteBuf 提供了以专门的方式来呈现其内容的视图。这类视图是通过以下方法被创建的：

• duplicate()；
• slice()；
• slice(int, int)；
• Unpooled.unmodifiableBuffer(…)；
• order(ByteOrder)；
• readSlice(int)。

每个这些方法都将返回一个新的 ByteBuf 实例，它具有自己的读索引、写索引和标记索引。其内部存储和 JDK 的 ByteBuffer 一样也是共享的。这使得派生缓冲区的创建成本是很低廉的，但是这也意味着，如果你修改了它的内容，也同时修改了其对应的源实例，所以要小心。

ByteBuf 复制：

如果需要一个现有缓冲区的真实副本，请使用 copy()或者 copy(int, int)方法。不同于派生缓冲区，由这个调用所返回的 ByteBuf 拥有独立的数据副本。

看段代码：

1、使用 slice(int, int)方法来操作 ByteBuf 的一个分段

Charset utf8 = Charset.forName("UTF-8");//创建一个用于保存给定字符串的字节的 ByteBufByteBuf buf = Unpooled.copiedBuffer("Netty in Action rocks!", utf8);//创建该 ByteBuf 从索引 0 开始到索引 15结束的一个新切片ByteBuf sliced = buf.slice(0, 15);//将打印“Netty in Action”System.out.println(sliced.toString(utf8));//更新索引 0 处的字节buf.setByte(0, (byte)'J');//将会成功，因为数据是共享的，对其中一个所做的更改对另外一个也是可见的assert buf.getByte(0) == sliced.getByte(0);

2、利用副本切片来操作 ByteBuf 的一个分段

Charset utf8 = Charset.forName("UTF-8");//创建 ByteBuf 以保存所提供的字符串的字节ByteBuf buf = Unpooled.copiedBuffer("Netty in Action rocks!", utf8);//创建该 ByteBuf 从索引 0 开始到索引 15结束的分段的副本ByteBuf copy = buf.copy(0, 15);//将打印“Netty in Action”System.out.println(copy.toString(utf8));////更新索引 0 处的字节buf.setByte(0, (byte) 'J');//将会失败，因为数据不是共享的assert buf.getByte(0) != copy.getByte(0);

4.9 读/写操作

正如我们所提到过的，有两种类别的读/写操作：

• get()和 set()操作，从给定的索引开始，并且保持索引不变；
• read()和 write()操作，从给定的索引开始，并且会根据已经访问过的字节数对索引进行调整。

下面是部分常用的get()方法：

下面是部分常用的set()方法：

get()和 set()方法的用法大家可以参照下面的例子：

Charset utf8 = Charset.forName("UTF-8");//创建一个新的 ByteBuf以保存给定字符串的字节ByteBuf buf = Unpooled.copiedBuffer("Netty in Action rocks!", utf8);//打印第一个字符'N'System.out.println((char)buf.getByte(0));//存储当前的 readerIndex和 writerIndexint readerIndex = buf.readerIndex();int writerIndex = buf.writerIndex();//将索引 0 处的字节更新为字符'B'buf.setByte(0, (byte)'B');////打印第一个字符'B'System.out.println((char)buf.getByte(0));//将会成功，因为这些操作并不会修改相应的索引assert readerIndex == buf.readerIndex();assert writerIndex == buf.writerIndex();

然后我们看一下read()操作，其作用于当前的 readerIndex 或 writerIndex。这些方法将用于从 ByteBuf 中读取数据，如同它是一个流。

常用的read()方法：

常见的writer()方法(PS:几乎每个 read()方法都有对应的 write()方法，用于将数据追加到 ByteBuf 中。注意：下列所列出的这些方法的参数是需要写入的值，而不是索引值。)

ByteBuf 上的 read()和 write()操作大家可以参照下列的代码：

Charset utf8 = Charset.forName("UTF-8");//创建一个新的ByteBuf 以保存给定字符串的字节ByteBuf buf = Unpooled.copiedBuffer("Netty in Action rocks!", utf8);//打印第一个字符'N'System.out.println((char)buf.readByte());//存储当前的存储当前的 readerIndexint readerIndex = buf.readerIndex();////存储当前的存储当前的 writerIndexint writerIndex = buf.writerIndex();//将字符'?'追加到缓冲区buf.writeByte((byte)'?');assert readerIndex == buf.readerIndex();//将会成功，因为 writeByte()方法移动了 writerIndexassert writerIndex != buf.writerIndex();

4.10 更多的操作

ByteBuf 还为我们提供了其他有用操作：

五、ByteBufHolder 接口

为了处理我们除了数据负载之外，还需要存储各种属性值的需求，Netty 提供了 ByteBufHolder。

ByteBufHolder 也为 Netty 的高级特性提供了支持，如缓冲区池化，其中可以从池中借用 ByteBuf，并且在需要时自动释放。ByteBufHolder 只有几种用于访问底层数据和引用计数的方法。

ByteBufHolder 的常见操作如下：

如果想要实现一个将其有效负载存储在 ByteBuf 中的消息对象，那么 ByteBufHolder 将是个不错的选择。

六、ByteBuf 分配6.1 按需分配：ByteBufAllocator 接口

为了降低分配和释放内存的开销，Netty 通过 interface ByteBufAllocator 实现了（ByteBuf 的）池化，它可以用来分配我们所描述过的任意类型的 ByteBuf 实例。

使用池化是特定于应用程序的决定，其并不会以任何方式改变 ByteBuf API。

下面列出了 ByteBufAllocator 常见的一些操作：

可以通过 Channel（每个都可以有一个不同的 ByteBufAllocator 实例）或者绑定到ChannelHandler 的 ChannelHandlerContext 获取一个到 ByteBufAllocator 的引用。

大家可以参考下面的代码：

Channel channel = ...;//从 Channel 获取一个到ByteBufAllocator 的引用ByteBufAllocator allocator = channel.alloc();....ChannelHandlerContext ctx = ...;//从 ChannelHandlerContext 获取一个到 ByteBufAllocator 的引用ByteBufAllocator allocator2 = ctx.alloc();...

Netty提供了两种ByteBufAllocator的实现：PooledByteBufAllocator和UnpooledByteBufAllocator。

前者池化了ByteBuf的实例以提高性能并最大限度地减少内存碎片。

后者的实现不池化ByteBuf实例，并且在每次它被调用时都会返回一个新的实例。

虽然Netty默认使用了PooledByteBufAllocator，但这可以很容易地通过ChannelConfig API或者在引导你的应用程序时指定一个不同的分配器来更改。

6.2 Unpooled 缓冲区

可能某些情况下，未能获取一个到 ByteBufAllocator 的引用。对于这种情况，Netty 提供了一个简单的称为 Unpooled 的工具类，它提供了静态的辅助方法来创建未池化的 ByteBuf实例。

下面是写常用的Unpooled方法：

Unpooled 类还使得 ByteBuf 同样可用于那些并不需要 Netty 的其他组件的非网络项目，使得其能得益于高性能的可扩展的缓冲区 API。

6.3 ByteBufUtil 类

ByteBufUtil 提供了用于操作 ByteBuf 的静态的辅助方法。因为这个 API 是通用的，并且和池化无关，所以这些方法已然在分配类的外部实现。

这些静态方法中最有价值方法：

• hexdump()方法：它以十六进制的表示形式打印ByteBuf 的内容。这在各种情况下都很有用，例如，出于调试的目的记录 ByteBuf 的内容。十六进制的表示通常会提供一个比字节值的直接表示形式更加有用的日志条目，此外，十六进制的版本还可以很容易地转换回实际的字节表示。
• boolean equals(ByteBuf, ByteBuf)：它被用来判断两个 ByteBuf实例的相等性。如果你实现自己的 ByteBuf 子类，你可能会发现 ByteBufUtil 的其他有用方法。

七、引用计数

引用计数是一种通过在某个对象所持有的资源不再被其他对象引用时释放该对象所持有的资源来优化内存使用和性能的技术。

引用计数背后的想法并不是特别的复杂；它主要涉及跟踪到某个特定对象的活动引用的数量。一个 ReferenceCounted 实现的实例将通常以活动的引用计数为 1 作为开始。只要引用计数大于 0，就能保证对象不会被释放。当活动引用的数量减少到 0 时，该实例就会被释放。

PS，虽然释放的确切语义可能是特定于实现的，但是至少已经释放的对象应该不可再用了。

引用计数对于池化实现（如 PooledByteBufAllocator）来说是至关重要的，它降低了内存分配的开销。

引用计数演示代码：

Channel channel = ...;//从 Channel 获取ByteBufAllocatorByteBufAllocator allocator = channel.alloc();....//从 ByteBufAllocator分配一个 ByteBufByteBuf buffer = allocator.directBuffer();//检查引用计数是否为预期的 1assert buffer.refCnt() == 1;...

释放引用计数：

ByteBuf buffer = ...;//减少到该对象的活动引用。当减少到 0 时，该对象被释放，并且该方法返回 trueboolean released = buffer.release();..

试图访问一个已经被释放的引用计数的对象，将会导致一个 IllegalReferenceCountException。注意，一个特定的（ReferenceCounted 的实现）类，可以用它自己的独特方式来定义它的引用计数规则。例如，我们可以设想一个类，其 release()方法的实现总是将引用计数设为
零，而不用关心它的当前值，从而一次性地使所有的活动引用都失效