如何使用多个ActionBlocks并行处理任务，实现多镜像文件下载？

问题背景 #

需要从多个镜像的慢速共享驱动器下载大型文件夹结构，目标是让单个文件从任意一个镜像下载，同时限制每个镜像的并发线程数。尝试用.NET Dataflow实现，但运行后发现仅第一个镜像对应的ActionBlock收到任务，其他镜像的块完全闲置，需要调整代码实现任务的负载均衡分发。

用户初始代码：

var buffer = new BufferBlock<string>();
var blockOptions = new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = threadsPerPath
};

IEnumerable<ActionBlock<string>> blocks = mirrors.Select(basePath =>
{
    return new ActionBlock<string>(
        file => {
            string destinationFile = Path.Combine(destination, file);
            Directory.CreateDirectory(Path.GetDirectoryName(destinationFile));
            File.Copy(Path.Combine(basePath, file), destinationFile);
        },
        blockOptions);
});

foreach (ActionBlock<string> block in blocks)
{
    buffer.LinkTo(block);
}

await Task.Run(() =>
{
    string top = mirrors[0];
    int baseLength = top.Length;
    IEnumerable<string> allFiles = Directory.EnumerateFiles(top, "*", SearchOption.AllDirectories);
    foreach (string path in allFiles)
    {
        buffer.Post(path[baseLength..]);
    }
    buffer.Complete();
});

问题原因 #

BufferBlock默认采用优先匹配第一个可用目标块的分发策略：只要第一个链接的ActionBlock能接收任务（默认无队列容量限制），所有任务都会优先发给它，后续块无法获得任务分配。

解决方法 #

1. 限制每个ActionBlock的队列容量（最简方案） #

为每个ActionBlock设置BoundedCapacity，使其队列最多容纳与并发线程数相等的任务。当第一个块的队列满时，BufferBlock会自动将任务转发给下一个可用的块，实现负载均衡。

修改后的关键代码：

// 调整块选项，添加队列容量限制
var blockOptions = new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = threadsPerPath,
    BoundedCapacity = threadsPerPath // 队列容量与并发数一致，避免单个块积压过多任务
};

// 链接时启用完成信号传播
foreach (ActionBlock<string> block in blocks)
{
    buffer.LinkTo(block, new DataflowLinkOptions { PropagateCompletion = true });
}

// 等待所有块完成任务（添加到代码末尾）
await Task.WhenAll(blocks.Select(b => b.Completion));

2. 自定义分发逻辑（进阶需求） #

如果需要更灵活的任务分配（比如按文件大小、镜像负载动态调整），可以在BufferBlock后添加一个TransformBlock，手动实现轮询或自定义分发逻辑：

int currentMirrorIndex = 0;
var distributor = new TransformBlock<string, (string, string)>(file =>
{
    // 轮询选择镜像
    var selectedMirror = mirrors[Interlocked.Increment(ref currentMirrorIndex) % mirrors.Count];
    return (selectedMirror, file);
});

// 将分发后的任务发送到对应镜像的ActionBlock
foreach (var mirror in mirrors)
{
    var block = new ActionBlock<(string, string)>(tuple =>
    {
        if (tuple.Item1 != mirror) return;
        // 执行下载逻辑
        string destinationFile = Path.Combine(destination, tuple.Item2);
        Directory.CreateDirectory(Path.GetDirectoryName(destinationFile));
        File.Copy(Path.Combine(mirror, tuple.Item2), destinationFile);
    }, blockOptions);
    distributor.LinkTo(block, t => t.Item1 == mirror, new DataflowLinkOptions { PropagateCompletion = true });
}

// 链接BufferBlock到分发器
buffer.LinkTo(distributor, new DataflowLinkOptions { PropagateCompletion = true });

额外建议 #

• 现成工具替代：无需自行开发，可使用成熟工具实现需求：
  • Robocopy：Windows自带，通过/MT:n开启多线程，配合脚本可实现多镜像分块下载；
  • FreeFileSync：可视化工具，支持多源路径和多线程下载；
  • rclone：跨平台工具，配置多镜像远程后，通过--multi-thread-streams参数实现并行下载。
• 并发线程数优化：共享驱动器的IO瓶颈明显，过多并发会导致拥堵，建议测试2-8之间的线程数找到最优值。
• 代码优化：移除Task.Run直接异步枚举文件，添加异常捕获与重试逻辑，确保所有任务完成后再退出程序。

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