TPL Dataflow y Rx Ejemplo combinado

Question

Solo quiero aprender ambos y cómo usarlos juntos. Entiendo que pueden complementarse entre sí. Simplemente no pude encontrar un ejemplo de alguien que realmente lo esté haciendo.

Answer 1

Permítanme comenzar con un poco de antecedentes.

El marco .NET tiene una serie de tipos especiales - clases o interfaces adecuadas - Task<T>, IObservable<T>, Nullable<T>, IEnumerable<T>, Lazy<T>, etc - que proporcionan poderes especiales al tipo subyacente T.

El TPL usa Task<T> para representar el cálculo asincrónico de un único valor de T.

Rx usa IObservable<T> para representar el cálculo asincrónico de cero o más valores de T.

Es el aspecto de "cálculo asincrónico" de ambos los que unen TPL y Rx.

Ahora, el TPL también utiliza el tipo Task para representar la ejecución asíncrona de un lambda Action, pero esto puede ser considerado como un caso especial de Task<T> donde T es void. Muy parecido a un método estándar en C# vuelve void así:

public void MyMethod() { } 

Rx también permite el mismo caso especial con el uso de un tipo especial llamado Unit.

La diferencia entre TPL y Rx está en el número de valores devueltos. TPL es uno y solo uno, mientras que Rx es cero o más.

Por lo tanto, si trata el Rx de una manera especial trabajando solo con secuencias observables que devuelven un valor único, puede hacer algunos cálculos de forma similar al TPL.

Por ejemplo, en el TPL podría escribir:

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

Y en Rx el equivalente sería:

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

Me podría dar un paso más en la Rx especificando que el TPL debe ser utilizado para ejecutar el cálculo de este modo: (. Por defecto se utiliza el grupo de subprocesos)

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

Ahora podría hacer algo de "mezclar y combinar". Si agrego una referencia al espacio de nombre System.Reactive.Threading.Tasks, puedo moverme entre las tareas y los observables con bastante facilidad.

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

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

Aviso los ToObservable() & .ToTask() llamadas y la consiguiente voltea de una biblioteca a la otra.

Si tengo un observable que devuelve más de un valor, puedo usar el método de extensión observable .ToArray() para convertir múltiples valores de secuencia en un solo valor de matriz que se puede convertir en una tarea.Como tal:

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

Creo que esta es una respuesta bastante básica a su pregunta. ¿Es lo que estabas esperando?