因为Parallel.ForEach缺乏缓冲、背压和异步支持，而TPL Dataflow通过BufferBlock、TransformBlock等可建模多阶段异步处理，支持BoundedCapacity控制内存、async/await避免阻塞、错误捕获不丢失及资源安全释放。
为什么不用 Parallel.ForEach 而选 TPL Dataflow
因为文件处理常含多个异步阶段（读取 → 解析 → 转换 → 写入），且各阶段吞吐量不均，Parallel.ForEach 无法天然缓冲、背压或动态调节。TPL Dataflow 的 BufferBlock<t></t>、TransformBlock<tin></tin> 和 ActionBlock<t></t> 能显式建模这些环节，避免内存暴涨或 I/O 阻塞拖垮整个流程。
常见错误是把所有逻辑塞进一个 TransformBlock：比如在读取文件后直接解析 JSON 并写入数据库——这会让阻塞操作（如 File.ReadAllText）拖慢整个数据流，也难单独监控某环节耗时。
- 读取阶段必须用
async/await，推荐File.ReadAllBytesAsync或Stream.CopyToAsync，而非同步 API - 每个
TransformBlock应只做一件事：例如「路径 → 文件字节」、「字节 → 对象」、「对象 → SQL 参数」 - 设置
MaxDegreeOfParallelism时注意：磁盘 I/O 密集型任务通常设为 2–4；CPU 密集型（如解密、校验）可设为Environment.ProcessorCount
如何让文件读取块真正异步且可控
TransformBlock<string byte></string> 是起点，但若直接调用 File.ReadAllBytes，它会在线程池中同步阻塞，违背异步初衷。正确做法是封装为 Task<byte></byte> 并启用 Task.Run（仅当必须用同步 API 时）或优先使用原生异步方法。
var readBlock = new TransformBlock<string, byte[]>(
async filePath =>
{
// ✅ 原生异步
await using var stream = File.OpenRead(filePath);
var buffer = new byte[(int)stream.Length];
await stream.ReadExactlyAsync(buffer); // .NET 5+
return buffer;
},
new ExecutionDataflowBlockOptions
{
MaxDegreeOfParallelism = 3,
BoundedCapacity = 10 // 防止大量小文件瞬间占满内存
});
注意：BoundedCapacity 必须设——否则上游快速投递文件路径，而下游读取慢时，未处理的路径会无限堆积在内存里。
- 不要用
File.ReadAllTextAsync处理大文件（可能触发 LOH 分配），改用流式读取 +StreamReader.ReadLineAsync - 若需按行处理 CSV/日志，用
TransformManyBlock<string string></string>拆分后再分发，避免单个大数组驻留内存 - 捕获
IOException并通过Post到错误处理块，别让异常终止整个 dataflow
如何连接多个块并确保错误不丢失
用 LinkTo 连接时，默认是“尽最大努力转发”，失败项会被丢弃。文件处理中，任何环节出错（如 JSON 格式错误、数据库连接中断）都必须捕获并记录，否则等于静默丢数据。
正确方式是启用 propagateCompletion = false，手动控制完成信号，并为每个块配置错误回调：
var parseBlock = new TransformBlock<byte[], MyRecord>(
data => JsonSerializer.Deserialize<MyRecord>(data),
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = Environment.ProcessorCount });
// 链接时指定 predicate，把异常转给错误块
parseBlock.LinkTo(errorBlock, new DataflowLinkOptions { PropagateCompletion = false },
_ => false); // 所有输出都走默认链，异常由 try-catch 捕获后 Post 到 errorBlock
// 在 transform 函数内处理异常
var safeParseBlock = new TransformBlock<byte[], (MyRecord?, Exception?)>(
data =>
{
try
{
var obj = JsonSerializer.Deserialize<MyRecord>(data);
return (obj, null);
}
catch (Exception ex)
{
return (null, ex);
}
});
- 不要依赖
Completion.ContinueWith来清理资源，改用try/finally或IDisposable块内释放（如FileStream） - 完成信号要等所有块都
Complete()后再Completion.WaitAsync()，否则可能漏掉最后几条数据 -
BatchBlock<t></t>适合聚合小文件写入 DB，但注意TriggerBatchSize和超时需配合业务节奏（如每 100 条或 5 秒 flush 一次）
实际部署时最易忽略的内存与生命周期问题
本地测试跑得通，一上生产就 OOM 或 GC 频繁——大概率是没控制住对象生命周期。文件内容（尤其是大文件字节数组）、反序列化后的对象、临时 StringBuilder 等都容易长期驻留 Gen2 或 LOH。
- 避免在
TransformBlock中缓存文件路径字符串以外的引用：比如把FileStream存在闭包里，它不会随 block 完成自动释放 - 对大文件，用
ArrayPool<byte>.Shared.Rent</byte>替代new byte[...]，并在处理完Return -
ExecutionDataflowBlockOptions中的CancellationToken必须传入，以便在服务关闭时主动Complete()并等待Completion，而不是粗暴中止导致文件写一半 - 不要让
ActionBlock<t></t>直接调用Console.WriteLine或短生命周期 logger —— 日志组件可能被 GC 提前回收，改用静态 logger 实例
真正难的不是搭起管道，而是让每个块在高压下不泄漏、不卡死、不静默失败。从第一行代码开始就得想清楚：这个数组谁释放？这个 stream 是否已 dispose？这条错误消息有没有落库？
