Erzwingen Aufgabe, um die für .NET 4.0 Task Parallel-Bibliotheken

Question

Ich habe ein Programm, das eine Tonne von Sensoren Daten auf einem recht hohen Preis hat produziert, und die Verbraucher, dass Notwendigkeit, sie zu verbrauchen. Die Verbraucher konsumieren zu sehr unterschiedlichen Geschwindigkeiten.

Da ich IObserver / IObservable verwende, war die triviale Lösung, einfach eine Aufgabe für jedes Ereignis zu erstellen und die OnNext () Aufruf und die Daten in einer Lamda wickeln. Das funktionierte sehr gut, ich war überrascht, wie wenig Aufwand es gab über Roh Anrufe.

Das Problem ist, dass einige dieser Verbraucher die Reihenfolge der Ereignisse müssen strikt durchgesetzt werden, und alle Ereignisse nicht verpassen. "PerferFairness" ist nicht gut genug.

Die beste Lösung, die ich mit oben kommen muß, ist stattdessen das Ereignis von Verpackungs / OnNext () Paar, ein Insert in eine ParallelQueue zu wickeln, eine Warteschlange pro Verbraucher und ein Gewinde am anderen Ende der Warteschlange das machen OnNext () Anrufe.

Drei unmittelbare Probleme mit diesem Ansatz. Es ist viel, viel langsamer als die Aufgabe / OnNext () Verpackungslösung. Es gibt keine Sperr Dequeue (oder gibt es?) Für ParallelQueue so die Umsetzung ein bisschen schwierig ist. Der dritter ist, dass dies scheint so ein häufiges Problem, dass ich mich nicht vorstellen kann, gibt es nicht eine Möglichkeit, um zu erzwingen, dass ich verpaßt, vielleicht so etwas wie mehrere Aufgabe Fabriken einen darunter liegenden Pool teilen, jede Fabrik mit einem gewissen Einstellung, die sie streng macht erzwingen bestellen.

Wer die richtige Art und Weise kennen zu erreichen, was ich versuche zu tun?

EDIT: Jede Lösung, die einen Thread pro Verbraucher beinhaltet oder Hersteller nicht funktioniert. Erzeuger / Verbraucher bilden lange Ketten gibt es Hunderte von jedem.

Answer 1

Die TPL Dataflow Bibliothek könnte eine gute Passform für Ihre Anwendung sein. Es erstreckt sich TPL mit einem Datenfluß Paradigma, das eine Bearbeitungs Graph und läuft in einer High-Performance-Ausführungsumgebung konfigurieren können.

TPL Datenfluss ist als Bibliothek oben auf .NET 4 und soll als Teil von .NET 4.5 versenden.

Answer 2

Kein Kommentar über die beste Abstraktion für partielle Ordnung durchzusetzen, aber wenn Sie einen BlockingCollection<T> Wrapper um einen ConcurrentQueue<T> verwenden, das wird Ihnen eine Sperr Take Betrieb dequeue Elemente. z.

// the default is ConcurrentQueue, so you don't have to specify, but if you
// wanted different behavior you could use e.g. ConcurrentStack

var coll = new BlockingCollection<int>(new ConcurrentQueue<int>());

coll.Add(5); // blocks if the collection is at max capacity

int five = coll.Take(); // blocks if the collection is empty