Reactive Extensions (RX) Utilità di pianificazione personalizzata C#

Question

È possibile creare un programma di pianificazione personalizzato in grado di esaminare il valore di ogni elemento che passa attraverso IObservable, al fine di decidere su quale thread elaborare l'elemento?

Ho il requisito di elaborare elementi con lo stesso tasto sequenziale ma tasti diversi in parallelo. Sarebbe logico che RX faccia la programmazione piuttosto che lasciare l'osservabile prima di quanto mi piacerebbe per allocare ogni valore a un thread.

Answer 1

Hai provato GroupBy seguito da ObserveOn?

Qualcosa di simile:

source 
    .GroupBy(item => item.Key) 
    .SelectMany(group => group 
     .ObserveOn(Scheduler.NewThread) 
     .Select(item => process(item)) 
    ) 
    .Subscribe(processResult => ...); 

Ciò si ripartisce il flusso a chiave, avviare un nuovo thread per ogni tasto, ed eseguire process() per ogni elemento in quella chiave.