Procesamiento de colas asíncronas con extensiones reactivas

Question

Hay un par de artículos sobre esto, y tengo esto funcionando ... pero quiero saber cómo establecer un número máximo de subprocesos de tareas para mis suscripciones observables a la vez.

Tengo el siguiente para paralelizar ahorro asíncrono de entradas de registro:

private BlockingCollection<ILogEntry> logEntryQueue; 

y

logEntryQueue = new BlockingCollection<ILogEntry>(); 
logEntryQueue.GetConsumingEnumerable().ToObservable(Scheduler.TaskPool).Subscribe(SaveLogEntry); 

Para programar mi ahorro ... pero ¿cómo puedo especificar los hilos máximos para que el programador usar a la vez?

Answer 1

Esto no es una función del Observable, sino una función del Programador. El Observable define qué y el planificador define donde.

Tendría que pasar un programador personalizado. Una forma simple de hacer esto sería sub-clase TaskScheduler y anular la propiedad "MaximumConcurrencyLevel".

http://msdn.microsoft.com/en-us/library/system.threading.tasks.taskscheduler.maximumconcurrencylevel.aspx

realidad me encontré con una muestra de este en MSDN:

http://msdn.microsoft.com/en-us/library/ee789351.aspx

Editar: Usted preguntó acerca de cómo pasar de TaskScheduler a iScheduler. Otro desarrollador sólo me dio un poquito de información:

var ischedulerForRx = new TaskPoolScheduler 
(
    new TaskFactory 
    (
     //This is your custom scheduler 
     new LimitedConcurrencyLevelTaskScheduler(1) 
    ) 
); 

Answer 2

Si crea su "trabajo" como IObservable<T> con ejecución diferida (. Es decir, que quieren hacer nada hasta suscrito a), se puede utilizar la sobrecarga de Merge que acepte una cantidad de suscripciones simultáneas máximas:

ISubject<QueueItem> synchronizedQueue = new Subject<QueueItem>().Synchronize(); 

queue 
    .Select(item => StartWork(item)) 
    .Merge(maxConcurrent: 5) // C# 4 syntax for illustrative purposes 
    .Subscribe(); 

// To enqueue: 
synchronizedQueue.OnNext(new QueueItem());