如何在Blob存储中实现并行分块上传以提升文件上传速度

原串行代码的问题在于读取文件和上传块是串行执行，每完成一个块的上传才会处理下一个，完全没利用到网络并行的优势。要实现真正的并行上传，需要把块的读取与上传解耦，同时控制并发量避免资源过载，具体实现步骤如下：

核心优化思路 #

1. 预计算所有块的元数据（偏移量、大小、块ID），让每个块的处理可以独立进行
2. 使用并行任务池执行块的读取与上传，通过信号量控制并发数
3. 所有块上传完成后，统一提交块列表

优化后的代码实现 #

using System.Collections.Concurrent;
using System.Threading;

public static async Task UploadInParallelBlocks(BlobContainerClient blobContainerClient, string localFilePath, int blockSize, int maxDegreeOfParallelism = 4)
{
    string fileName = Path.GetFileName(localFilePath);
    BlockBlobClient blobClient = blobContainerClient.GetBlockBlobClient(fileName);
    long fileLength = new FileInfo(localFilePath).Length;

    // 预生成所有块的元数据：块ID、起始偏移、块大小
    var blockInfos = new List<BlockInfo>();
    long currentOffset = 0;
    while (currentOffset < fileLength)
    {
        long currentBlockSize = Math.Min(blockSize, fileLength - currentOffset);
        string blockId = Convert.ToBase64String(Encoding.UTF8.GetBytes(Guid.NewGuid().ToString()));
        blockInfos.Add(new BlockInfo
        {
            BlockId = blockId,
            Offset = currentOffset,
            Size = currentBlockSize
        });
        currentOffset += currentBlockSize;
    }

    // 并发控制信号量
    using var semaphore = new SemaphoreSlim(maxDegreeOfParallelism);
    var uploadTasks = new List<Task>();
    // 线程安全的块ID收集容器
    var completedBlockIds = new ConcurrentBag<string>();

    foreach (var blockInfo in blockInfos)
    {
        await semaphore.WaitAsync();
        uploadTasks.Add(Task.Run(async () =>
        {
            try
            {
                // 每个任务独立打开文件流，避免线程安全问题
                using var fileStream = new FileStream(localFilePath, FileMode.Open, FileAccess.Read, FileShare.Read);
                fileStream.Position = blockInfo.Offset;
                
                byte[] buffer = new byte[blockInfo.Size];
                await fileStream.ReadAsync(buffer, 0, (int)blockInfo.Size);
                
                using var memoryStream = new MemoryStream(buffer);
                await blobClient.StageBlockAsync(blockInfo.BlockId, memoryStream);
                
                completedBlockIds.Add(blockInfo.BlockId);
            }
            finally
            {
                semaphore.Release();
            }
        }));
    }

    // 等待所有块上传完成
    await Task.WhenAll(uploadTasks);

    // 注意：提交块列表时需要保持块的顺序与文件顺序一致
    var orderedBlockIds = blockInfos.Select(info => info.BlockId).ToArray();
    await blobClient.CommitBlockListAsync(orderedBlockIds);
}

// 块元数据辅助类
private class BlockInfo
{
    public string BlockId { get; set; }
    public long Offset { get; set; }
    public long Size { get; set; }
}

关键注意事项 #

• 并发数控制：maxDegreeOfParallelism建议根据网络带宽和服务器配置调整，一般设置为4-8，过高的并发可能导致Azure Blob服务限流或本地资源耗尽
• 文件读取线程安全：每个并行任务独立打开FileStream，避免共享流导致的Position混乱
• 块顺序保持：提交块列表时必须严格按照文件的块顺序提交，否则最终文件会损坏，因此不能直接用ConcurrentBag的无序结果，需要依赖预生成的有序块信息
• 异常处理：实际使用中建议添加重试逻辑，处理网络波动导致的上传失败

内容的提问来源于stack exchange，提问作者John