数据缓冲区和编解码器

Java NIO 提供了 ByteBuffer,但许多库在此之上构建了自己的字节缓冲区 API,特别是对于网络操作,其中重用缓冲区和/或使用直接缓冲区有助于提高性能。例如,Netty 有 ByteBuf 层次结构,Jetty 使用带有回调函数的池化字节缓冲区来释放,等等。spring-core 模块提供了一组抽象来处理各种字节缓冲区 API,如下所示:

DataBufferFactory

DataBufferFactory 用于以以下两种方式之一创建数据缓冲区:

  1. 分配一个新的数据缓冲区,如果容量已知,可以选择预先指定容量,这更高效,尽管 DataBuffer 的实现可以按需增长和缩小。

  2. 包装一个现有的 byte[]java.nio.ByteBuffer,它用 DataBuffer 实现装饰给定的数据,并且不涉及分配。

请注意,WebFlux 应用程序不直接创建 DataBufferFactory,而是通过 ServerHttpResponse 或客户端的 ClientHttpRequest 访问它。工厂的类型取决于底层客户端或服务器,例如,Reactor Netty 的 NettyDataBufferFactory,其他情况的 DefaultDataBufferFactory

DataBuffer

DataBuffer 接口提供与 java.nio.ByteBuffer 类似的操作,但也带来了一些额外的优点,其中一些受到了 Netty ByteBuf 的启发。以下是部分优点列表:

  • 读写具有独立的 position,即无需调用 flip() 来在读写之间切换。

  • 容量按需扩展,如同 java.lang.StringBuilder

  • 通过 PooledDataBuffer 实现的池化缓冲区和引用计数。

  • 将缓冲区视为 java.nio.ByteBufferInputStreamOutputStream

  • 确定给定字节的索引或最后一个索引。

PooledDataBuffer

ByteBuffer 的 Javadoc 中所述,字节缓冲区可以是直接的或非直接的。直接缓冲区可以驻留在 Java 堆之外,这消除了本地 I/O 操作的复制需求。这使得直接缓冲区对于通过套接字接收和发送数据特别有用,但它们的创建和释放成本也更高,这导致了池化缓冲区的想法。

PooledDataBufferDataBuffer 的一个扩展,有助于引用计数,这对于字节缓冲区池化至关重要。它是如何工作的?当分配一个 PooledDataBuffer 时,引用计数为 1。调用 retain() 会增加计数,而调用 release() 会减少计数。只要计数大于 0,就保证缓冲区不会被释放。当计数减少到 0 时,池化缓冲区可以被释放,这实际上可能意味着为缓冲区保留的内存会返回到内存池。

请注意,在大多数情况下,最好使用 DataBufferUtils 中的便利方法,它们仅当 DataBufferPooledDataBuffer 的实例时才对其应用释放或保留操作,而不是直接操作 PooledDataBuffer

DataBufferUtils

DataBufferUtils 提供了许多实用方法来操作数据缓冲区:

  • 将数据缓冲区流合并到单个缓冲区中,可能进行零拷贝,例如,如果底层字节缓冲区 API 支持,则通过复合缓冲区实现。

  • InputStream 或 NIO Channel 转换为 Flux<DataBuffer>,反之,将 Publisher<DataBuffer> 转换为 OutputStream 或 NIO Channel

  • 如果缓冲区是 PooledDataBuffer 的实例,则提供释放或保留 DataBuffer 的方法。

  • 从字节流中跳过或获取直到达到特定字节数。

编解码器 (Codecs)

org.springframework.core.codec 包提供以下策略接口:

  • Encoder 用于将 Publisher<T> 编码为数据缓冲区流。

  • Decoder 用于将 Publisher<DataBuffer> 解码为更高级别的对象流。

spring-core 模块提供了 byte[]ByteBufferDataBufferResourceString 的编码器和解码器实现。spring-web 模块增加了 Jackson JSON、Jackson Smile、JAXB2、Protocol Buffers 和其他编码器和解码器。请参阅 WebFlux 部分的 编解码器

使用 DataBuffer

处理数据缓冲区时,必须特别小心以确保缓冲区被释放,因为它们可能是 池化的。我们将使用编解码器来说明其工作原理,但这些概念更普遍适用。让我们看看编解码器在内部必须如何管理数据缓冲区。

一个 Decoder 是在创建更高级别对象之前最后读取输入数据缓冲区的,因此它必须按如下方式释放它们:

  1. 如果 Decoder 只是简单地读取每个输入缓冲区并准备立即释放它,它可以通过 DataBufferUtils.release(dataBuffer) 来完成。

  2. 如果 Decoder 使用 FluxMono 运算符,例如 flatMapreduce 等在内部预取和缓存数据项,或者使用 filterskip 等运算符遗漏项,那么必须将 doOnDiscard(DataBuffer.class, DataBufferUtils::release) 添加到组合链中,以确保此类缓冲区在被丢弃之前被释放,这可能也是错误或取消信号的结果。

  3. 如果 Decoder 以任何其他方式持有一个或多个数据缓冲区,它必须确保在完全读取时,或在缓存的数据缓冲区被读取和释放之前发生错误或取消信号时,它们会被释放。

请注意,DataBufferUtils#join 提供了一种安全高效的方法,可以将数据缓冲区流聚合到单个数据缓冲区中。同样,skipUntilByteCounttakeUntilByteCount 是解码器可以使用的其他安全方法。

一个 Encoder 分配数据缓冲区供其他组件读取(并释放)。因此 Encoder 没有太多事情要做。然而,如果填充缓冲区数据时发生序列化错误,Encoder 必须注意释放数据缓冲区。例如:

  • Java

  • Kotlin

DataBuffer buffer = factory.allocateBuffer();
boolean release = true;
try {
	// serialize and populate buffer..
	release = false;
}
finally {
	if (release) {
		DataBufferUtils.release(buffer);
	}
}
return buffer;
val buffer = factory.allocateBuffer()
var release = true
try {
	// serialize and populate buffer..
	release = false
} finally {
	if (release) {
		DataBufferUtils.release(buffer)
	}
}
return buffer

Encoder 的消费者负责释放它接收到的数据缓冲区。在 WebFlux 应用程序中,Encoder 的输出用于写入 HTTP 服务器响应或客户端 HTTP 请求,在这种情况下,释放数据缓冲区是写入服务器响应或客户端请求的代码的责任。

请注意,在 Netty 上运行时,有针对 缓冲区泄漏故障排除 的调试选项。

© . This site is unofficial and not affiliated with VMware.