Spring Docs

主题

数据缓冲区 (Data Buffers) 与编解码器 (Codecs) ​

Java NIO 提供了 ByteBuffer，但在其之上，许多库构建了自己的字节缓冲区 API，特别是对于需要复用缓冲区和/或使用直接缓冲区（Direct Buffers）以提升性能的网络操作。例如，Netty 有其 ByteBuf 体系结构，Jetty 使用带有释放回调的池化字节缓冲区，等等。

spring-core 模块提供了一组抽象，用于处理各种字节缓冲区 API，如下所示：

DataBufferFactory ​

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

  1. 分配一个新的数据缓冲区。如果已知容量，可以预先指定，这样效率更高，尽管 DataBuffer 的实现可以按需增长或缩小。
  2. 包装现有的 byte[] 或 java.nio.ByteBuffer。这会用 DataBuffer 实现装饰给定的数据，且不涉及内存分配。

请注意，WebFlux 应用程序通常不直接创建 DataBufferFactory，而是通过客户端的 ClientHttpRequest 或服务端的 ServerHttpResponse 来获取它。工厂的具体类型取决于底层的引擎，例如 Reactor Netty 使用 NettyDataBufferFactory，其他引擎使用 DefaultDataBufferFactory。

DataBuffer ​

DataBuffer 接口提供了与 java.nio.ByteBuffer 相似的操作，但还带来了一些额外的改进，其中一些灵感源自 Netty 的 ByteBuf。以下是其部分优点：

  • 独立的读写位置：读和写拥有独立的指针，这意味着在读写切换时不需要调用 flip()。
  • 按需扩容：类似于 java.lang.StringBuilder，容量可以根据需求自动扩展。
  • 引用计数：通过 PooledDataBuffer 支持池化缓冲区和引用计数。
  • 多视图支持：可以将缓冲区视为 java.nio.ByteBuffer、InputStream 或 OutputStream。
  • 查找功能：可以确定给定字节的索引或最后索引。

池化数据缓冲区 (PooledDataBuffer) ​

正如 ByteBuffer 的 Javadoc 所述，字节缓冲区可以是直接的（Direct）或非直接的。直接缓冲区可以驻留在 Java 堆外，从而在执行原生 I/O 操作时消除数据拷贝的需要。这使得直接缓冲区在处理 Socket 发送和接收数据时特别有用，但它们的创建和释放成本也更高，因此引入了池化缓冲区的概念。

PooledDataBuffer 是 DataBuffer 的一个扩展，它有助于进行引用计数，这对于缓冲区池化至关重要。其工作机制如下： 分配 PooledDataBuffer 时，其引用计数为 1。调用 retain() 会增加计数，而调用 release() 会减少计数。只要计数大于 0，缓冲区就保证不会被释放。当计数减至 0 时，池化缓冲区可以被回收，在实践中这通常意味着其预留的内存归还到了内存池中。

请注意，与其直接操作 PooledDataBuffer，大多情况下更好的做法是使用 DataBufferUtils 中的便捷方法，这些方法仅在 DataBuffer 是 PooledDataBuffer 实例时才会应用释放或保留操作。

DataBufferUtils ​

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

  • 将数据缓冲区流合并成一个单一缓冲区，如果底层 API 支持（例如通过复合缓冲区），则可以实现零拷贝。
  • 将 InputStream 或 NIO Channel 转换为 Flux<DataBuffer>，反之亦然。
  • 如果缓冲区是 PooledDataBuffer 的实例，则执行释放（release）或保留（retain）操作。
  • 在字节流中跳过（skip）或截取（take）指定数量的字节。

编解码器 (Codecs) ​

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

  • Encoder：将 Publisher<T> 编码成数据缓冲区流。
  • Decoder：将 Publisher<DataBuffer> 解码成更高级的对象流。

spring-core 模块提供了 byte[]、ByteBuffer、DataBuffer、Resource 和 String 的编码器及解码器实现。spring-web 模块则增加了 Jackson JSON、Jackson Smile、JAXB2、Protocol Buffers 等实现。详见 WebFlux 章节中的编解码器。

使用 DataBuffer ​

在使用数据缓冲区时，必须特别小心确保缓冲区被释放，因为它们可能是池化的。我们将以编解码器为例进行说明，但这些概念具有普适性。

解码器管理逻辑 ​

Decoder 是最后一个读取输入数据缓冲区并在此基础上创建高级对象的组件，因此它必须按如下方式释放它们：

  1. 如果 Decoder 只是简单读取每个输入缓冲区并准备立即释放，可以通过 DataBufferUtils.release(dataBuffer) 完成。
  2. 如果 Decoder 使用了 flatMap、reduce 等内部会预取和缓存数据的操作符，或者使用了 filter、skip 等会丢弃项的操作符，则必须在组合链中添加 doOnDiscard(DataBuffer.class, DataBufferUtils::release)。这确保了缓冲区在被丢弃之前（包括出错或取消信号的情况）得到释放。
  3. 如果 Decoder 以任何其他方式持有缓冲区，必须确保在完整读取后，或者在读取前发生错误/取消信号时将其释放。

请注意，DataBufferUtils#join 是聚合缓冲区流的一种既安全又高效的方式。

编码器管理逻辑 ​

Encoder 分配数据缓冲区供他人读取（并释放），因此它的工作不多。但是，如果在往缓冲区填充数据时发生了序列化错误，Encoder 必须负责释放已分配的缓冲区。

java
DataBuffer buffer = factory.allocateBuffer();
boolean release = true;
try {
	// 序列化并填充缓冲区...
	release = false;
}
finally {
	if (release) {
		DataBufferUtils.release(buffer);
	}
}
return buffer;
kotlin
val buffer = factory.allocateBuffer()
var release = true
try {
	// 序列化并填充缓冲区...
	release = false
} finally {
	if (release) {
		DataBufferUtils.release(buffer)
	}
}
return buffer

Encoder 的消费者负责释放收到的数据缓冲区。在 WebFlux 应用中，Encoder 的输出通常被写往 HTTP 响应，这种情况下，释放缓冲区的责任由写入响应的底层代码承担。

提示

在 Netty 上运行时，可以使用排查缓冲区泄漏的调试选项。

补充教学 ​

1. 为什么 Spring 不直接用 java.nio.ByteBuffer？ ​

虽然 ByteBuffer 是标准，但它在高性能响应式场景下存在几个痛点：

  • 反直觉的 flip()：在 ByteBuffer 中，读写共用一个位置指针，切换极其繁琐。Spring DataBuffer 借鉴了 Netty 的思想，采用双指针设计（readerIndex 和 writerIndex），大幅降低了复杂度。
  • 容量固定：ByteBuffer 一旦分配大小固定。DataBuffer 支持动态伸缩，对于长报文处理更友好。
  • 内存池化的缺失：原生 Java 直到很晚才引入受控的内存清理，而 Spring 需要在各种容器（Netty, Jetty, Tomcat）之间提供统一的内存管理和引用计数机制。

2. 生死攸关：内存泄漏与引用计数 ​

在 WebFlux/响应式编程中，内存泄漏往往源于“忘记释放缓冲区”。

  • 谁消费，谁释放：这是 AOP 编程或底层 I/O 编程的铁律。如果你在自定义 Decoder 中把 DataBuffer 转换成了 String，那么原来的 DataBuffer 就必须被释放。
  • 零拷贝（Zero-Copy）的代价：当你使用零拷贝技术（如 Netty 的复合缓冲区）时，实际上多个逻辑缓冲区共享同一块物理内存。如果其中一个引用没有正确 release，这整块内存都无法归还物理池，后果严重。

3. 理解 doOnDiscard ​

这是响应式编程中最容易被忽视的防御性代码。

java
flux.doOnDiscard(DataBuffer.class, DataBufferUtils::release)

想象一个场景：你正在从网络流中读取 100 个内存块。突然下游抛出个异常说“我不要了”。如果没有 doOnDiscard，那已经预取的块就会随着垃圾回收一起由于计数不为 0 而变成死内存。

4. 什么时候需要操心这些？ ​

  • 普通开发者：通常不需要。只要你使用的是 Spring 内置的 WebClient 或标准的 ResponseEntity，Spring 和底层引擎（如 Netty）会自动帮你关好门。
  • 中间件/框架开发者：如果你在手写自定义的 Decoder、Encoder 或者底层的流式过滤器（WebFilter），那么理解 DataBuffer 的生命周期是你的必修课。

Based on Spring Framework.

Copyright © 2024 Spring 中文社区