嵌套Parallel.ForEach循环与文件生成问题：基于TPL批量处理千万级数据

Hey，处理1000万条患者事件生成独立文件这个场景，用TPL确实是个高效的方向，但两层嵌套Parallel.ForEach很容易踩IO竞争、线程过载的坑，我来给你梳理下问题点和优化方案，顺便给你个更健壮的代码参考。

• 磁盘IO竞争过载：内层循环同时触发大量文件写入，磁盘（尤其是HDD）的IO能力是瓶颈，过多并行线程反而会因为磁盘队列阻塞拖慢整体速度，甚至出现文件写入失败、损坏的情况。
• 目录创建冲突：如果多个线程同时为同一个患者创建d:\EVENTS\PATIENTID目录，哪怕Directory.CreateDirectory有幂等性，高并发下也可能抛出异常。
• 线程调度浪费：两层并行会创建远超CPU核心数的线程，导致上下文切换频繁，反而消耗更多资源。

1. 调整并行粒度，避免双层并行 #

与其两层都并行，不如外层按患者维度并行，内层串行处理该患者的所有事件：同一个患者的文件都在同一路径下，串行写能避免目录操作冲突，同时控制单患者的IO压力，让磁盘更高效地顺序处理。

2. 预创建所有患者目录 #

在开始并行处理前，先批量创建好所有需要的患者目录，这样处理事件时就不用反复判断目录是否存在，彻底避免并发创建的冲突，也减少了额外的IO操作。

3. 手动控制并行度 #

别用Parallel.ForEach的默认最大并行数，根据你的磁盘类型（SSD/HDD）和CPU核心数设置合理值：比如SSD可以设为Environment.ProcessorCount * 2，HDD建议设为和CPU核心数一致，避免线程过载。

4. 增加错误处理与重试 #

生成百万级文件难免遇到IO异常（磁盘空间不足、权限问题等），一定要加重试逻辑，同时记录失败的记录，方便后续补处理。

5. 可选：结合异步IO提升性能 #

如果用的是SSD，异步IO能进一步挖掘性能，但要注意.NET 6+支持的Parallel.ForEachAsync，避免在并行循环里阻塞线程。

using System;
using System.Collections.Generic;
using System.IO;
using System.Threading.Tasks;

public class PatientEventFileGenerator
{
    private const string BaseOutputPath = @"d:\EVENTS";

    public void GenerateEventFiles(IEnumerable<Patient> patientList)
    {
        // 第一步：预创建所有患者目录，避免并行时的目录冲突
        var uniquePatientIds = new HashSet<int>();
        foreach (var patient in patientList)
        {
            uniquePatientIds.Add(patient.Id);
        }

        foreach (var patientId in uniquePatientIds)
        {
            var patientDir = Path.Combine(BaseOutputPath, patientId.ToString());
            Directory.CreateDirectory(patientDir); // 幂等操作，提前创建好所有目录
        }

        // 第二步：按患者维度并行处理，内层串行写文件
        var parallelSettings = new ParallelOptions
        {
            // 根据磁盘性能调整：SSD可以设为CPU核心数*2，HDD建议和CPU核心数一致
            MaxDegreeOfParallelism = Environment.ProcessorCount
        };

        Parallel.ForEach(patientList, parallelSettings, patient =>
        {
            var patientDir = Path.Combine(BaseOutputPath, patient.Id.ToString());
            foreach (var eventRecord in patient.Events)
            {
                var eventFilePath = Path.Combine(patientDir, $"{eventRecord.Id}.txt");
                try
                {
                    // 写入事件内容，若需要异步可改用File.WriteAllTextAsync（配合Parallel.ForEachAsync）
                    File.WriteAllText(eventFilePath, eventRecord.EventContent);
                }
                catch (IOException ex)
                {
                    Console.WriteLine($"写入文件失败：{eventFilePath}，错误信息：{ex.Message}");
                    // 带指数退避的重试逻辑
                    RetryWriteEventFile(eventFilePath, eventRecord.EventContent);
                }
            }
        });
    }

    private void RetryWriteEventFile(string filePath, string content, int maxRetries = 3)
    {
        for (int retry = 0; retry < maxRetries; retry++)
        {
            try
            {
                File.WriteAllText(filePath, content);
                return;
            }
            catch (IOException)
            {
                // 指数退避等待，避免频繁重试加剧IO压力
                Task.Delay(100 * (int)Math.Pow(2, retry)).Wait();
            }
        }
        Console.WriteLine($"重试{maxRetries}次后仍失败：{filePath}");
    }
}

// 假设的实体类，可根据你的实际结构调整
public class Patient
{
    public int Id { get; set; }
    public IEnumerable<EventRecord> Events { get; set; }
}

public class EventRecord
{
    public int Id { get; set; }
    public string EventContent { get; set; }
}

• 如果数据库查询是分批获取的（比如每次取1000个患者），可以结合分批查询和并行处理，避免一次性加载1000万条记录到内存导致溢出。
• 对于超大规模的数据，还可以考虑分批次输出，比如每处理1000个患者就做一次资源回收，避免内存占用过高。

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