Parallelisierung Erzeuger und Verbraucher mit internem Zustand

Question

Ich würde gerne wissen, ob der folgende Ansatz ein guter Weg ist, einen Produzenten und Verbraucher Muster in C# .NET 4.6.1

Beschreibung zu implementieren, was ich tun mag:
Ich möchte Dateien lesen, Berechnungen an den Daten durchführen und das Ergebnis speichern. Jede Datei hat einen Ursprung (ein Gerät, z. B. Datenlogger) und abhängig von diesem Ursprung sollten verschiedene Berechnungen sowie Ausgabeformate verwendet werden. Die Datei enthält verschiedene Werte, z. Temperaturanzeigen mehrerer Sensoren. Es ist wichtig, dass die Berechnungen einen Zustand haben. Zum Beispiel könnte dies der letzte Wert der vorherigen Berechnung sein, z. wenn ich alle Werte eines Ursprungs summieren möchte. Ich möchte die Verarbeitung pro Herkunft parallelisieren. Alle Dateien von einem Ursprung müssen sequentiell (oder genauer chronologisch) verarbeitet werden und können nicht parallelisiert werden.

Ich denke, dass der TPL Dataflow eine geeignete Lösung dafür sein könnte.

Dies ist der Prozess, den ich kam:
Der Messwert würde von einem TransformBlock getan werden. Als Nächstes würde ich Instanzen der Klassen erstellen, die Operationen für die Daten für jeden Ursprung ausführen. Sie werden mit den notwendigen Parametern initialisiert, damit sie wissen, wie sie Dateien für ihre Herkunft verarbeiten können. Dann würde ich TransformBlocks für jedes erstellte Objekt erstellen (also grundsätzlich für jeden Ursprung). Jede TransformBlocks würde eine Funktion des entsprechenden Objekts ausführen. Die Lesen der Dateien würde mit einem verknüpft werden, die mit jedem TransformBlock für die Verarbeitung pro Herkunft verknüpft ist. Die linking would be conditional, so dass nur Daten empfangen werden, die die Verarbeitung TranformBlock eines Ursprungs erreichen sollen. Die Ausgabe der Verarbeitungsblöcke würde mit einem ActionBlock zum Schreiben der Ausgabedateien verknüpft werden.
Die maxDegreeOfParallelism wird für jeden Block auf 1 gesetzt.

Ist das eine praktikable Lösung? Ich habe darüber nachgedacht, dies mit Tasks und der BlockingCollection zu implementieren, aber es scheint, dass dies der einfachere Ansatz wäre.

Weitere Informationen:

Die Menge der zu verarbeitenden Dateien groß sein kann in der Größe oder Zahl werden sofort geladen. Lesen und Schreiben sollte gleichzeitig mit der Verarbeitung erfolgen. Da E/A Zeit benötigt und Daten nach der Verarbeitung zu einer Ausgabedatei gesammelt werden müssen, ist eine Pufferung unerlässlich.

Answer 1

Da die Ursprünge sind unabhängig, und die Elemente für jeden Ursprung sind vollständig abhängig dieses Problem eine einfache Lösung hat:

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(); 

Prozess jeder Herkunft sequentiell und Ursprung in parallel.

Wenn Sie eine Notwendigkeit haben, IO zu begrenzen, können Sie in einen SemaphoreSlim werfen, um die IO-Pfade zu schützen.