并行处理Azure Blob中Zip文件时遇文件头损坏错误的排查

问题场景 #

Azure Blob存储中存放包含A、B两类文件的Zip包，需要高效并行读取并处理其中文件。使用指定代码时随机出现部分文件处理失败，错误信息为A local file header is corrupt（例如10个文件中3个成功、7个失败）。

问题代码 #

private static async Task DownloadFromStream(BlobClient blobClient)
{
    using var archive = new ZipArchive(await blobClient.OpenReadAsync(
        new BlobOpenReadOptions(false)));

    // 获取不同类型文件
    var typeAFiles = GetFiles(archive, "TYPE_A");
    var typeBFiles= GetFiles(archive, "TYPE_B");

    // 尝试并行处理两类文件
    var fileProcessTasks = new List<Task>
        {
            Task.Run(async () => await ProcessData(typeAFiles)),
            Task.Run(async () => await ProcessData(typeBFiles))
        };

    await Task.WhenAll(fileProcessTasks);
}

private static async Task ProcessData(IEnumerable<ZipArchiveEntry> files)
{
    foreach (var entry in files)
    {
        try
        {
            var content = await new System.IO.StreamReader(entry.Open(),
                Encoding.UTF8).ReadToEndAsync();
            // Console.WriteLine(content.Length);
            // 处理逻辑
            Console.WriteLine("*************");
        }
        catch (Exception e)
        {
            Console.WriteLine(e);
        }
    }
}

private static IEnumerable<ZipArchiveEntry> GetFiles(ZipArchive archive,
    string fileType)
{
    return archive.Entries.Where(y => y.Name.StartsWith(fileType));
}

备注：可正常运行的代码（已修正语法错误） #

var tasks = new List<Func<Task>>
        {
            () => ProcessData(typeAFiles),
            () => ProcessData(typeBFiles)
        };
        
await Task.WhenAll(tasks.AsParallel().Select(async task => await task()));

问题分析 #

1. IEnumerable延迟执行引发并发冲突 #

GetFiles返回的是Linq查询结果（IEnumerable<ZipArchiveEntry>），这类集合属于延迟执行：只有在foreach遍历阶段，才会真正去枚举archive.Entries并执行过滤逻辑。

2. ZipArchive与底层流不支持多线程并发 #

ZipArchive及其依赖的Blob读取流不是线程安全的。当两个Task.Run线程同时启动ProcessData中的foreach时，会在不同线程同时操作底层流：

• 一个线程枚举archive.Entries时会移动流的读写指针
• 另一个线程读取文件内容也会修改指针位置
  这种并发修改会导致流读取到错误的字节数据，最终触发A local file header is corrupt错误。

解决方案 #

1. 提前枚举结果到内存集合 #

将延迟执行的IEnumerable转为List<ZipArchiveEntry>，让枚举和过滤操作在主线程完成，后续并行处理的是内存中的只读集合，彻底避免对ZipArchive底层流的并发操作：

// 修改获取文件的代码，提前转为List
var typeAFiles = GetFiles(archive, "TYPE_A").ToList();
var typeBFiles= GetFiles(archive, "TYPE_B").ToList();

2. 优化并行处理逻辑 #

• 无需用Task.Run包装异步方法：ProcessData本身是IO密集型异步操作，直接调用即可，避免不必要的线程切换
• 可细化到单文件并行，进一步提升处理效率：

private static async Task DownloadFromStream(BlobClient blobClient)
{
    using var archive = new ZipArchive(await blobClient.OpenReadAsync(
        new BlobOpenReadOptions(false)));

    var typeAFiles = GetFiles(archive, "TYPE_A").ToList();
    var typeBFiles = GetFiles(archive, "TYPE_B").ToList();

    // 合并所有文件，并行处理每个文件
    var allProcessTasks = typeAFiles.Concat(typeBFiles)
                                    .Select(ProcessSingleEntry);

    await Task.WhenAll(allProcessTasks);
}

// 单独处理单个Zip条目，职责更清晰
private static async Task ProcessSingleEntry(ZipArchiveEntry entry)
{
    try
    {
        using var reader = new StreamReader(entry.Open(), Encoding.UTF8);
        var content = await reader.ReadToEndAsync();
        // 处理逻辑
        Console.WriteLine("*************");
    }
    catch (Exception e)
    {
        Console.WriteLine(e);
    }
}

3. 关于备注代码的说明 #

原备注代码存在语法错误（无法直接将Task赋值给Func<Task>），修正后能运行的原因是AsParallel()的调度可能让两个任务的执行没有真正并发枚举流，但这不是可靠的解决方案。只有提前将延迟查询转为内存集合，才能从根本上避免流的并发访问冲突。

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