如何使用java開發工具包異步上傳多部分到AmazonS3

在我的 java 應用程式中，我需要將資料寫入 S3，我事先不知道大小，而且大小通常很大，因此在 AWS S3 檔案中建議我使用的是Using the Java AWS SDKs (low-level-level API)將資料寫入 s3 存盤桶。

在我的應用程式中，我提供S3BufferedOutputStream了一個實作OutputStream，應用程式中的其他類可以使用此流寫入 s3 存盤桶。

我將資料存盤在緩沖區和回圈中，一旦資料大于存盤桶大小，我將緩沖區中的資料作為單個上傳UploadPartRequest 這是 S3BufferedOutputStream 的 write 方法的實作

@Override
public void write(byte[] b, int off, int len) throws IOException {
    this.assertOpen();
    int o = off, l = len;
    int size;
    while (l > (size = this.buf.length - position)) {
        System.arraycopy(b, o, this.buf, this.position, size);
        this.position  = size;
        flushBufferAndRewind();
        o  = size;
        l -= size;
    }
    System.arraycopy(b, o, this.buf, this.position, l);
    this.position  = l;
}

整個實作類似于：code repo

我這里的問題是每個 UploadPartRequest 都是同步完成的，所以我們必須等待一個部分上傳才能上傳下一部分。而且因為我使用的是 AWS S3 低級 API，我無法從TransferManager提供的并行上傳中受益

有沒有辦法使用低級SDK實作并行上傳？或者可以進行一些代碼更改以異步操作而不會破壞上傳的資料并保持資料的順序？

uj5u.com熱心網友回復：

這是我擁有的一個類的一些示例代碼。它將零件提交給 anExecutorService并保留回傳的Future. 這是為 v1 Java SDK 撰寫的；如果您使用的是 v2 SDK，您可以使用異步客戶端而不是顯式執行緒池：

// WARNING: data must not be updated by caller; make a defensive copy if needed
public synchronized void uploadPart(byte[] data, boolean isLastPart)
{
    partNumber  ;
    logger.debug("submitting part {} for s3://{}/{}", partNumber, bucket, key);

    final UploadPartRequest request = new UploadPartRequest()
                                      .withBucketName(bucket)
                                      .withKey(key)
                                      .withUploadId(uploadId)
                                      .withPartNumber(partNumber)
                                      .withPartSize(data.length)
                                      .withInputStream(new ByteArrayInputStream(data))
                                      .withLastPart(isLastPart);

    futures.add(
        executor.submit(new Callable<PartETag>()
        {
            @Override
            public PartETag call() throws Exception
            {
                int localPartNumber = request.getPartNumber();
                logger.debug("uploading part {} for s3://{}/{}", localPartNumber, bucket, key);
                UploadPartResult response = client.uploadPart(request);
                String etag = response.getETag();
                logger.debug("uploaded part {} for s3://{}/{}; etag is {}", localPartNumber, bucket, key, etag);
                return new PartETag(localPartNumber, etag);
            }
        }));
}

注意：此方法是synchronized為了確保零件不會亂序提交。

提交所有部分后，您可以使用此方法等待它們完成，然后完成上傳：

public void complete()
{
    logger.debug("waiting for upload tasks of s3://{}/{}", bucket, key);
    List<PartETag> partTags = new ArrayList<>();
    for (Future<PartETag> future : futures)
    {
        try
        {
            partTags.add(future.get());
        }
        catch (Exception e)
        {
            throw new RuntimeException(String.format("failed to complete upload task for s3://%s/%s"), e);
        }
    }

    logger.debug("completing multi-part upload for s3://{}/{}", bucket, key);
    CompleteMultipartUploadRequest request = new CompleteMultipartUploadRequest()
                                              .withBucketName(bucket)
                                              .withKey(key)
                                              .withUploadId(uploadId)
                                              .withPartETags(partTags);
    client.completeMultipartUpload(request);
    logger.debug("completed multi-part upload for s3://{}/{}", bucket, key);
}

您還需要一種abort()取消未完成部分并中止上傳的方法。這個，以及課程的其他部分，留給讀者作為練習。

標籤：

上一篇：AWSCreateDeviceFleet操作失敗，因為“賬戶ID沒有存盤桶的所有權”

下一篇：如何從以前的提交中恢復檔案？