对于大型上传,您需要使用S3多部分上传功能。通过一点点的工作,您可以编写一个包装器流,将每个最小上传部分(5MB)缓冲到MemoryStream中,并将它们作为部分刷新。
这是我为这个特定情况编写的代码:
using System.IO.Compression;
using Microsoft.Extensions.Logging;
using Amazon.S3;
using Amazon.S3.Model;
using Atelie.Dev.Utils;
var s3Client = new AmazonS3Client();
var req = new InitiateMultipartUploadRequest { BucketName = "...", Key = "...", ContentType = "application/zip" };
var res = await s3Client.InitiateMultipartUploadAsync(req);
var loggerFactory = new LoggerFactory();
var logger = loggerFactory.CreateLogger<S3MultipartUploadStream>();
await using (var upload = new S3MultipartUploadStream(s3Client, res, logger))
{
using (var archive = new ZipArchive(upload, ZipArchiveMode.Create, leaveOpen: true))
{
var ze = archive.CreateEntry("...");
await using var es = ze.Open();
}
await upload.CompleteUploadAsync();
}
以下是 S3MultipartUploadStream
类的代码(注意,这里可能有风险):
using System;
using System.Collections.Concurrent;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Amazon.S3;
using Amazon.S3.Model;
using Microsoft.Extensions.Logging;
namespace Atelie.Dev.Utils;
public class S3MultipartUploadStream : Stream
{
private const int MinPartSize = 5 * 1024 * 1024;
private const int BufferCapacity = MinPartSize;
private readonly IAmazonS3 _s3Client;
private readonly InitiateMultipartUploadResponse _mur;
private readonly ILogger<S3MultipartUploadStream> _logger;
private readonly ConcurrentBag<PartETag> _parts = new();
private readonly ConcurrentBag<Task> _uploadTasks = new();
private readonly CancellationTokenSource _cts = new();
private readonly int _concurrency;
private volatile MemoryStream _mem = new(BufferCapacity);
private Task? _completedTask;
private long _realPosition;
private int _partNumber;
public S3MultipartUploadStream(IAmazonS3 s3Client, InitiateMultipartUploadResponse mur, ILogger<S3MultipartUploadStream> logger, int concurrency = 4)
{
_s3Client = s3Client;
_mur = mur;
_logger = logger;
_concurrency = concurrency;
}
public Task CompleteUploadAsync()
{
if (_completedTask != null)
throw new InvalidOperationException("Upload already finished or aborted");
return _completedTask = Task.Run(async () =>
{
if (_mem.Position > 0)
await UploadPart(true);
Task.WaitAll(_uploadTasks.ToArray());
var req = new CompleteMultipartUploadRequest
{
BucketName = _mur.BucketName,
Key = _mur.Key,
UploadId = _mur.UploadId,
PartETags = _parts.ToList(),
};
_logger.LogInformation("Finished upload of {Count} parts", req.PartETags.Count);
await _s3Client.CompleteMultipartUploadAsync(req);
});
}
private Task AbortUploadAsync()
{
return _completedTask ??= Task.Run(async () =>
{
_cts.Cancel();
var req = new AbortMultipartUploadRequest
{
BucketName = _mur.BucketName,
Key = _mur.Key,
UploadId = _mur.UploadId,
};
_logger.LogInformation("Aborting upload of: {Key}", req.Key);
await _s3Client.AbortMultipartUploadAsync(req);
});
}
private async Task UploadPart(bool isLastPart)
{
if (_mem.Position == 0)
return;
if (_mem.Length < MinPartSize && !isLastPart)
throw new InvalidOperationException("You can't upload a part smaller than 5MB, unless it's the last one");
while (_uploadTasks.Count(t => !t.IsCompleted) >= _concurrency)
await Task.Delay(100, _cts.Token);
var partNumber = Interlocked.Increment(ref _partNumber);
var msToWrite = _mem;
_mem = new MemoryStream(BufferCapacity);
_uploadTasks.Add(Task.Run(async () =>
{
await using (msToWrite)
{
msToWrite.Position = 0;
var req = new UploadPartRequest
{
BucketName = _mur.BucketName,
Key = _mur.Key,
InputStream = msToWrite,
PartNumber = partNumber,
PartSize = msToWrite.Length,
UploadId = _mur.UploadId,
IsLastPart = isLastPart,
};
_logger.LogInformation("Part {Seq}: staring upload of {Size} bytes", req.PartNumber, req.PartSize);
var res = await _s3Client.UploadPartAsync(req, _cts.Token);
_parts.Add(new PartETag(req.PartNumber, res.ETag));
_logger.LogInformation("Part {Seq}: upload finished", res.PartNumber);
}
}));
}
#region Dispose and DisposeAsync
protected override void Dispose(bool disposing)
{
var endTask = AbortUploadAsync();
endTask.GetAwaiter().GetResult();
endTask.Dispose();
_mem.Dispose();
_cts.Dispose();
base.Dispose(disposing);
}
public override async ValueTask DisposeAsync()
{
GC.SuppressFinalize(this);
var endTask = AbortUploadAsync();
await Task.WhenAll(_mem.DisposeAsync().AsTask(), endTask, base.DisposeAsync().AsTask());
endTask.Dispose();
_cts.Dispose();
await base.DisposeAsync();
}
#endregion
#region Write and WriteAsync
public override void Write(byte[] buffer, int offset, int count)
{
while (count > 0)
{
var limit = _mem.Capacity - (int) _mem.Position;
if (limit == 0)
UploadPart(false).GetAwaiter().GetResult();
else
{
var written = Math.Min(limit, count);
_mem.Write(buffer, offset, written);
offset += written;
count -= written;
Interlocked.Add(ref _realPosition, written);
}
}
}
public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
await WriteAsync(buffer.AsMemory(offset, count), cancellationToken);
}
public override async ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
{
var unwrittenBytes = buffer.Length;
var offset = 0;
while (unwrittenBytes > 0)
{
var limit = _mem.Capacity - (int) _mem.Position;
if (limit == 0)
await UploadPart(false);
else
{
var currentSliceSize = Math.Min(limit, unwrittenBytes);
await _mem.WriteAsync(buffer.Slice(offset, currentSliceSize), cancellationToken);
offset += currentSliceSize;
unwrittenBytes -= currentSliceSize;
Interlocked.Add(ref _realPosition, currentSliceSize);
}
}
}
#endregion
#region Remaining of Stream implementation
public override int Read(byte[] buffer, int offset, int count) => throw new InvalidOperationException();
public override long Seek(long offset, SeekOrigin origin) => throw new InvalidOperationException();
public override void SetLength(long value) => throw new InvalidOperationException();
public override void Flush() {}
public override bool CanRead => false;
public override bool CanSeek => false;
public override bool CanWrite => true;
public override long Length => throw new InvalidOperationException();
public override long Position
{
get => _realPosition;
set => throw new InvalidOperationException();
}
#endregion
}