Spring WebFlux上传S3时如何将Flux<DataBuffer>转为Flux<ByteBuffer>

你可以直接通过DataBuffer自带的asByteBuffer()方法完成对流中元素的映射转换，示例代码如下：

import org.springframework.core.io.buffer.DataBufferUtils;
import reactor.core.publisher.Flux;
import java.nio.ByteBuffer;

// 你的业务逻辑中已经拿到了FilePart对象
Flux<ByteBuffer> byteBufferFlux = filePart.content()
        .map(dataBuffer -> {
            // 转换为ByteBuffer视图，无额外内存拷贝
            ByteBuffer buffer = dataBuffer.asByteBuffer();
            // 释放原始DataBuffer，避免堆外内存泄漏
            DataBufferUtils.release(dataBuffer);
            return buffer;
        });

注意事项 #

• asByteBuffer()返回的是DataBuffer底层的视图，不会产生额外的内存复制开销，性能表现优异
• 如果你后续还需要对原始DataBuffer做其他处理，不要调用release方法，避免提前释放内存导致报错
• 如果需要调整输出的ByteBuffer分片大小以适配S3分片上传的阈值要求，可以在转换后追加buffer()操作自定义分片大小，示例：

  // 合并为最大1MB的分片再向下游传递
  Flux<ByteBuffer> sizedBufferFlux = byteBufferFlux
          .buffer(1024 * 1024)
          .map(buffers -> {
              ByteBuffer merged = ByteBuffer.allocate(buffers.stream().mapToInt(ByteBuffer::remaining).sum());
              buffers.forEach(merged::put);
              merged.flip();
              return merged;
          });
  

内容的提问来源于stack exchange，提问作者Emdadul Sawon