本文旨在解决Java应用中，从JNI获取的Direc++t Buffer在上传至S3时，因传统方法需额外复制到JVM堆内存而导致的性能瓶颈。我们将介绍一种高效的解决方案，通过自定义ByteSource和InputStream封装ByteBuffer，实现数据流的直接读取和上传，从而避免不必要的内存拷贝，提升大数据处理效率。
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中提供数据流，而无需将其复制到堆内存。
阿贝智能
阿贝智能是基于AI技术辅助创作儿童绘本、睡前故事和有声书的平台，助你创意实现、梦想成真。
17 查看详情
核心思想是创建一个ByteBufferByteSource类，它包含一个ByteBuffer实例，并能够返回一个InputStream。这个InputStream将直接从ByteBuffer中读取数据。
3.1 ByteBufferByteSource 和 ByteBufferInputStream 实现
以下是实现自定义ByteSource和其内部InputStream的代码：
import com.google.common.base.Preconditions; // 或者使用 java.util.Objects.requireNonNullimport 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 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 提供的 ByteSourcePayloadimport 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的状态管理和异常处理，以确保系统的稳定性和可靠性。
以上就是Java JNI Direct Buffer免拷贝上传S3的实现教程的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/326975.html
微信扫一扫 
支付宝扫一扫 
