.NET Core百万级CSV文件同步性能优化方案咨询

你当前的批处理程序处理10万行CSV耗时30分钟，核心瓶颈集中在单条数据的数据库交互逻辑——每个循环都单独创建Scope、查询单条数据、执行单条增改、调用一次SaveChanges。这种细粒度操作在数据量上升后会产生巨大开销，并行反而可能加剧数据库连接池竞争，进一步拖慢效率。以下是针对性的优化方案：

1. 重构数据转换与去重逻辑，消除线程阻塞 #

原代码用Parallel.ForEach转换实体时，因List<InvoiceDb>非线程安全而加锁，导致严重的线程阻塞，完全抵消并行收益。可以直接在转换阶段完成去重，避免无效数据流转：

// 直接转换+去重，跳过全量List再分组的冗余步骤
var distinctInvoices = data
    .Select(invoice => new InvoiceDb
    {
        Id_Invoice = invoice.IdInvoice,
        DateCheck = invoice.DateCheck
    })
    .GroupBy(c => c.Id_Invoice)
    .Where(g => g.Count() == 1)
    .Select(g => g.First())
    .ToList();

// 提取重复ID（无需生成完整重复实体）
var duplicateIds = data
    .GroupBy(c => c.IdInvoice)
    .Where(g => g.Count() > 1)
    .Select(g => g.Key)
    .Distinct()
    .ToList();

if (duplicateIds.Any())
{
    Log.Warning("重复发票ID: {duplicata}", duplicateIds);
}

2. 批量数据库操作，替代单条CRUD（核心优化） #

把“查单条→增/改单条→单条保存”改成“批量查询存在项→批量新增+批量更新→单次保存”，彻底减少数据库交互次数：

步骤1：批量查询已存在的发票ID #

一次性获取所有待处理ID的存在状态，避免N次单条查询：

using var scope = _serviceProvider.CreateScope();
var service = scope.ServiceProvider.GetRequiredService<IInvoiceService>();
var dbContext = service.GetDbContext(); // 假设服务可直接获取DbContext

var invoiceIds = distinctInvoices.Select(i => i.Id_Invoice).ToList();
// 批量查询已存在的ID
var existingIds = await dbContext.Invoices
    .Where(i => invoiceIds.Contains(i.Id_Invoice))
    .Select(i => i.Id_Invoice)
    .ToListAsync();

步骤2：拆分新增与更新集合 #

// 新增集合：数据库中不存在的发票
var toAdd = distinctInvoices.Where(i => !existingIds.Contains(i.Id_Invoice)).ToList();

// 更新集合：数据库中已存在的发票
var toUpdate = distinctInvoices.Where(i => existingIds.Contains(i.Id_Invoice)).ToList();

步骤3：批量插入与更新 #

批量插入

直接添加集合后单次SaveChanges：

if (toAdd.Any())
{
    dbContext.Invoices.AddRange(toAdd);
}

批量更新

用EF Core 7+原生ExecuteUpdate实现批量更新（旧版EF可循环更新实体但只调用一次SaveChanges）：

if (toUpdate.Any())
{
    var updateDict = toUpdate.ToDictionary(i => i.Id_Invoice, i => i.DateCheck);
    
    await dbContext.Invoices
        .Where(i => updateDict.ContainsKey(i.Id_Invoice))
        .ExecuteUpdateAsync(setters => setters
            .SetProperty(i => i.DateCheck, i => updateDict[i.Id_Invoice])
        );
}

// 仅调用一次SaveChanges，减少数据库事务提交开销
await dbContext.SaveChangesAsync();

3. 取消不必要的Parallel.ForEachAsync #

数据库是IO密集型资源，过度并行会导致连接池耗尽、锁竞争加剧。批量操作的效率远高于并行单条操作，因此无需再用Parallel.ForEachAsync处理数据库逻辑。

4. 其他性能优化建议 #

• 流式读取CSV：不要一次性将300万行读进内存，用CsvHelper的GetRecordsAsync配合异步流读取，边读边处理，降低内存占用。
• 数据库索引优化：给InvoiceDb表的Id_Invoice字段添加唯一索引，既加速查询，又能在数据库层面避免重复插入。
• 连接池配置：在连接字符串中调整Max Pool Size（例如设为100），避免数据库连接不足：

  "ConnectionStrings": {
    "DefaultConnection": "Server=...;Database=...;Max Pool Size=100;"
  }
  

• 批量发送Kafka：将PublishInvoice改为批量发送，比如每1000条数据发一次，减少Kafka交互开销。
• 监控调优：用SQL Server Profiler查看慢查询，用.NET Diagnostics工具分析内存、CPU瓶颈，定位剩余性能卡点。

重构后核心代码示例 #

public async Task<bool> ImportDataFromCsvFile()
{
    try
    {
        // 流式读取CSV（替换为你的Azure Blob读取逻辑）
        await using var stream = await _azureBlobService.GetCsvStreamAsync();
        using var reader = new StreamReader(stream);
        using var csv = new CsvReader(reader, CultureInfo.InvariantCulture);
        
        // 异步读取+实时去重，避免全量加载内存
        var distinctInvoices = new Dictionary<string, InvoiceDb>();
        var duplicateIds = new HashSet<string>();
        
        await foreach (var csvInvoice in csv.GetRecordsAsync<CsvInvoice>())
        {
            if (distinctInvoices.ContainsKey(csvInvoice.IdInvoice))
            {
                duplicateIds.Add(csvInvoice.IdInvoice);
                continue;
            }
            
            distinctInvoices.Add(csvInvoice.IdInvoice, new InvoiceDb
            {
                Id_Invoice = csvInvoice.IdInvoice,
                DateCheck = csvInvoice.DateCheck
            });
        }

        if (duplicateIds.Any())
        {
            Log.Warning("重复发票ID: {duplicata}", duplicateIds);
        }

        var invoiceList = distinctInvoices.Values.ToList();
        if (!invoiceList.Any())
        {
            Log.Information("无有效数据需要导入");
            await PublishInvoiceInBatches(invoiceList);
            return true;
        }

        // 批量数据库操作
        using var scope = _serviceProvider.CreateScope();
        var service = scope.ServiceProvider.GetRequiredService<IInvoiceService>();
        var dbContext = service.GetDbContext();

        var invoiceIds = invoiceList.Select(i => i.Id_Invoice).ToList();
        var existingIds = await dbContext.Invoices
            .Where(i => invoiceIds.Contains(i.Id_Invoice))
            .Select(i => i.Id_Invoice)
            .ToListAsync();

        var toAdd = invoiceList.Where(i => !existingIds.Contains(i.Id_Invoice)).ToList();
        var toUpdateDict = invoiceList.Where(i => existingIds.Contains(i.Id_Invoice))
            .ToDictionary(i => i.Id_Invoice, i => i.DateCheck);

        if (toAdd.Any())
        {
            dbContext.Invoices.AddRange(toAdd);
        }

        if (toUpdateDict.Any())
        {
            await dbContext.Invoices
                .Where(i => toUpdateDict.ContainsKey(i.Id_Invoice))
                .ExecuteUpdateAsync(setters => setters
                    .SetProperty(i => i.DateCheck, i => toUpdateDict[i.Id_Invoice])
                );
        }

        await dbContext.SaveChangesAsync();
        Log.Information("{result} 条数据导入成功", invoiceList.Count);

        // 批量发送Kafka
        await PublishInvoiceInBatches(invoiceList);
        return true;
    }
    catch (Exception ex)
    {
        Log.Error("数据导入失败: {errorMessage}", ex.Message);
        throw;
    }
}

// Kafka批量发送辅助方法
private async Task PublishInvoiceInBatches(List<InvoiceDb> invoices, int batchSize = 1000)
{
    for (int i = 0; i < invoices.Count; i += batchSize)
    {
        var batch = invoices.Skip(i).Take(batchSize).ToList();
        await PublishInvoice(batch); // 假设PublishInvoice支持批量发送
    }
}

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