Java JNI Direct Buffer免拷贝上传S3的实现教程

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;
}

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 关联了某些需要释放的资源（例如文件句柄），
            // 则需要在更上层进行管理。
        }
    }
}

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;
}