900万XML文件导入SQL Server：多线程读取优化方案问询

核心优化思路 #

当前瓶颈完全在单线程XML读取（耗时4小时），而SqlBulkCopy写入仅需5分钟，优化重点聚焦三点：

1. 替换低效的XmlDocument为流式读取的XmlReader，降低内存占用并提升单文件读取速度
2. 用多线程并行读取多个XML文件，同时控制并发数避免磁盘IO争用
3. 批量处理+批量写入，减少数据库交互开销

具体实现步骤 #

1. 替换XML读取方式

XmlDocument需加载整个XML到内存，900万文件场景下内存压力极大且速度慢，改用XmlReader流式读取：

// 单个XML文件读取方法（流式读取，低内存占用）
public XmlData ParseXml(string filePath)
{
    var data = new XmlData();
    using var reader = XmlReader.Create(filePath);
    
    while (reader.Read())
    {
        if (!reader.IsStartElement()) continue;
        
        switch (reader.LocalName)
        {
            case "Id":
                data.Id = int.TryParse(reader.ReadString(), out var id) ? id : 0;
                break;
            case "UserName":
                data.UserName = reader.ReadString().Trim();
                break;
            // 按你的XML节点结构添加其他字段处理逻辑
        }
    }
    return data;
}

// 对应数据库表的实体类
public class XmlData
{
    public int Id { get; set; }
    public string UserName { get; set; }
    // 其他字段与数据库表列一一对应
}

2. 多线程并行读取+批量写入

使用Parallel.ForEach控制并发，结合线程安全集合批量缓存数据，达到阈值后统一写入：

public void BatchImportXmls(string xmlDir, string connectionString)
{
    // 延迟加载文件路径（避免一次性加载900万条路径到内存）
    var xmlFiles = Directory.EnumerateFiles(xmlDir, "*.xml", SearchOption.AllDirectories);
    
    // 控制最大并发数：建议设为CPU核心数的1-2倍，避免磁盘IO过载
    var parallelOpts = new ParallelOptions
    {
        MaxDegreeOfParallelism = Environment.ProcessorCount * 2
    };

    const int batchThreshold = 1000; // 每1000条数据写入一次
    var dataBatch = new ConcurrentBag<XmlData>();

    Parallel.ForEach(xmlFiles, parallelOpts, filePath =>
    {
        try
        {
            var xmlData = ParseXml(filePath);
            dataBatch.Add(xmlData);

            // 达到批量阈值时写入数据库（加锁避免多线程同时触发写入）
            if (dataBatch.Count >= batchThreshold)
            {
                lock (dataBatch)
                {
                    if (dataBatch.Count >= batchThreshold)
                    {
                        BulkWriteToSql(dataBatch.ToList(), connectionString);
                        dataBatch.Clear();
                    }
                }
            }
        }
        catch (Exception ex)
        {
            // 单独记录错误文件，不中断整体任务
            File.AppendAllText(@"D:\XmlImportErrors.log", $"[{DateTime.Now}] 处理失败 {filePath}: {ex.Message}{Environment.NewLine}");
        }
    });

    // 处理剩余的不足批量阈值的数据
    if (dataBatch.Count > 0)
    {
        BulkWriteToSql(dataBatch.ToList(), connectionString);
    }
}

3. 批量写入SQL Server（复用SqlBulkCopy）

private void BulkWriteToSql(List<XmlData> dataList, string connectionString)
{
    using var conn = new SqlConnection(connectionString);
    conn.Open();

    using var bulkCopy = new SqlBulkCopy(conn)
    {
        DestinationTableName = "dbo.TargetTable", // 你的目标表名
        BatchSize = dataList.Count
    };

    // 映射实体字段与数据库列
    bulkCopy.ColumnMappings.Add(nameof(XmlData.Id), "Id");
    bulkCopy.ColumnMappings.Add(nameof(XmlData.UserName), "UserName");
    // 其他字段映射

    // 将List转换为DataTable（或直接实现IDataReader更高效）
    var dataTable = ConvertToDataTable(dataList);
    bulkCopy.WriteToServer(dataTable);
}

// 辅助方法：将实体List转为DataTable
private DataTable ConvertToDataTable<T>(List<T> dataList)
{
    var dt = new DataTable();
    var props = typeof(T).GetProperties(BindingFlags.Public | BindingFlags.Instance);

    foreach (var prop in props)
    {
        dt.Columns.Add(prop.Name, Nullable.GetUnderlyingType(prop.PropertyType) ?? prop.PropertyType);
    }

    foreach (var item in dataList)
    {
        var row = dt.NewRow();
        foreach (var prop in props)
        {
            row[prop.Name] = prop.GetValue(item) ?? DBNull.Value;
        }
        dt.Rows.Add(row);
    }

    return dt;
}

额外优化建议 #

• 并发数测试：先测试8、16、32等不同并发数，找到磁盘IO不瓶颈的最优值（太多线程会导致磁盘读写排队）
• 生产者-消费者模式：如果读取和写入的速度差异较大，可改用Channel或BlockingCollection分离读取线程和写入线程，避免写入阻塞读取
• 预读缓存：对于机械硬盘，可开启Windows的预读功能，或使用第三方IO库优化磁盘读取
• 错误重试：对读取失败的文件，可添加有限次数的重试逻辑
• 监控进度：添加计数器实时监控处理的文件数、成功数、失败数，方便排查问题

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