我需要使用AWS S3 Java SDK从S3下载照片(不在同一个目录中),将它们打包成ZIP文件,然后再次上传到S3。这个ZIP文件的大小可能会达到几GB。我目前正在使用AWS Lambda,但是其临时存储有500MB的限制。因此,我不想将ZIP文件保存到磁盘上,而是希望直接将流式传输的ZIP文件(动态创建并使用从S3下载的照片)直接传输到S3。我需要使用AWS S3 Java SDK来实现这一点。
我需要使用AWS S3 Java SDK从S3下载照片(不在同一个目录中),将它们打包成ZIP文件,然后再次上传到S3。这个ZIP文件的大小可能会达到几GB。我目前正在使用AWS Lambda,但是其临时存储有500MB的限制。因此,我不想将ZIP文件保存到磁盘上,而是希望直接将流式传输的ZIP文件(动态创建并使用从S3下载的照片)直接传输到S3。我需要使用AWS S3 Java SDK来实现这一点。
final AmazonS3 client = AmazonS3ClientBuilder.defaultClient();
final PipedOutputStream pipedOutputStream = new PipedOutputStream();
final PipedInputStream pipedInputStream = new PipedInputStream(pipedOutputStream);
final Thread s3In = new Thread(() -> {
try (final ZipOutputStream zipOutputStream = new ZipOutputStream(pipedOutputStream)) {
S3Objects
// It's just a convenient way to list all the objects. Replace with you own logic.
.inBucket(client, "bucket")
.forEach((S3ObjectSummary objectSummary) -> {
try {
if (objectSummary.getKey().endsWith(".png")) {
System.out.println("Processing " + objectSummary.getKey());
final ZipEntry entry = new ZipEntry(
UUID.randomUUID().toString() + ".png" // I'm too lazy to extract file name from the
// objectSummary
);
zipOutputStream.putNextEntry(entry);
IOUtils.copy(
client.getObject(
objectSummary.getBucketName(),
objectSummary.getKey()
).getObjectContent(),
zipOutputStream
);
zipOutputStream.closeEntry();
}
} catch (final Exception all) {
all.printStackTrace();
}
});
} catch (final Exception all) {
all.printStackTrace();
}
});
final Thread s3Out = new Thread(() -> {
try {
client.putObject(
"another-bucket",
"previews.zip",
pipedInputStream,
new ObjectMetadata()
);
pipedInputStream.close();
} catch (final Exception all) {
all.printStackTrace();
}
});
s3In.start();
s3Out.start();
s3In.join();
s3Out.join();
但请注意,它会打印一个警告:
WARNING: No content length specified for stream data. Stream contents will be buffered in memory and could result in out of memory errors.
final S3Client client = S3Client.create();
final PipedOutputStream pipedOutputStream = new PipedOutputStream();
final PipedInputStream pipedInputStream = new PipedInputStream(pipedOutputStream);
final Thread s3In = new Thread(() -> {
try (final ZipOutputStream zipOutputStream = new ZipOutputStream(pipedOutputStream)) {
client.listObjectsV2Paginator(
ListObjectsV2Request
.builder()
.bucket("bucket")
.build()
)
.contents()
.forEach((S3Object object) -> {
try {
if (object.key().endsWith(".png")) {
System.out.println("Processing " + object.key());
final ZipEntry entry = new ZipEntry(
UUID.randomUUID().toString() + ".png" // I'm too lazy to extract file name from the object
);
zipOutputStream.putNextEntry(entry);
client.getObject(
GetObjectRequest
.builder()
.bucket("bucket")
.key(object.key())
.build(),
ResponseTransformer.toOutputStream(zipOutputStream)
);
zipOutputStream.closeEntry();
}
} catch (final Exception all) {
all.printStackTrace();
}
});
} catch (final Exception all) {
all.printStackTrace();
}
});
final Thread s3Out = new Thread(() -> {
try {
client.putObject(
PutObjectRequest
.builder()
.bucket("another-bucket")
.key("previews.zip")
.build(),
RequestBody.fromBytes(
IOUtils.toByteArray(pipedInputStream)
)
);
} catch (final Exception all) {
all.printStackTrace();
}
});
s3In.start();
s3Out.start();
s3In.join();
s3Out.join();
它遭受同样的问题:上传前需要在内存中准备ZIP。
如果您感兴趣,我已经准备了一个演示项目,您可以玩转代码。
new ZipOutputStream(new S3OutputStream(s3Client, "my_bucket", "path"))
。import java.io.ByteArrayInputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.AbortMultipartUploadRequest;
import com.amazonaws.services.s3.model.CannedAccessControlList;
import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest;
import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
import com.amazonaws.services.s3.model.InitiateMultipartUploadResult;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.PartETag;
import com.amazonaws.services.s3.model.PutObjectRequest;
import com.amazonaws.services.s3.model.UploadPartRequest;
import com.amazonaws.services.s3.model.UploadPartResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class S3OutputStream extends OutputStream {
private static final Logger LOG = LoggerFactory.getLogger(S3OutputStream.class);
/** Default chunk size is 10MB */
protected static final int BUFFER_SIZE = 10000000;
/** The bucket-name on Amazon S3 */
private final String bucket;
/** The path (key) name within the bucket */
private final String path;
/** The temporary buffer used for storing the chunks */
private final byte[] buf;
/** The position in the buffer */
private int position;
/** Amazon S3 client. TODO: support KMS */
private final AmazonS3 s3Client;
/** The unique id for this upload */
private String uploadId;
/** Collection of the etags for the parts that have been uploaded */
private final List<PartETag> etags;
/** indicates whether the stream is still open / valid */
private boolean open;
/**
* Creates a new S3 OutputStream
* @param s3Client the AmazonS3 client
* @param bucket name of the bucket
* @param path path within the bucket
*/
public S3OutputStream(AmazonS3 s3Client, String bucket, String path) {
this.s3Client = s3Client;
this.bucket = bucket;
this.path = path;
this.buf = new byte[BUFFER_SIZE];
this.position = 0;
this.etags = new ArrayList<>();
this.open = true;
}
/**
* Write an array to the S3 output stream.
*
* @param b the byte-array to append
*/
@Override
public void write(byte[] b) {
write(b,0,b.length);
}
/**
* Writes an array to the S3 Output Stream
*
* @param byteArray the array to write
* @param o the offset into the array
* @param l the number of bytes to write
*/
@Override
public void write(final byte[] byteArray, final int o, final int l) {
this.assertOpen();
int ofs = o, len = l;
int size;
while (len > (size = this.buf.length - position)) {
System.arraycopy(byteArray, ofs, this.buf, this.position, size);
this.position += size;
flushBufferAndRewind();
ofs += size;
len -= size;
}
System.arraycopy(byteArray, ofs, this.buf, this.position, len);
this.position += len;
}
/**
* Flushes the buffer by uploading a part to S3.
*/
@Override
public synchronized void flush() {
this.assertOpen();
LOG.debug("Flush was called");
}
protected void flushBufferAndRewind() {
if (uploadId == null) {
LOG.debug("Starting a multipart upload for {}/{}",this.bucket,this.path);
final InitiateMultipartUploadRequest request = new InitiateMultipartUploadRequest(this.bucket, this.path)
.withCannedACL(CannedAccessControlList.BucketOwnerFullControl);
InitiateMultipartUploadResult initResponse = s3Client.initiateMultipartUpload(request);
this.uploadId = initResponse.getUploadId();
}
uploadPart();
this.position = 0;
}
protected void uploadPart() {
LOG.debug("Uploading part {}",this.etags.size());
UploadPartResult uploadResult = this.s3Client.uploadPart(new UploadPartRequest()
.withBucketName(this.bucket)
.withKey(this.path)
.withUploadId(this.uploadId)
.withInputStream(new ByteArrayInputStream(buf,0,this.position))
.withPartNumber(this.etags.size() + 1)
.withPartSize(this.position));
this.etags.add(uploadResult.getPartETag());
}
@Override
public void close() {
if (this.open) {
this.open = false;
if (this.uploadId != null) {
if (this.position > 0) {
uploadPart();
}
LOG.debug("Completing multipart");
this.s3Client.completeMultipartUpload(new CompleteMultipartUploadRequest(bucket, path, uploadId, etags));
}
else {
LOG.debug("Uploading object at once to {}/{}",this.bucket,this.path);
final ObjectMetadata metadata = new ObjectMetadata();
metadata.setContentLength(this.position);
final PutObjectRequest request = new PutObjectRequest(this.bucket, this.path, new ByteArrayInputStream(this.buf, 0, this.position), metadata)
.withCannedAcl(CannedAccessControlList.BucketOwnerFullControl);
this.s3Client.putObject(request);
}
}
}
public void cancel() {
this.open = false;
if (this.uploadId != null) {
LOG.debug("Aborting multipart upload");
this.s3Client.abortMultipartUpload(new AbortMultipartUploadRequest(this.bucket, this.path, this.uploadId));
}
}
@Override
public void write(int b) {
this.assertOpen();
if (position >= this.buf.length) {
flushBufferAndRewind();
}
this.buf[position++] = (byte)b;
}
private void assertOpen() {
if (!this.open) {
throw new IllegalStateException("Closed");
}
}
}
非常抱歉回复晚了,我昨天为我的最新项目完成了这个任务,请查看下面的完整代码
假设当我们在S3上上传文件时,它会返回已上传文件的ObjectKey,因此我创建了一个名为FileKey的类来处理此操作。
package com.myprojectName.model.key;
import java.time.Instant;
import javax.persistence.Entity;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@Entity
@NoArgsConstructor
public class FileKey {
private String fileObjectKey;
private String fileName;
private int fileSize;
private String fileType;
}
我将presignedUrl的返回值存储在DownloadDetailsDTO中。
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Getter;
import lombok.NoArgsConstructor;
import java.net.URL;
@NoArgsConstructor
@AllArgsConstructor
@Getter
@Builder
public class FileDownloadDetailsDTO {
private String name;
private Long size;
private String contentType;
private URL preSignedDownloadUrl;
public FileDownloadDetailsDTO(PreSignedUrlAndMetadata objectMetadata) {
this.name = objectMetadata.getName();
this.size = objectMetadata.getSize();
this.contentType = objectMetadata.getContentType();
this.preSignedDownloadUrl = objectMetadata.getUrl();
}
}
PreSignedUrlAndMetaData 包含在 s3 存储桶上创建的 URL,如果不确定,请查看下面的代码
public class PreSignedUrlAndMetadata {
private final URL url;
private final String name;
private final String contentType;
private final Long size;
}
public FileDownloadDetailsDTO getDownloadFilesInZipDetails(String zipFileName, List<FileKey> files) {
PreSignedUrlAndMetadata preSignedUrlAndMetadata;
File zipFile = null;
try {
zipFile = File.createTempFile(zipFileName, "file");
try (FileOutputStream fos = new FileOutputStream(zipFile); ZipOutputStream zos = new ZipOutputStream(fos)) {
for (FileKey file : files) {
String name = null;
if (ObjectUtils.isNotEmpty(file.getFileName())) {
name = file.getFileName();
}
ZipEntry entry = new ZipEntry(name);
try (InputStream inputStream = getInputStreamForFileKey(file.getFileObjectKey())) {
zos.putNextEntry(entry);
IOUtils.copy(inputStream, zos);
zos.closeEntry();
}
}
}
try (FileInputStream fis = new FileInputStream(zipFile)) {
TempFileObjectKey fileObjectKey =uploadTemp(fis, zipFile.length());
preSignedUrlAndMetadata = new PreSignedUrlAndMetadata(url, metadata.getUserMetaDataOf(USER_METADATA_NAME), contentType, metadata.getContentLength());
}
} catch (Exception e) {
throw new ApplicationException("Error while creating zip file for " + archiveRequestDTO.getArchiveName(), e, ApplicationErrorCode.INTERNAL_SERVER_ERROR);
} finally {
FileUtils.deleteQuietly(zipFile);
}
return FileDownloadDetailsDTO.builder().name(archiveRequestDTO.getArchiveName() + ".zip")
.size(preSignedUrlAndMetadata.getSize()).preSignedDownloadUrl(preSignedUrlAndMetadata.getUrl()).build();
}
public InputStream getInputStreamForFileKey(String key) {
TempFileObjectKey tempFileObjectKey = new TempFileObjectKey(getActualPrefix(key));
return storageService.getInputStream(tempFileObjectKey);
}
String getActualPrefix(String prefix){
return prefix.replaceAll("_","/");
}
public TempFileObjectKey uploadTemp(InputStream inputStream, Long length) {
TempFileObjectKey tempFileObjectKey = s3StorageManager.buildTempFileFullKey();
ObjectMetadata objectMetadata = new ObjectMetadata();
if (length != null) {
objectMetadata.setContentLength(length);
}
Upload upload = com.amazonaws.services.s3.transfer.TransferManager.upload(getBucketName(abstractObjectKey), abstractObjectKey.getObjectKey(), inputStream, objectMetadata);
try {
upload.waitForCompletion();
} catch (InterruptedException e) {
throw new ApplicationException(e.getMessage(), e, ApplicationErrorCode.INTERNAL_SERVER_ERROR);
}
return tempFileObjectKey;
}
希望这对你们有所帮助。 如果还有任何疑问,欢迎向我提问。 谢谢。
我已经创建了一个 GitHub 存储库,你可以在这里查看我的实现: https://github.com/yufeikang/serverless-zip-s3 该存储库包括一个使用 Node.js 编写的 Lambda 函数,你可以使用它来压缩 S3 目录,并提供了部署和使用该函数的说明。