我在不同的任务上运行了一个非常典型的生产者/消费者模型。

任务 1:从二进制文件中读取成批的 byte[] 并为每个字节数组集合启动一个新任务。 (出于内存管理目的，该操作是批处理的)。

任务 2-n:这些是工作任务，每个任务都对传入的字节数组集合(来自 Tasks1)进行操作并反序列化字节数组，按特定标准对它们进行排序，然后存储结果对象的集合(每个字节数组在并发字典中反序列化为此类对象。

任务 (n+1) 我选择了一个并发字典，因为这个任务的工作是合并存储在并发字典中的那些集合，其顺序与它们来自 Task1 的顺序相同。我通过从 Task1 一直向下传递到此任务的 collectionID(它的类型为 int 并为 Task1 中的每个新集合递增)来实现这一点。此任务基本上检查下一个预期的 collectionID 是否已存储在并发字典中，如果是，则将其取出，将其添加到最终队列并检查并发字典中的下一个集合。

现在，根据我阅读的内容和观看的视频，在我看来，TPL Dataflow 可能是这种生产者/消费者模型的完美候选者。我似乎无法设计出一个设计并因此开始，因为我从未使用过 TPL Dataflow。在吞吐量和延迟方面，这个库是否能胜任这项任务？我目前在生成的集合中每秒处理 250 万个字节数组和对象。 TPL Dataflow 可以帮助简化吗？我对以下问题的答案特别感兴趣:TPL 数据流能否在产生工作任务时保留来自 Task1 的收集批处理的顺序，并在工作任务完成后重新合并它们？它优化了东西吗？在分析了整个结构之后，我觉得由于旋转和涉及的并发集合太多而浪费了很多时间。

有什么想法、想法吗？

最佳答案

编辑: 结果证明我错了。 TransformBlock 确实按照它们进入的相同顺序返回项目，即使它被配置为并行。因此，我原始答案中的代码完全无用且正常 TransformBlock可以代替使用。

原始答案:

据我所知，.Net 中只有一个并行结构支持按照它们进入的顺序返回已处理的项目:PLINQ with AsOrdered() .但是在我看来，PLINQ 并不能很好地满足您的需求。

另一方面，我认为 TPL 数据流非常适合，但它没有支持并行和同时按顺序返回项目的 block (TransformBlock 支持这两者，但不支持同一时间)。幸运的是，Dataflow block 在设计时考虑到了可组合性，因此我们可以构建自己的 block 来实现这一点。

但首先，我们必须弄清楚如何对结果进行排序。像您建议的那样，使用并发字典以及一些同步机制肯定会起作用。但我认为有一个更简单的解决方案:使用 Task 的队列秒。在输出任务中，您出列了一个 Task ，等待它完成(异步)，当它完成时，您发送它的结果。当队列为空时，我们仍然需要一些同步，但如果我们巧妙地选择使用哪个队列，我们​​可以免费获得同步。

所以，总体思路是这样的:我们正在写的将是一个 IPropagatorBlock ，有一些输入和一些输出。创建自定义的最简单方法 IPropagatorBlock是创建一个处理输入的 block ，另一个产生结果的 block ，并使用 DataflowBlock.Encapsulate() 将它们视为一个 block .

输入 block 必须以正确的顺序处理传入的项目，因此那里没有并行化。它将创建一个新的 Task (实际上，一个 TaskCompletionSource ，以便我们稍后可以设置 Task 的结果)，将其添加到队列中，然后发送项目进行处理，以及一些设置正确 Task 结果的方法。 .因为我们不需要将这个 block 链接到任何东西，所以我们可以使用 ActionBlock .

输出 block 必须采用Task s 从队列中异步等待它们，然后将它们一起发送。但是因为所有的 block 都有一个嵌入其中的队列，并且接受委托(delegate)的 block 有内置的异步等待，这将非常简单:new TransformBlock<Task<TOutput>, TOutput>(t => t) .该 block 将同时用作队列和输出 block 。因此，我们不必处理任何同步。

拼图的最后一 block 实际上是并行处理项目。为此，我们可以使用另一个 ActionBlock , 这次是 MaxDegreeOfParallelism放。它将接受输入，处理它，并设​​置正确的 Task 的结果。在队列中。

放在一起，它可能看起来像这样:

public static IPropagatorBlock<TInput, TOutput>
    CreateConcurrentOrderedTransformBlock<TInput, TOutput>(
    Func<TInput, TOutput> transform)
{
    var queue = new TransformBlock<Task<TOutput>, TOutput>(t => t);

    var processor = new ActionBlock<Tuple<TInput, Action<TOutput>>>(
        tuple => tuple.Item2(transform(tuple.Item1)),
        new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
        });

    var enqueuer = new ActionBlock<TInput>(
        async item =>
        {
            var tcs = new TaskCompletionSource<TOutput>();
            await processor.SendAsync(
                new Tuple<TInput, Action<TOutput>>(item, tcs.SetResult));
            await queue.SendAsync(tcs.Task);
        });

    enqueuer.Completion.ContinueWith(
        _ =>
        {
            queue.Complete();
            processor.Complete();
        });

    return DataflowBlock.Encapsulate(enqueuer, queue);
}

说了这么多，我认为这是相当少的代码。

您似乎非常关心性能，因此您可能需要微调这段代码。例如，设置 MaxDegreeOfParallelism 可能有意义的 processor阻止类似 Environment.ProcessorCount 的内容, 以避免超额认购。此外，如果延迟对您来说比吞吐量更重要，那么设置 MaxMessagesPerTask 可能是有意义的同一 block 的 1(或另一个小数字)，这样当一个项目的处理完成时，它会立即发送到输出。

此外，如果您想限制传入的项目，您可以设置 BoundedCapacity的 enqueuer .

关于c# - 这是 TPL Dataflow 的工作吗？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/11052972/