将生产者和消费者与内部状态并行化

Question

我想知道以下方法是否是在 C# .NET 4.6.1 中实现生产者和消费者模式的好方法

我想做什么的描述：
我想读取文件，对其中的数据进行计算并保存结果。每个文件都有一个来源（一个设备，例如数据记录器），并且根据该来源，应该使用不同的计算和输出格式。该文件包含不同的值，例如几个传感器的温度读数。重要的是计算有一个状态。例如，这可能是先前计算的最后一个值，例如如果我想对一个来源的所有值求和。 我想并行化每个来源的处理。来自一个来源的所有文件都需要按顺序（或更具体地按时间顺序）处理，并且不能并行化。

我认为 TPL 数据流可能是一个合适的解决方案。

这是我想出的过程：
阅读将由TransformBlock 完成。接下来，我将创建对每个来源的数据执行操作的类的实例。他们使用必要的参数进行初始化，以便他们知道如何处理文件的来源。 然后我会为每个创建的对象创建TransformBlocks（基本上是每个来源）。每个TransformBlocks 都会执行对应对象的一个​​函数。 TransformBlock 读取文件 将链接到BufferBlock，该BufferBlock 链接到每个TransformBlock 以进行每个来源的处理。 linking would be conditional，以便只接收旨在到达源的处理TranformBlock 的数据。处理块的输出将与ActionBlock 链接，用于编写输出文件。
对于每个 Block，maxDegreeOfParallelism 设置为 1。

这是一个可行的解决方案吗？我考虑过使用 Tasks 和 BlockingCollection 来实现这一点，但似乎这将是更简单的方法。

附加信息：

处理的文件数量可能很大或一次加载的数量很大。 读取和写入应该与处理同时发生。由于 I/O 需要时间，并且需要在处理后收集数据以形成输出文件，因此缓冲是必不可少的。

Answer 1

由于来源是独立的，并且每个来源的项目完全依赖，这个问题有一个简单的解决方案：

var origins = (from f in files
               group f by f.origin into g
               orderby g.Count() descending
               select g);

 var results =
 Partitioner.Create(origins) //disable chunking
 .AsParallel()
 .AsOrdered() //try process the biggest groups first
 .Select(originGroup => {
  foreach (var x in originGroup.OrderBy(...)) Process(x);
  return someResult;
 })
 .ToList();

按顺序处理每个原点并并行处理。

如果您需要限制 IO，可以通过某种方式输入 SemaphoreSlim 来保护 IO 路径。