1. 引言与问题背景
在高性能计算场景中，java应用常通过jni（java native interface）与c/c++代码进行交互，特别是涉及大块内存数据时，通常会利用direct buffer（直接缓冲区）来避免jvm堆内存与原生内存之间的数据拷贝。例如，当c++进程将数据写入共享内存，java侧通过jni将这块共享内存映射为java.nio.bytebuffer的direct buffer时，这种零拷贝的优势尤为明显。
然而，当需要将Direct Buffer中的数据上传至云存储服务（如Amazon S3）时，传统的上传方式往往会破坏这种零拷贝的优化。许多S3客户端库，如jclouds，其默认的Payload实现（例如ByteArrayPayload）通常要求提供byte[]数组，这意味着Direct Buffer中的数据必须先复制到JVM堆上的byte[]数组中。对于50MB甚至更大的数据块，这种额外的内存拷贝不仅消耗CPU时间，还会增加GC压力和内存占用，抵消了Direct Buffer带来的性能优势。
本文将探讨如何避免这种不必要的内存拷贝，实现Direct Buffer到S3的直接上传。
2. 问题分析与传统方法局限
考虑以下场景：
- C++进程写入50MB数据到共享内存。
- Java JNI函数将共享内存地址映射为ByteBuffer：
JNIEXPORT jobject JNICALL Java_service_SharedMemoryJNIService_getDirectByteBuffer (JNIEnv *env, jclass jobject, jlong buf_addr, jint buf_len){ return env->NewDirectByteBuffer((void *)buf_addr, buf_len); } - Java侧需要将此ByteBuffer中的数据上传到S3。
传统的jclouds上传代码可能如下所示：
立即学习“Java免费学习笔记（深入）”；
public String uploadByteBuffer(String container, String objectKey, ByteBuffer bb) {
// ... 获取 BlobStoreContext 和 BlobStore
BlobStoreContext context = getBlobStoreContext();
BlobStore blobStore = context.getBlobStore();
// 问题所在：将 Direct Buffer 数据复制到 JVM 堆内存
byte[] buf = new byte[bb.capacity()];
bb.get(buf); // 发生内存拷贝
ByteArrayPayload payload = new ByteArrayPayload(buf);
Blob blob = blobStore.blobBuilder(objectKey)
.payload(payload)
.contentLength(bb.capacity())
.build();
blobStore.putBlob(container, blob);
return objectKey;
}bb.get(buf)这一行是性能瓶颈的关键，它强制将Direct Buffer中的数据复制到byte[]数组，导致了不必要的内存开销和延迟。
3. 解决方案：自定义ByteSource实现直接上传
为了避免内存拷贝，我们需要利用S3客户端库（如jclouds）提供的高级Payload机制。jclouds的BlobBuilder.payload()方法除了接受byte[]或String外，还可以接受ByteSource对象。ByteSource是一个抽象类，它定义了如何提供一个InputStream来读取数据。通过实现一个自定义的ByteSource，我们可以直接从ByteBuffer中提供数据流，而无需将其复制到堆内存。
Productivity Google Workspace
下载
Access Gmail, Calendar, Drive, Docs, and Sheets via the gws CLI or Python.
核心思想是创建一个ByteBufferByteSource类，它包含一个ByteBuffer实例，并能够返回一个InputStream。这个InputStream将直接从ByteBuffer中读取数据。
3.1 ByteBufferByteSource 和 ByteBufferInputStream 实现
以下是实现自定义ByteSource和其内部InputStream的代码：
import com.google.common.base.Preconditions; // 或者使用 java.util.Objects.requireNonNull
import com.google.common.io.ByteSource;
import java.io.IOException;
import java.io.InputStream;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
/**
* 一个 ByteSource 实现，允许直接从 ByteBuffer 中读取数据流，
* 避免将 ByteBuffer 的内容复制到 JVM 堆内存。
*/
public class ByteBufferByteSource extends ByteSource {
private final ByteBuffer buffer;
/**
* 构造函数。
* @param buffer 要从中读取数据的 ByteBuffer。
* 此 ByteBuffer 应该是可读的，且其 position 和 limit 应设置正确。
*/
public ByteBufferByteSource(ByteBuffer buffer) {
// 确保传入的 ByteBuffer 不为空
this.buffer = Preconditions.checkNotNull(buffer, "ByteBuffer cannot be null");
// 如果需要，可以考虑在这里对 buffer 进行 slice() 或 duplicate()
// 以确保每次 openStream() 都从缓冲区的起始位置开始读取，
// 并且不影响原始 buffer 的状态。
// 例如：this.buffer = buffer.asReadOnlyBuffer().slice();
// 在本例中，我们假设传入的 buffer 已经准备好被读取，并且其状态可以在 openStream() 中管理。
}
/**
* 返回一个 InputStream，用于从 ByteBuffer 中读取数据。
* 每次调用都会返回一个新的 InputStream 实例，该实例从 ByteBuffer 的当前位置开始读取。
* 如果希望每次都从头开始读取，请确保在构造 ByteBufferByteSource 时传入一个副本或使用 slice()。
*/
@Override
public InputStream openStream() {
// 创建一个新的 ByteBuffer 视图，以确保每次 openStream() 都能独立操作，
// 不会影响原始 ByteBuffer 的 position 和 limit，也不会影响其他 InputStream 实例。
// duplicate() 创建一个与原 ByteBuffer 共享内容但拥有独立 position, limit, mark 的新 ByteBuffer。
return new ByteBufferInputStream(this.buffer.duplicate());
}
/**
* 一个私有的 InputStream 实现，直接从 ByteBuffer 中读取字节。
* 避免了将数据复制到中间 byte[] 数组。
*/
private static final class ByteBufferInputStream extends InputStream {
private final ByteBuffer buffer;
private boolean closed = false;
ByteBufferInputStream(ByteBuffer buffer) {
this.buffer = buffer;
}
@Override
public synchronized int read() throws IOException {
if (closed) {
throw new IOException("Stream already closed");
}
try {
// 直接从 ByteBuffer 中获取一个字节
return buffer.get() & 0xFF; // 返回无符号字节值
} catch (BufferUnderflowException bue) {
// 当 ByteBuffer 中没有更多字节可读时，表示流已结束
return -1;
}
}
/**
* 重写 read(byte[], int, int) 方法以提高效率。
* 默认的 InputStream.read(byte[], int, int) 是通过循环调用 read() 实现的，效率低下。
* 此方法直接利用 ByteBuffer 的批量读取能力。
*/
@Override
public synchronized int read(byte[] b, int off, int len) throws IOException {
if (closed) {
throw new IOException("Stream already closed");
}
if (b == null) {
throw new NullPointerException();
} else if (off < 0 || len < 0 || len > b.length - off) {
throw new IndexOutOfBoundsException();
} else if (len == 0) {
return 0;
}
// 计算 ByteBuffer 中剩余可读字节数
int bytesRemaining = buffer.remaining();
if (bytesRemaining == 0) {
return -1; // 没有更多数据可读
}
// 实际可以读取的字节数是 len 和 bytesRemaining 中的较小值
int bytesToRead = Math.min(len, bytesRemaining);
// 将数据从 ByteBuffer 直接批量复制到目标 byte 数组
buffer.get(b, off, bytesToRead);
return bytesToRead;
}
@Override
public int available() throws IOException {
if (closed) {
throw new IOException("Stream already closed");
}
return buffer.remaining(); // 返回 ByteBuffer 中剩余可读字节数
}
@Override
public void close() throws IOException {
super.close();
closed = true;
// 注意：这里没有对 underlying ByteBuffer 进行 close 操作，
// 因为 ByteBuffer 通常不是一个需要关闭的资源。
// 如果 ByteBuffer 关联了某些需要释放的资源（例如文件句柄），
// 则需要在更上层进行管理。
}
}
}代码解析：
-
ByteBufferByteSource:
- 构造函数接收一个ByteBuffer。为了确保每次调用openStream()都能从缓冲区的一个独立视图开始读取，我们建议在构造ByteBufferByteSource时，对传入的ByteBuffer调用asReadOnlyBuffer().slice()或duplicate()。在openStream()中调用duplicate()是更常见的做法，它创建了一个与原ByteBuffer共享内容但拥有独立position、limit和mark的新ByteBuffer。
- openStream()方法返回一个ByteBufferInputStream实例，该实例将从ByteBuffer中读取数据。
-
ByteBufferInputStream:
- 这是一个私有静态内部类，继承自InputStream。
- read()方法直接调用buffer.get()来获取单个字节。当ByteBuffer中没有更多数据时，会抛出BufferUnderflowException，此时read()返回-1，表示流的结束。
- 效率优化：read(byte[] b, int off, int len)方法被重写。默认的InputStream实现会循环调用read()逐字节读取，效率非常低。通过重写此方法，我们直接利用ByteBuffer.get(byte[], int, int)进行批量读取，大大提高了数据传输效率。
- available()方法返回buffer.remaining()，表示当前ByteBuffer中还有多少字节可读。
- close()方法设置一个closed标志，以防止在流关闭后继续操作。
4. 集成与使用
有了ByteBufferByteSource，现在可以将S3上传代码修改为：
import com.google.common.io.ByteSource;
import org.jclouds.blobstore.BlobStore;
import org.jclouds.blobstore.BlobStoreContext;
import org.jclouds.blobstore.domain.Blob;
import org.jclouds.io.Payload;
import org.jclouds.io.payloads.ByteSourcePayload; // jclouds 提供的 ByteSourcePayload
import java.nio.ByteBuffer;
public String uploadByteBufferDirectly(String container, String objectKey, ByteBuffer bb) {
// ... 获取 BlobStoreContext 和 BlobStore
BlobStoreContext context = getBlobStoreContext();
BlobStore blobStore = context.getBlobStore();
// 创建自定义的 ByteSource
ByteSource byteSource = new ByteBufferByteSource(bb);
// 使用 jclouds 提供的 ByteSourcePayload 封装 ByteSource
// 或者直接将 ByteSource 传给 payload() 方法（如果 BlobBuilder 支持）
Payload payload = new ByteSourcePayload(byteSource);
payload.setContentLength((long) bb.capacity()); // 确保设置正确的长度
Blob blob = blobStore.blobBuilder(objectKey)
.payload(payload)
.contentLength((long) bb.capacity()) // 再次确认长度
.build();
blobStore.putBlob(container, blob);
return objectKey;
}注意事项：
- ByteBuffer状态管理：确保传入ByteBufferByteSource的ByteBuffer处于可读状态（即position和limit设置正确）。如果ByteBuffer在创建ByteBufferByteSource后被其他线程或操作修改了position或limit，可能会导致数据读取不完整或错误。在ByteBufferByteSource的openStream()方法中使用buffer.duplicate()是一个很好的实践，它创建了一个独立的ByteBuffer视图，避免了多线程或多次读取时对原始ByteBuffer状态的干扰。
- jclouds版本：确保您的jclouds库版本支持ByteSourcePayload或直接接受ByteSource作为payload参数。
- 错误处理：在实际应用中，需要添加更健壮的错误处理机制，例如对IOException进行捕获和日志记录。
- ContentLength：务必设置正确的ContentLength。ByteBuffer.capacity()表示缓冲区的总容量，通常也是需要上传的数据长度。
5. 总结
通过实现自定义的ByteBufferByteSource和ByteBufferInputStream，我们成功地解决了从JNI获取的Direct Buffer在上传至S3时需要额外内存拷贝的问题。这种方法使得数据可以直接从原生内存流式传输到S3，避免了JVM堆内存的中间缓冲，从而显著降低了内存占用、减少了GC压力，并提升了大数据上传的整体性能。这对于处理大规模数据或追求极致性能的Java应用来说，是一个重要的优化手段。在实际应用中，请务必注意ByteBuffer的状态管理和异常处理，以确保系统的稳定性和可靠性。
