上传至Azure Storage时是否应使用Parallel.Foreach？代码优化咨询

问题背景 #

现有一段从MongoDB获取图片并上传至Azure Storage的异步代码，当前通过foreach循环结合Task.WhenAll实现并行上传。想知道改用Parallel.ForEach能否提升性能，或是有没有更高效的实现方式？

原代码如下：

public async Task UploadAssetsAsync(Func<GridFSFileInfo, string> prefixSelector, List<GridFSFileInfo> files, Func<GridFSFileInfo, Task<Stream>> streamOpener, Func<string, Task> progressAction)
{
    if (flyersContainerClient == null)
        throw new Exception("Container client not initialized. Please initialize before doing blob operations.");
    var q = new Queue<Task<Response<BlobContentInfo>>>();
    progressAction?.Invoke($"{files.Count}");
    foreach (var f in files)
    {
        var pathPrefix = prefixSelector(f);
        var blobClient = flyersContainerClient.GetBlobClient($"{pathPrefix}/{f.Filename.Replace("_copy", "")}");
        IDictionary<string, string> metadata = new Dictionary<string, string>();
        var blobhttpheader = new BlobHttpHeaders();
        if (f.Filename.EndsWith("svg"))
        {
            blobhttpheader.ContentType = "image/svg+xml";
        }
        var stream = await streamOpener(f);
        if (pathPrefix == "thumbnails")
        {
            var format = ImageFormat.Jpeg;
            Bitmap cropped = null;
            using (Image image = Image.FromStream(stream))
            {
                format = image.RawFormat;
                Rectangle rect = new Rectangle(0, 0, image.Width, (image.Width * 3) / 4);
                cropped = new Bitmap(image.Width, (image.Width * 3) / 4);
                using (Graphics g = Graphics.FromImage(cropped))
                {
                    g.DrawImage(image, new Rectangle(0, 0, cropped.Width, cropped.Height), rect, GraphicsUnit.Pixel);
                }
            }
            stream.Dispose();                    
            stream = new MemoryStream();                                     
            cropped.Save(stream, format);
            stream.Position = 0;
        }
        q.Enqueue(blobClient.UploadAsync(stream, new BlobUploadOptions { HttpHeaders = blobhttpheader, TransferOptions = new Azure.Storage.StorageTransferOptions { MaximumConcurrency = 8, InitialTransferSize = 50 * 1024 * 1024 } }));
    }
    await Task.WhenAll(q);
}        

核心结论：不要改用Parallel.ForEach #

Parallel.ForEach是为CPU密集型同步任务设计的，它会占用线程池线程并阻塞等待任务完成。而图片上传属于IO密集型异步操作，异步await会自动释放线程池线程去处理其他任务，避免线程资源浪费。改用Parallel.ForEach不仅不会提升性能，反而会因为线程阻塞导致资源利用率下降，甚至触发线程池饥饿。

当前代码的潜在问题 #

1. 无全局并发限制：一次性发起所有上传请求，可能触发Azure Storage的限流机制，同时打开过多流会导致本地内存/句柄耗尽。
2. 资源泄漏风险：Bitmap cropped未用using包裹，手动调用stream.Dispose()容易遗漏，导致内存泄漏。
3. 进度反馈缺失：仅上传前显示文件总数，无实时上传进度更新。
4. 分块并发设置混淆：TransferOptions.MaximumConcurrency是单文件分块上传的并发数，不是全局上传任务的并发数。

更高效的实现方案 #

1. 用SemaphoreSlim控制全局并发 #

通过SemaphoreSlim限制同时运行的上传任务数（建议设置为10-20，根据Azure配额和本地资源调整），避免请求过载。

2. 严格管理资源 #

所有实现IDisposable的对象（Stream、Bitmap、Image等）必须用using包裹，自动释放资源。

3. 实时进度跟踪 #

用线程安全的计数器跟踪已完成任务数，实时更新进度。

4. 优化图片处理逻辑 #

避免不必要的Stream复制，直接在using块内完成缩略图生成与保存。

优化后的代码示例 #

public async Task UploadAssetsAsync(Func<GridFSFileInfo, string> prefixSelector, List<GridFSFileInfo> files, Func<GridFSFileInfo, Task<Stream>> streamOpener, Func<string, Task> progressAction)
{
    if (flyersContainerClient == null)
        throw new InvalidOperationException("Container client not initialized. Please initialize before doing blob operations.");

    // 设置全局并发限制，根据实际情况调整
    const int maxConcurrency = 15;
    var semaphore = new SemaphoreSlim(maxConcurrency);
    var completedCount = 0;
    var totalFiles = files.Count;

    await progressAction?.Invoke($"开始上传，共{totalFiles}个文件");

    var tasks = files.Select(async f =>
    {
        await semaphore.WaitAsync();
        try
        {
            var pathPrefix = prefixSelector(f);
            var blobPath = $"{pathPrefix}/{f.Filename.Replace("_copy", "")}";
            var blobClient = flyersContainerClient.GetBlobClient(blobPath);

            var blobHeaders = new BlobHttpHeaders();
            if (f.Filename.EndsWith("svg", StringComparison.OrdinalIgnoreCase))
            {
                blobHeaders.ContentType = "image/svg+xml";
            }

            using var originalStream = await streamOpener(f);
            Stream uploadStream = originalStream;

            if (pathPrefix == "thumbnails")
            {
                using var image = Image.FromStream(originalStream);
                var targetHeight = (image.Width * 3) / 4;
                using var croppedBitmap = new Bitmap(image.Width, targetHeight);
                using var graphics = Graphics.FromImage(croppedBitmap);
                
                graphics.DrawImage(image, new Rectangle(0, 0, croppedBitmap.Width, croppedBitmap.Height),
                    new Rectangle(0, 0, image.Width, targetHeight), GraphicsUnit.Pixel);

                var memoryStream = new MemoryStream();
                croppedBitmap.Save(memoryStream, image.RawFormat);
                memoryStream.Position = 0;
                uploadStream = memoryStream;
            }

            var uploadOptions = new BlobUploadOptions
            {
                HttpHeaders = blobHeaders,
                TransferOptions = new StorageTransferOptions
                {
                    MaximumConcurrency = 8, // 单文件分块上传的并发数
                    InitialTransferSize = 50 * 1024 * 1024 // 50MB分块大小
                }
            };

            await blobClient.UploadAsync(uploadStream, uploadOptions);

            // 更新进度
            var currentCompleted = Interlocked.Increment(ref completedCount);
            await progressAction?.Invoke($"已完成{currentCompleted}/{totalFiles}");
        }
        finally
        {
            semaphore.Release();
        }
    });

    await Task.WhenAll(tasks);
    await progressAction?.Invoke("所有文件上传完成");
}

额外优化建议 #

• 批量获取MongoDB流：如果streamOpener是从GridFS读取单个文件流，可以考虑批量读取，减少数据库连接开销。
• Azure Storage区域选择：确保MongoDB和Azure Storage在同一区域，减少网络延迟。
• 错误处理：添加异常捕获逻辑，避免单个任务失败导致全部上传终止（可根据需求选择重试或跳过失败文件）。

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