如何线程安全地收集C# ActionBlock异步任务的执行结果？

这个问题我之前也碰到过！ActionBlock确实适合处理串行（或并行）的异步任务，但它的设计就是只执行操作不返回结果，所以要收集结果的话，有两种比较优雅的方式：

方法一：用TransformBlock替代ActionBlock（推荐，符合TPL DataFlow设计） #

TransformBlock<TInput, TOutput>是ActionBlock的“升级版”，它既可以处理输入任务，还能返回处理后的结果，而且内部已经帮你处理好了线程安全的结果传递。

假设你的Process(int)方法返回Task<string>，示例代码如下：

// 定义TransformBlock：输入int，输出Process的结果类型
var transformBlock = new TransformBlock<int, string>(async i => await Process(i));

// 用BufferBlock来接收并存储结果，也可以直接后续处理
var resultsBuffer = new BufferBlock<string>();

// 链接两个块，设置PropagateCompletion让完成信号传递下去
transformBlock.LinkTo(resultsBuffer, new DataflowLinkOptions { PropagateCompletion = true });

// 发布任务
for(int i = 0; i < 100; i++)
{
    transformBlock.Post(i);
}

// 标记任务发布完成
transformBlock.Complete();
// 等待所有任务处理完毕
await transformBlock.Completion;

// 从BufferBlock收集所有结果
var finalResults = new List<string>();
while(resultsBuffer.TryReceive(out var singleResult))
{
    finalResults.Add(singleResult);
}

这种方法的好处是：

• 完全遵循TPL DataFlow的组件设计，不需要自己手动处理线程安全
• 默认会保持结果和输入的顺序一致（即使你设置了MaxDegreeOfParallelism并行处理，只要不修改TransformBlock的EnsureOrdered参数，结果顺序还是和输入对应）

方法二：保留ActionBlock，用线程安全集合收集结果 #

如果你不想替换现有ActionBlock，可以直接用.NET提供的线程安全集合来存储结果，比如ConcurrentBag<T>（专门为多线程添加场景设计，所有操作都是线程安全的）。

示例代码：

// 初始化线程安全的结果集合，替换成你的Process返回类型
var results = new ConcurrentBag<string>();

var actionBlock = new ActionBlock<int>(async i => 
{
    // 执行异步任务并获取结果
    var taskResult = await Process(i);
    // 直接添加到ConcurrentBag，无需额外锁
    results.Add(taskResult);
});

// 发布任务
for(int i = 0; i < 100; i++)
{
    actionBlock.Post(i);
}

actionBlock.Complete();
await actionBlock.Completion;

// 此时results中就是所有任务的结果，注意：如果ActionBlock设置了并行处理，结果顺序可能和发布顺序不一致

如果需要保持结果顺序，记得不要给ActionBlock设置MaxDegreeOfParallelism（默认是1，串行处理），或者改用方法一。

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