¿Cómo (y si) escribir una cola de un solo consumidor usando TPL?

Question

Recientemente escuché un montón de podcasts sobre TPL en .NET 4.0.La mayoría de ellos describen actividades en segundo plano, como descargar imágenes o realizar un cálculo, utilizando tareas para que el trabajo no interfiera con un hilo de GUI.

La mayor parte del código en el que trabajo tiene más bien una versión de múltiples productores/un solo consumidor, donde los elementos de trabajo de múltiples fuentes deben ponerse en cola y luego procesarse en orden.Un ejemplo sería el registro, donde las líneas de registro de varios subprocesos se secuencian en una única cola para su eventual escritura en un archivo o base de datos.Todos los registros de una sola fuente deben permanecer en orden, y los registros del mismo momento deben estar "cercanos" entre sí en la salida final.

Entonces, múltiples subprocesos o tareas o lo que sea están invocando una cola:

lock( _queue ) // or use a lock-free queue!
{
   _queue.enqueue( some_work );
   _queueSemaphore.Release();
}

Y un hilo de trabajo dedicado procesa la cola:

while( _queueSemaphore.WaitOne() )
{
   lock( _queue )
   {
      some_work = _queue.dequeue();     
   }
   deal_with( some_work );
}

Siempre ha parecido razonable dedicar un hilo de trabajo al lado del consumidor de estas tareas.¿Debería escribir programas futuros utilizando alguna construcción de TPL?¿Cuál?¿Por qué?

Answer 1

Puede utilizar una tarea en ejecución tiempo para procesar los artículos en un BlockingCollection según lo sugerido por Wilka. He aquí un ejemplo que cumple más o menos sus requisitos de las aplicaciones. Verá algo salida como esta:

Log from task B
Log from task A
Log from task B1
Log from task D
Log from task C

No es que las salidas de la A, B, C y D aparecen al azar, ya que dependen de la hora de inicio de las discusiones, pero siempre aparece antes de B B1.

public class LogItem 
{
    public string Message { get; private set; }

    public LogItem (string message)
    {
        Message = message;
    }
}

public void Example()
{
    BlockingCollection<LogItem> _queue = new BlockingCollection<LogItem>();

    // Start queue listener...
    CancellationTokenSource canceller = new CancellationTokenSource();
    Task listener = Task.Factory.StartNew(() =>
        {
            while (!canceller.Token.IsCancellationRequested)
            {
                LogItem item;
                if (_queue.TryTake(out item))
                    Console.WriteLine(item.Message);
            }
        },
    canceller.Token, 
    TaskCreationOptions.LongRunning,
    TaskScheduler.Default);

    // Add some log messages in parallel...
    Parallel.Invoke(
        () => { _queue.Add(new LogItem("Log from task A")); },
        () => { 
            _queue.Add(new LogItem("Log from task B")); 
            _queue.Add(new LogItem("Log from task B1")); 
        },
        () => { _queue.Add(new LogItem("Log from task C")); },
        () => { _queue.Add(new LogItem("Log from task D")); });

    // Pretend to do other things...
    Thread.Sleep(1000);

    // Shut down the listener...
    canceller.Cancel();
    listener.Wait();
}

Answer 2

Sé que esta respuesta es alrededor de un año de retraso, pero echar un vistazo a MSDN .

que muestra cómo crear un LimitedConcurrencyLevelTaskScheduler de la clase TaskScheduler. Al limitar la concurrencia a una sola tarea, que luego debe procesar sus tareas en el orden en que se ponen en cola a través de:

LimitedConcurrencyLevelTaskScheduler lcts = new LimitedConcurrencyLevelTaskScheduler(1);
TaskFactory factory = new TaskFactory(lcts);

factory.StartNew(()=> 
{
   // your code
});

Answer 3

No estoy seguro de que la TPL es adecuado en su caso de uso. Desde mi entender el caso de uso principal para el TPL es dividir una tarea enorme en varias tareas más pequeñas que se pueden ejecutar al lado del otro. Por ejemplo, si usted tiene una lista grande y desea aplicar la misma transformación en cada elemento. En este caso, puede tener varias tareas aplicar la transformación en un subconjunto de la lista.

El caso que usted describe no parece encajar en esta imagen para mí. En el caso de que no tiene varias tareas que hacen lo mismo en paralelo. Tiene varias tareas diferentes que cada uno hace es propio puesto de trabajo (los productores) y una tarea que consume. Tal vez TPL podría ser utilizado para la parte de los consumidores si quieren tener varios consumidores ya que en este caso, cada consumidor hace el mismo trabajo (suponiendo que encuentre una lógica para hacer cumplir la consistencia temporal busca).

Bueno, esto por supuesto es sólo mi punto de vista sobre el tema personnal

Larga vida y prosperidad

Answer 4

BlockingCollection haría ser útil para usted. Así que para su código anterior, se podría utilizar algo como (suponiendo _queue es una instancia BlockingCollection):

// for your producers 
_queue.Add(some_work);

Un subproceso de trabajo dedicado a la tramitación de la cola:

foreach (var some_work in _queue.GetConsumingEnumerable())
{
    deal_with(some_work);
}

Nota: cuando todos sus productores han terminado de producir cosas, tendrá que llamar CompleteAdding() en _queue de lo contrario será atrapado su consumidor espera de más trabajo

.