为什么不用 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？这条错误消息有没有落库？
