MinIO .NET SDK 文件上传逻辑分析

概述

MinIO .NET SDK 在 ObjectOperations.cs 文件中实现了完整的文件上传逻辑，支持单次上传和**分块上传（Multipart Upload）**两种模式。系统会根据文件大小自动选择合适的上传方式。

---

一、核心入口方法

1.1 PutObjectAsync 方法

这是文件上传的主入口方法，位于第 667-763 行。

方法签名：

public async Task<PutObjectResponse> PutObjectAsync(
    PutObjectArgs args,
    CancellationToken cancellationToken = default
)

核心参数：

• PutObjectArgs - 封装了上传所需的所有信息：
  • 桶名称（BucketName）
  • 对象名称（ObjectName）
  • 文件名（FileName）或数据流（ObjectStreamData）
  • 文件大小（ObjectSize）
  • 内容类型（ContentType）
  • 服务器端加密配置（SSE）
  • 标签、合规保留配置等

---

二、上传模式决策逻辑

2.1 关键常量定义

在 Constants.cs 文件中定义了上传相关的重要常量：

// 最小分块大小：5MB
public static long MinimumPartSize = 5 * 1024L * 1024L;

// PUT操作的最小分块大小：16MB
public static long MinimumPUTPartSize = 16 * 1024L * 1024L;

// COPY操作的最小分块大小：512MB
public static long MinimumCOPYPartSize = 512 * 1024L * 1024L;

// 最大分块大小：5GB
public static long MaximumPartSize = 5 * 1024L * 1024L * 1024L;

// 最大分块数量：10000
public static int MaxParts = 10000;

// 单次PUT上传的最大对象大小：5GB
public static long MaxSinglePutObjectSize = 1024L * 1024L * 1024L * 5;

// 分块上传的最大对象大小：5TB
public static long MaxMultipartPutObjectSize = 1024L * 1024L * 1024L * 1024L * 5;

2.2 上传模式选择逻辑（第 684-701 行）

// 判断条件：文件大小 < 5MB 或者是 Snowball 对象
if ((args.ObjectSize < Constants.MinimumPartSize || isSnowball)
    && args.ObjectSize >= 0
    && args.ObjectStreamData is not null)
{
    // 使用单次上传模式
    var bytes = await ReadFullAsync(args.ObjectStreamData, (int)args.ObjectSize)
        .ConfigureAwait(false);
    
    // 验证读取的字节数
    var bytesRead = bytes.Length;
    if (bytesRead != (int)args.ObjectSize)
        throw new UnexpectedShortReadException(...);
    
    // 执行单次上传
    return await PutObjectSinglePartAsync(args, cancellationToken, true)
        .ConfigureAwait(false);
}

选择单次上传的条件：

1. 文件大小 < 5MB (Constants.MinimumPartSize)
2. 或 是 Snowball 对象（AWS Snowball 数据导入场景）
3. 且 文件大小已知（≥ 0）
4. 且 数据流不为空

选择分块上传的条件：

• 文件大小 ≥ 5MB

---

三、单次上传模式（PutObjectSinglePartAsync）

3.1 方法定义（第 1009-1058 行）

private async Task<PutObjectResponse> PutObjectSinglePartAsync(
    PutObjectArgs args,
    CancellationToken cancellationToken = default,
    bool singleFile = false
)

3.2 上传流程

1. 创建 HTTP 请求

   var requestMessageBuilder = await this.CreateRequest(args).ConfigureAwait(false);
   

2. 执行上传请求

   using (response = await this.ExecuteTaskAsync(
       requestMessageBuilder,
       cancellationToken: cancellationToken
   ).ConfigureAwait(false))
   

3. 进度报告（如果提供了进度回调）

   if (singleFile && args.Progress is not null)
   {
       var statArgs = new StatObjectArgs()
           .WithBucket(args.BucketName)
           .WithObject(args.ObjectName);
       var stat = await StatObjectAsync(statArgs, cancellationToken).ConfigureAwait(false);
       
       if (response.StatusCode == HttpStatusCode.OK)
       {
           progressReport.Percentage = 100;
           progressReport.TotalBytesTransferred = stat.Size;
       }
       
       args.Progress.Report(progressReport);
   }
   

4. 返回响应

   return new PutObjectResponse(
       response.StatusCode,
       response.Content,
       response.Headers,
       args.ObjectSize,
       args.ObjectName
   );
   

   5.注意项

   单次上传有一个容易忽略的点，就是在上传时没有判断是文件流还是文件，这个在多文件上传时体现出来了代码，我之前纠结过这样的原因，排查后发现了PutObjectAsync方法入口有如下代码：

       args?.Validate();
   
   

   这行代码并不是只单纯做了验证，而是调用了如下方法：

     private void Populate()
     {
         if (!string.IsNullOrWhiteSpace(FileName))
         {
             var fileInfo = new FileInfo(FileName);
             ObjectSize = fileInfo.Length;
             ObjectStreamData = new FileStream(FileName, FileMode.Open, FileAccess.Read);
         }
     }
   

   可以看到是直接将文件转为了文件流的，这也就解释了为什么文单文件上传是不需要判断是文件流还是文件的原因

---

四、分块上传模式（Multipart Upload）

分块上传是 MinIO SDK 的核心功能，用于处理大文件上传。整个过程分为三个阶段：

4.1 阶段一：初始化分块上传（NewMultipartUploadAsync）

代码位置： 第 704-717 行

// 创建分块上传请求参数
var multipartUploadArgs = new NewMultipartUploadPutArgs()
    .WithBucket(args.BucketName)
    .WithObject(args.ObjectName)
    .WithVersionId(args.VersionId)
    .WithHeaders(args.Headers)
    .WithContentType(args.ContentType)
    .WithTagging(args.ObjectTags)
    .WithLegalHold(args.LegalHoldEnabled)
    .WithRetentionConfiguration(args.Retention)
    .WithServerSideEncryption(args.SSE);

// 获取 uploadId
var uploadId = await NewMultipartUploadAsync(multipartUploadArgs, cancellationToken)
    .ConfigureAwait(false);

返回值： uploadId - 用于标识此次分块上传任务的唯一ID

实现细节： 第 1241-1255 行

private async Task<string> NewMultipartUploadAsync(
    NewMultipartUploadPutArgs args,
    CancellationToken cancellationToken = default
)
{
    args?.Validate();
    var requestMessageBuilder = await this.CreateRequest(args).ConfigureAwait(false);
    using var response = await this.ExecuteTaskAsync(
        requestMessageBuilder,
        cancellationToken: cancellationToken
    ).ConfigureAwait(false);
    
    var uploadResponse = new NewMultipartUploadResponse(response.StatusCode, response.Content);
    return uploadResponse.UploadId;
}

4.2 阶段二：上传各个分块（PutObjectPartAsync）

代码位置： 第 1074-1133 行

4.2.1 分块大小计算

var multiPartInfo = Utils.CalculateMultiPartSize(args.ObjectSize);
var partSize = multiPartInfo.PartSize;      // 每个分块的大小
var partCount = multiPartInfo.PartCount;    // 总分块数
var lastPartSize = multiPartInfo.LastPartSize; // 最后一个分块的大小

分块大小计算逻辑： （Utils.cs 第 261-276 行）

public static MultiPartInfo CalculateMultiPartSize(long size, bool copy = false)
{
    // 如果大小未知，使用最大流对象大小
    if (size == -1) size = Constants.MaximumStreamObjectSize;

    // 检查是否超过最大允许大小（5TB）
    if (size > Constants.MaxMultipartPutObjectSize)
        throw new EntityTooLargeException(...);

    // 计算每个分块的大小
    var partSize = (double)Math.Ceiling((decimal)size / Constants.MaxParts);
    var minPartSize = copy ? Constants.MinimumCOPYPartSize : Constants.MinimumPUTPartSize;
    
    // 确保分块大小是最小分块大小的倍数
    partSize = (double)Math.Ceiling((decimal)partSize / minPartSize) * minPartSize;
    
    // 计算分块数量和最后一个分块的大小
    var partCount = Math.Ceiling(size / partSize);
    var lastPartSize = size - ((partCount - 1) * partSize);

    return new MultiPartInfo { 
        PartSize = partSize, 
        PartCount = partCount, 
        LastPartSize = lastPartSize 
    };
}

分块大小计算规则：

1. 分块大小 = ceil(文件总大小 / 10000)
2. 分块大小必须是 16MB 的倍数（PUT操作）或 512MB 的倍数（COPY操作）
3. 分块数量最多 10000 个
4. 最后一个分块可能小于标准分块大小

4.2.2 循环上传每个分块

var etags = new Dictionary<int, string>();
var progressReport = new ProgressReport();
args.Progress?.Report(progressReport);

for (partNumber = 1; partNumber <= partCount; partNumber++)
{
    // 1. 读取分块数据
    var dataToCopy = await ReadFullAsync(args.ObjectStreamData, (int)partSize)
        .ConfigureAwait(false);
    
    // 如果没有数据且已上传部分分块，则退出
    if (dataToCopy.IsEmpty && numPartsUploaded > 0)
        break;
    
    // 最后一个分块使用 lastPartSize
    if (partNumber == partCount)
        expectedReadSize = lastPartSize;
    
    // 2. 构造分块上传参数
    var putObjectArgs = new PutObjectArgs(args)
        .WithRequestBody(dataToCopy)
        .WithUploadId(args.UploadId)
        .WithPartNumber(partNumber);
    
    // 3. 上传单个分块
    var putObjectResponse = await PutObjectSinglePartAsync(putObjectArgs, cancellationToken)
        .ConfigureAwait(false);
    
    // 4. 记录 ETag
    var etag = putObjectResponse.Etag;
    numPartsUploaded++;
    totalParts[partNumber - 1] = new Part {
        PartNumber = partNumber, 
        ETag = etag, 
        Size = (long)expectedReadSize
    };
    etags[partNumber] = etag;
    
    // 5. 更新进度
    if (!dataToCopy.IsEmpty)
        progressReport.TotalBytesTransferred += dataToCopy.Length;
    if (args.ObjectSize != -1)
        progressReport.Percentage = (int)(100 * partNumber / partCount);
    args.Progress?.Report(progressReport);
}

关键点：

• 每个分块都有唯一的 partNumber（从 1 开始）
• 每个分块上传后会返回一个 ETag，用于后续验证
• 支持进度回调，可实时获取上传进度
• 如果分块数量不匹配且大小已知，会清理上传任务

4.2.3 数据读取辅助方法（ReadFullAsync）

代码位置： 第 1365-1383 行

internal static async Task<ReadOnlyMemory<byte>> ReadFullAsync(Stream data, int currentPartSize)
{
    Memory<byte> result = new byte[currentPartSize];
    var totalRead = 0;
    
    // 循环读取直到读满 currentPartSize 或到达流末尾
    while (totalRead < currentPartSize)
    {
        var curData = result[totalRead..currentPartSize];
        var curRead = await data.ReadAsync(curData).ConfigureAwait(false);
        
        // 流已结束
        if (curRead == 0)
            break;
        
        totalRead += curRead;
    }

    // 如果没有读取到任何数据，返回 null
    if (totalRead == 0)
        return null;

    // 只返回实际读取的数据部分（无需额外分配）
    return result[..totalRead];
}

特点：

• 确保每次读取指定大小的数据（除非流结束）
• 使用 ReadOnlyMemory<byte> 避免不必要的内存拷贝
• 自动处理流末尾情况

4.3 阶段三：完成分块上传（CompleteMultipartUploadAsync）

代码位置： 第 751-762 行（调用处）

// 构造完成分块上传的参数
var completeMultipartUploadArgs = new CompleteMultipartUploadArgs()
    .WithBucket(args.BucketName)
    .WithObject(args.ObjectName)
    .WithUploadId(uploadId)
    .WithETags(etags);  // 包含所有分块的 ETag

// 完成分块上传
var putObjectResponse = await CompleteMultipartUploadAsync(
    completeMultipartUploadArgs,
    cancellationToken
).ConfigureAwait(false);

putObjectResponse.Size = args.ObjectSize;
return putObjectResponse;

实现细节： 第 1338-1357 行

private async Task<PutObjectResponse> CompleteMultipartUploadAsync(
    CompleteMultipartUploadArgs args,
    CancellationToken cancellationToken
)
{
    args?.Validate();
    var requestMessageBuilder = await this.CreateRequest(args).ConfigureAwait(false);
    
    using var response = await this.ExecuteTaskAsync(
        requestMessageBuilder,
        cancellationToken: cancellationToken
    ).ConfigureAwait(false);
    
    return new PutObjectResponse(
        response.StatusCode,
        response.Content,
        response.Headers,
        -1,
        args.ObjectName
    );
}

功能：

• 向服务器发送所有分块的 ETag 列表
• 服务器验证所有分块并合并成完整对象
• 返回最终上传结果

4.4 异常处理和清理

如果分块上传过程中发现分块数量不匹配（且文件大小已知），会自动清理：

// 代码位置：第 1122-1130 行
if (partCount != numPartsUploaded && args.ObjectSize != -1)
{
    var removeUploadArgs = new RemoveUploadArgs()
        .WithBucket(args.BucketName)
        .WithObject(args.ObjectName)
        .WithUploadId(args.UploadId);
    
    await RemoveUploadAsync(removeUploadArgs, cancellationToken).ConfigureAwait(false);
    return null;
}

---

五、文件来源处理

SDK 支持两种文件来源：

5.1 从文件上传

代码位置： 第 731-742 行

if (!string.IsNullOrEmpty(args.FileName))
{
    var fileStream = new FileStream(args.FileName, FileMode.Open, FileAccess.Read);
    using (fileStream)
    {
        putObjectPartArgs = putObjectPartArgs
            .WithStreamData(fileStream)
            .WithObjectSize(fileStream.Length)
            .WithRequestBody(null);
        
        etags = await PutObjectPartAsync(putObjectPartArgs, cancellationToken)
            .ConfigureAwait(false);
    }
}

5.2 从数据流上传

代码位置： 第 744-749 行

else
{
    // 直接使用提供的数据流
    etags = await PutObjectPartAsync(putObjectPartArgs, cancellationToken)
        .ConfigureAwait(false);
}

---

六、安全特性

6.1 服务器端加密（SSE）

在初始化分块上传时，SDK 会自动处理加密配置：

// 代码位置：第 673 行
args.SSE?.Marshal(args.Headers);

支持的加密类型：

• SSE-S3 - Amazon S3 管理的密钥
• SSE-KMS - AWS Key Management Service 管理的密钥
• SSE-C - 客户提供的密钥

6.2 数据完整性验证

每个分块上传时会计算 SHA256 哈希值：

// 代码位置：PutObjectArgs.BuildRequest 方法（第 111-123 行）
if (!RequestBody.IsEmpty)
{
    var hash = SHA256.HashData(RequestBody.Span);
    var hex = BitConverter.ToString(hash)
        .Replace("-", string.Empty, StringComparison.OrdinalIgnoreCase)
        .ToLowerInvariant();
    
    requestMessageBuilder.AddOrUpdateHeaderParameter("x-amz-content-sha256", hex);
    requestMessageBuilder.SetBody(RequestBody);
}

6.3 对象锁定和合规性

支持对象保留和法律保留配置：

// 对象保留配置
if (Retention is not null)
{
    requestMessageBuilder.AddOrUpdateHeaderParameter(
        "x-amz-object-lock-retain-until-date", 
        Retention.RetainUntilDate
    );
    requestMessageBuilder.AddOrUpdateHeaderParameter(
        "x-amz-object-lock-mode", 
        Retention.Mode.ToString()
    );
}

// 法律保留配置
if (LegalHoldEnabled is not null)
    requestMessageBuilder.AddOrUpdateHeaderParameter(
        "x-amz-object-lock-legal-hold",
        LegalHoldEnabled == true ? "ON" : "OFF"
    );

---

七、上传流程完整示意图

┌─────────────────────────────────────────────────────────┐
│                   PutObjectAsync                         │
│                   (入口方法)                              │
└───────────────────────┬─────────────────────────────────┘
                        │
                        ▼
         ┌──────────────────────────────┐
         │    验证参数和加密配置          │
         └──────────────┬───────────────┘
                        │
                        ▼
         ┌──────────────────────────────┐
         │  判断文件大小和特殊标志        │
         │  - 是否 < 5MB?                │
         │  - 是否 Snowball 对象?        │
         └──────────────┬───────────────┘
                        │
           ┌────────────┴────────────┐
           │                         │
           ▼                         ▼
    ┌─────────────┐          ┌─────────────────┐
    │  单次上传    │          │  分块上传        │
    │  (< 5MB)    │          │  (≥ 5MB)        │
    └──────┬──────┘          └────────┬────────┘
           │                          │
           │                          ▼
           │              ┌──────────────────────────┐
           │              │ 1. 初始化分块上传         │
           │              │    NewMultipartUpload    │
           │              │    → 获取 uploadId       │
           │              └───────────┬──────────────┘
           │                          │
           │                          ▼
           │              ┌──────────────────────────┐
           │              │ 2. 计算分块大小           │
           │              │    - partSize            │
           │              │    - partCount           │
           │              │    - lastPartSize        │
           │              └───────────┬──────────────┘
           │                          │
           │                          ▼
           │              ┌──────────────────────────┐
           │              │ 3. 循环上传每个分块       │
           │              │    for (1 to partCount)  │
           │              │    {                     │
           │              │      - 读取分块数据       │
           │              │      - 上传分块          │
           │              │      - 记录 ETag         │
           │              │      - 更新进度          │
           │              │    }                     │
           │              └───────────┬──────────────┘
           │                          │
           │                          ▼
           │              ┌──────────────────────────┐
           │              │ 4. 完成分块上传           │
           │              │    CompleteMultipartUpload│
           │              │    - 提交所有 ETags      │
           │              │    - 服务器合并分块      │
           │              └───────────┬──────────────┘
           │                          │
           └──────────────┬───────────┘
                          │
                          ▼
              ┌───────────────────────┐
              │   返回上传结果         │
              │   PutObjectResponse   │
              └───────────────────────┘

---

八、分块上传的优势

1. 大文件支持

   • 单次上传限制：5GB
   • 分块上传限制：5TB（10000个分块 × 最大5GB/分块）

2. 断点续传

   • 可以查询未完成的上传任务
   • 使用 ListIncompleteUploadsEnumAsync 列出未完成的上传
   • 使用 RemoveIncompleteUploadAsync 清理未完成的上传

3. 并行上传（理论上支持，但当前实现是串行）

   • 分块独立上传
   • 可以实现分块的并行上传以提高速度

4. 错误恢复

   • 单个分块失败只需重传该分块
   • 不需要重新上传整个文件

5. 进度监控

   • 实时获取上传进度
   • 精确到每个分块的上传状态

---

九、使用示例

10.1 简单上传（小文件）

var minioClient = new MinioClient()
    .WithEndpoint("play.min.io")
    .WithCredentials("minioadmin", "minioadmin")
    .Build();

var args = new PutObjectArgs()
    .WithBucket("mybucket")
    .WithObject("myobject")
    .WithFileName("path/to/file")
    .WithContentType("application/octet-stream");

await minioClient.PutObjectAsync(args);

10.2 带进度回调的上传

var progress = new Progress<ProgressReport>(report =>
{
    Console.WriteLine($"Progress: {report.Percentage}% ({report.TotalBytesTransferred} bytes)");
});

var args = new PutObjectArgs()
    .WithBucket("mybucket")
    .WithObject("largefile")
    .WithFileName("path/to/largefile")
    .WithProgress(progress);

await minioClient.PutObjectAsync(args);

10.3 从流上传

using var fileStream = File.OpenRead("path/to/file");

var args = new PutObjectArgs()
    .WithBucket("mybucket")
    .WithObject("myobject")
    .WithStreamData(fileStream)
    .WithObjectSize(fileStream.Length)
    .WithContentType("application/pdf");

await minioClient.PutObjectAsync(args);

10.4 带加密的上传

// 使用 SSE-S3 加密
var sse = new SSES3();

var args = new PutObjectArgs()
    .WithBucket("mybucket")
    .WithObject("encrypted-object")
    .WithFileName("path/to/file")
    .WithServerSideEncryption(sse);

await minioClient.PutObjectAsync(args);

---

十一、性能优化建议

1. 合理设置分块大小

   • 默认分块大小：16MB（PUT操作）
   • 可以根据网络状况调整

2. 使用进度回调

   • 实时监控上传进度
   • 及时发现上传问题

3. 异常重试机制

   • SDK 内置重试机制（通过 IRetryPolicyHandler）
   • 可以自定义重试策略

4. 清理未完成的上传

   • 定期清理未完成的分块上传
   • 避免存储空间浪费

5. 并行上传优化（需要自定义实现）

   • 当前实现是串行上传分块
   • 可以修改代码实现并行上传以提高速度

---

十二、总结

MinIO .NET SDK 的文件上传功能设计精良，主要特点：

1. 自动模式选择：根据文件大小自动选择单次上传或分块上传
2. 分块上传机制：支持大文件（最大 5TB）的可靠上传
3. 进度监控：实时获取上传进度
4. 安全性：支持多种服务器端加密方式和数据完整性验证
5. 灵活性：支持文件和流两种数据源
6. 可靠性：内置异常处理和清理机制

分块上传流程清晰：初始化 → 上传分块 → 完成上传，每个阶段都有完善的错误处理和验证机制。