从3+天到3.8小时：为 SQL Server 扩展 .NET CSV 导入器

Source: Dev.to

“足够好”却不够的解决方案

每个项目都有那项任务：“只要把这个巨大的 CSV 加载到数据库一次，就完成了。”
这就是我的起点。我有一个 40 GB CSV 文件，里面包含复杂的嵌套 JSON 结构（模型、图表、热点、元素），需要解析后保存到规范化的 SQL Server 架构中。

导入器的第一个版本非常直接：

1. 读取一行。
2. 解析 JSON。
3. 创建实体。
4. 保存到数据库。

它能够工作，但耗时 超过 3 天（约 92 小时） 才能完成。对于一次性迁移来说，这虽然痛苦但还能接受——你在星期五启动它，期望在星期一完成。

然后需求变了

业务决定这不是一次性事件。我们需要加载 多个类似大小的文件，并可能定期更新它们。突然之间，3 天的运行时间成了阻碍。一次性排队加载文件将需要数周时间，导致分析和开发陷入瘫痪。原本天真的顺序导入器不再只是慢——它已经无法满足新的工作流需求。

挑战：为什么这么慢？

解析并插入数据听起来很简单，但在大规模（40 GB，数百万复杂对象）时，“标准”方法会遇到硬性限制：

• 顺序处理 – 逐行读取并逐个解析 JSON，使 CPU 在等待数据库时空闲，反之亦然。
• 数据库往返 – 单独（或小批量）保存实体导致巨大的开销。数据库花在管理事务和网络调用的时间比实际存储数据的时间还多。
• 内存压力 – 为每行加载完整的 JsonDocument 对象，导致巨大的垃圾回收压力。
• 脆弱性 – 处理两天后出现的单个错误就可能导致整个管道崩溃，必须重新启动。

解决方案：高性能架构

为了满足新的“多文件”需求，我重新设计了系统，使其 并行、批处理且具备弹性。

1. 使用 SemaphoreSlim 实现受控并行

我没有使用单线程，而是采用了生产者‑消费者模式，利用 SemaphoreSlim 将并发度限制在 8 个并行工作者。

• 原因： 这样可以恰到好处地饱和 CPU 和数据库连接池，既快又不会让服务器卡死。若使用无限并行（Parallel.ForEach），会导致数据库性能崩溃。
• 安全性： 每个工作者通过 IDbContextFactory 获取自己的 DbContext，从而在不产生锁竞争的情况下保证线程安全。

2. 通过 EF Core 批量插入（关键收益）

这是最关键的改动。原来的循环中使用 context.Add(entity); context.SaveChanges();，现在改为 在内存中累计实体并以 100 条以上为一批进行刷新。

• 影响： 网络往返次数减少约 100 倍，事务日志开销显著下降。

3. 架构与 SOLID 原则

为了保持代码可维护，我将解析逻辑拆分为独立的 Processors，每个 Processor 负责 JSON 的特定部分（例如 ModelProcessor、DiagramProcessor）。

• SRP（单一职责原则）： 每个 Processor 只处理其所属的领域切片。
• DIP（依赖倒置原则）： 高层服务依赖抽象（IEntityFactory、IUnitOfWork），使系统易于测试和扩展。

4. 可靠性特性

• 重试策略： 对瞬时数据库错误（死锁、超时）最多重试 25 次。
• 优雅降级： 若某个 Processor 在处理错误数据时失败，会记录错误并继续执行，而不是让整个导入过程崩溃。
• 优化解析： 切换为 JsonElement 和 TryGetProperty，实现更快、低分配的 JSON 遍历。

结果：提升 24 倍

性能提升巨大，将“周末任务”变成了“午休任务”。

关键要点

• 上下文很重要： 3 天的脚本一次运行没问题。如果需要重复运行就致命。始终问自己，“我们会多频繁运行它？”
• 批处理是王道： 在 EF Core 中，从单条插入改为批处理往往是你能做的最有效的性能提升。
• 并行需要限制： 向 SQL Server 投入 100 个线程只会让它变慢。找到“甜 spot”（例如 8 个工作线程）是关键。
• 弹性是一项特性： 长时间运行时，网络会出现短暂中断，死锁会发生。重试策略把崩溃变成轻微的日志警告。

Future Plans

• 添加全面的测试（xUnit + Moq），实现 85 %+ 的覆盖率，涵盖所有处理器。
• 对各个管道阶段进行性能分析，以找出下一个瓶颈（可能是 JSON 解析的 CPU 时间）。
• 将配置（批处理大小、线程数）公开，以便根据不同服务器规格动态调整。

代码仓库

查看代码： GitHub 仓库链接

[GitHub repository](https://github.com/belochka1-04/ParsCsvSaveInDb)