TPL Dataflow und Rx Kombiniertes Beispiel

Question

Ich will nur beide lernen und wie man sie zusammen benutzt. Ich verstehe, dass sie sich gegenseitig ergänzen können Ich konnte einfach kein Beispiel dafür finden, dass jemand es tatsächlich tut.

Answer 1

Lassen Sie mich mit ein wenig Hintergrund beginnen.

Der .NET-Framework verfügt über eine Reihe von speziellen Typen - entweder geeigneter Klassen oder Schnittstellen - Task<T>, IObservable<T>, Nullable<T>, IEnumerable<T>, Lazy<T>, usw. -, die T auf den zugrunde liegenden Typen besondere Kräfte zur Verfügung stellen.

Die TPL verwendet Task<T>, um die asynchrone Berechnung eines einzelnen Werts von T darzustellen.

Rx verwendet IObservable<T> zur asynchronen Berechnung von null oder mehr Werten von T.

Es ist der Aspekt der "asynchronen Berechnung" von beiden, der TPL und Rx zusammenbringt.

Nun wurde der TPL verwendet auch die Art Task die asynchrone Ausführung eines Action Lambda darzustellen, dies kann jedoch einen Sonderfall von Task<T> in Betracht gezogen werden, wo Tvoid ist. Sehr viel wie eine Standard-Methode in C# gibt void wie so:

public void MyMethod() { } 

Rx auch für den gleichen Sonderfall unter Verwendung eines speziellen Typ Unit genannt ermöglichen.

Der Unterschied zwischen TPL und Rx liegt in der Anzahl der zurückgegebenen Werte. TPL ist eins und nur eins, während Rx Null oder mehr ist.

Wenn Sie also Rx auf eine spezielle Weise behandeln, indem Sie nur mit beobachtbaren Sequenzen arbeiten, die einen einzelnen Wert zurückgeben, können Sie einige Berechnungen in ähnlicher Weise wie die TPL durchführen.

Zum Beispiel in der TPL kann ich schreiben:

Task.Factory 
    .StartNew(() => "Hello") 
    .ContinueWith(t => Console.WriteLine(t.Result)); 

Und in Rx wäre die äquivalent sein:

Observable 
    .Start(() => "Hello") 
    .Subscribe(x => Console.WriteLine(x)); 

ich durch die Angabe einen Schritt weiter in Rx gehen könnte, dass die TPL sollte verwendet werden, um die Berechnung wie so auszuführen: (. Standardmäßig ist der Thread-Pool verwendet)

Observable 
    .Start(() => "Hello", Scheduler.TaskPool) 
    .Subscribe(x => Console.WriteLine(x)); 

Jetzt könnte ich etwas "Mischen und Anpassen" tun. Wenn ich einen Verweis auf den Namespace System.Reactive.Threading.Tasks hinzufüge, kann ich ziemlich leicht zwischen Aufgaben und Observablen wechseln.

Task.Factory 
    .StartNew(() => "Hello") 
    .ToObservable() 
    .Subscribe(x => Console.WriteLine(x)); 

Observable 
    .Start(() => "Hello") 
    .ToTask() 
    .ContinueWith(t => Console.WriteLine(t.Result)); 

Beachten Sie die ToObservable() & .ToTask() Anrufe und die daraus resultierende Flips von einer Bibliothek in die andere.

Wenn ich eine Observable habe, die mehr als einen Wert zurückgibt, kann ich die beobachtbare Erweiterungsmethode .ToArray() verwenden, um mehrere Sequenzwerte in einen einzigen Array-Wert umzuwandeln, der in eine Task umgewandelt werden kann.Wie so:

Observable 
    .Interval(TimeSpan.FromSeconds(1.0)) 
    .Take(5) // is IObservable<long> 
    .ToArray() 
    .ToTask() // is Task<long[]> 
    .ContinueWith(t => Console.WriteLine(t.Result.Length)); 

Ich denke, das ist eine ziemlich einfache Antwort auf Ihre Frage. Ist es das, was Sie erwartet haben?