从Azure Storage并行下载多文件生成Zip时遇流已关闭异常求助

问题根源 #

1. ZipArchive 非线程安全：.NET的ZipArchive类及其内部的ZipArchiveEntry不支持多线程并发操作，并行调用CreateEntry或写入Entry流会导致内部流状态混乱，触发流关闭异常。
2. 锁机制无效：你代码中lock(zipEntry)的锁对象是每个线程单独创建的ZipArchiveEntry，无法阻止多个线程同时操作ZipArchive本身，起不到同步作用。
3. 模式选择错误：使用ZipArchiveMode.Update模式创建新ZIP不合适，该模式用于修改已有ZIP文件，新创建ZIP应使用Create模式，前者的流管理逻辑更容易引发线程安全问题。

解决方案 #

方案一：先并行下载文件到内存，再串行打包ZIP

这种方式将IO密集型的下载操作并行化，CPU密集型的ZIP打包操作串行化，既保证效率又避免线程安全问题：

ParallelOptions options = new() { MaxDegreeOfParallelism = 10 };
// 并发容器存储下载后的文件名和内容
var downloadedFiles = new ConcurrentBag<(string FileName, byte[] Content)>();

Parallel.ForEach(supportingDocLists, options, supportingDoc =>
{
    if (!string.IsNullOrWhiteSpace(supportingDoc.FileURL) && !string.IsNullOrWhiteSpace(supportingDoc.FileName))
    {
        BlobClient blobClient = containerClient.GetBlobClient(supportingDoc.FileURL);
        
        if (blobClient.Exists())
        {
            using var tempStream = new MemoryStream();
            blobClient.DownloadTo(tempStream);
            downloadedFiles.Add((supportingDoc.FileName, tempStream.ToArray()));
        }                                
    }
});

// 串行创建ZIP文件
using (MemoryStream zipStream = new MemoryStream())
{
    using (ZipArchive zipArchive = new ZipArchive(zipStream, ZipArchiveMode.Create, false))
    {
        foreach (var file in downloadedFiles)
        {
            ZipArchiveEntry zipEntry = zipArchive.CreateEntry(file.FileName);
            using (Stream entryStream = zipEntry.Open())
            {
                entryStream.Write(file.Content, 0, file.Content.Length);
            }
        }
    }

    zipStream.Position = 0;
    return zipStream.ToArray();
}

方案二：并行下载+串行写入ZIP（用信号量同步）

如果不想一次性占用过多内存存储所有文件，可以先将单个文件下载到临时内存流，再通过信号量控制串行写入ZIP：

SemaphoreSlim semaphore = new(1, 1);
ParallelOptions options = new() { MaxDegreeOfParallelism = 10 };

using (MemoryStream zipStream = new MemoryStream())
{
    using (ZipArchive zipArchive = new ZipArchive(zipStream, ZipArchiveMode.Create, false))
    {
        Parallel.ForEach(supportingDocLists, options, supportingDoc =>
        {
            if (string.IsNullOrWhiteSpace(supportingDoc.FileURL) || string.IsNullOrWhiteSpace(supportingDoc.FileName))
                return;

            BlobClient blobClient = containerClient.GetBlobClient(supportingDoc.FileURL);
            if (!blobClient.Exists())
                return;

            // 先下载到临时流，避免持有信号量过长时间
            using var tempStream = new MemoryStream();
            blobClient.DownloadTo(tempStream);
            tempStream.Position = 0;

            // 获取信号量，确保同一时间只有一个线程操作ZipArchive
            semaphore.Wait();
            try
            {
                ZipArchiveEntry zipEntry = zipArchive.CreateEntry(supportingDoc.FileName);
                using (Stream entryStream = zipEntry.Open())
                {
                    tempStream.CopyTo(entryStream);
                }
            }
            finally
            {
                // 必须释放信号量，避免死锁
                semaphore.Release();
            }
        });
    }

    zipStream.Position = 0;
    return zipStream.ToArray();
}

关键优化点 #

• 替换ZipArchiveMode.Update为Create模式，符合新ZIP文件的创建场景。
• 确保所有对ZipArchive的操作（创建Entry、写入流）串行执行，要么通过先下载后打包的方式，要么用信号量严格同步。
• 避免在持有同步锁的过程中执行IO操作（比如直接下载到Entry流），防止锁持有时间过长降低并行效率。

内容的提问来源于stack exchange，提问作者mnu-nasir