TPL数据流 - 控制任何时候在流中的项目

Question

我需要处理商家发送给我们的文件。每个文件名都包含商家的ID，例如：Products - 118763.csv，其中数字是商家ID。

我想设置一个数据流来处理这些文件。可以同时处理多个文件。

然而，必须按顺序处理每个单独商家的文件。也就是说，不能同时处理特定商家的两个文件。

我不确定如何用数据流建模。我可以监视当前正在处理哪些商家，但我不知道如何确保我不会同时处理同一商家的两个文件，同时尽可能处理尽可能多的文件。

似乎我需要过滤队列中的内容，但我每次只能获取一个项目，一旦获取到，接下来该怎么做呢？我可以重新排队，但如果只有两个文件，它们都来自同一商家，那将意味着大量的重新排队。

有什么好的想法吗？

- bornfromanegg

---

可能是Hashed/Sharded ActionBlocks的重复问题。 - i3arnon

Answer 1

您可以创建多个 ActionBlock（尽可能多地并行处理），每个 ActionBlock 一次仅处理一个文件，并使用文件名中的数字选择正确的 ActionBlock。这将允许并行处理，同时确保具有相同商家 ID 的文件按顺序处理。

初始化：

_actionBlocks = new ActionBlock<File>[DegreeOfParallelism];
for (var i = 0; i < _actionBlocks.Length; i++)
{
    _actionBlocks[i] = new ActionBlock<File>(file => ProcessFile(file));
}

使用方法：

void ProcessFile(string path)
{
    _actionBlocks[(uint) int.Parse(Path.GetFileNameWithoutExtension(path))%_actionBlocks.Length].Post(
        ReadFile(path));
}

- i3arnon