¿Cómo puedo crear un "Flujo" de Procesamiento constante usando el TPL en C# 4

Question

No estoy seguro si lo siguiente es posible, pero me gustaría invocar una serie de Acciones, en Paralell, de una manera acelerada, pero mantenga el flujo de procesamiento continuo, sin volver a utilizar temporizadores o ciclos de bucle/reposo.

Hasta ahora lo he hecho funcionar que carga un lote grande de entradas de alguna fuente ... y luego las procesa en paralelo de manera controlada & en bucles como a continuación.

static void Main(string[] args) 
{ 
    while(true) //Simulate a Timer Elapsing... 
    { 
     IEnumerable<int> inputs = new List<int>() {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}; 
     //Simulate querying database queue tables for next batch of entries 

     RunAllActions(inputs, 3); //Max 3 at a time. 
    } 
} 

static void RunAllActions(IEnumerable<int> inputs, int maxConcurrency) 
{ 
    var options = new ParallelOptions() {MaxDegreeOfParallelism = maxConcurrency}; 

    Parallel.ForEach<int>(inputs, options, DoWork); 
    //Blocks here until all inputs are processed. 
    Console.WriteLine("Batch of Work Done!!!"); 
} 

static void DoWork(int input) 
{ 
    Console.WriteLine("Starting Task {0}", input); 
    System.Threading.Thread.Sleep(3000); 
    Console.WriteLine("Finishing Task {0}", input); 
} 

lo que me gustaría saber es, ¿Hay una construcción en el TPL que podría utilizar para mantenerlo siempre corriendo ... para que pueda reemplazar el "temporizador transcurrido" & "Sondeo de base de datos" con un evento recibido de MessageQueue.

La siguiente es una versión aproximada de lo que me gustaría lograr ... hay otras formas en que puedo hacerlo, pero quiero saber si este tipo de patrón está integrado en el TPL.

internal class Engine 
{ 
    private MessageQueue mq; 
    private Queue<int> myInternalApplicationQueue; 

    public Engine() 
    { 
     //Message Queue to get new task inputs from 
     mq = new MessageQueue(); 
     mq.ReceiveCompleted += new ReceiveCompletedEventHandler(mq_ReceiveCompleted); 

     // internal Queue to put them in. 
     myInternalApplicationQueue = new Queue<int>(); 
    } 

    void mq_ReceiveCompleted(object sender, ReceiveCompletedEventArgs e) 
    { 
     //On MQ Receive, pop the input in a queue in my app 
     int input = (int) e.Message.Body; 

     myInternalApplicationQueue.Enqueue(input); 
    } 

    public void StartWorking() 
    { 
     //Once this gets called, it doesn't stop... it just keeps processing/watching that queue 
     //processing the tasks as fast as it's allowed while the app is running. 
     var options = new ParallelOptions() { MaxDegreeOfParallelism = 3 }; 
     Parallel.KeepWorkingOnQueue<int>(myInternalApplicationQueue, options, DoWork); 
     //  ^^^^^^^^^^^^^^^^^^ <----- THIS GUY 
    } 

} 

Answer 1

Puede utilizar BlockingCollection<T> para manejar este tipo de operación, que es efectivamente un escenario productor/consumidor.

Básicamente, configuraría un BlockingCollection<T> y lo usaría como su "productor". A continuación, tendrá tres (o cualquier número) de tareas de consumidor (que a menudo se configuran como tareas de larga ejecución) que procesan elementos (llamando al blockingCollection.GetConsumingEnumerable() en un bucle foreach estándar).

A continuación, agregue elementos según sea necesario a la colección, y se procesarán continuamente. Cuando hayas terminado, llamarás al BlockingCollection<T>.CompleteAdding, lo que hará que los bucles foreach se completen, y todo se detendrá.

Como nota al margen: normalmente no desea utilizar Parallel.ForEach en el GetConsumingEnumerable() desde un BlockingCollection<T> - al menos no a menos que maneje la partición usted mismo. Por lo general, es mejor usar múltiples tareas y hacer que cada iteración se realice de forma secuencial. El motivo es que el esquema de particionamiento predeterminado en Parallel.ForEach causará problemas (espera hasta que haya disponible un "fragmento" de datos, en lugar de procesar elementos inmediatamente, y los "fragmentos" se vuelven cada vez más grandes).

Answer 2

Como señala Reed, BlockingCollection es una buena manera "manual" de ir aquí. El inconveniente es que también debe administrar a los consumidores usted mismo.

Otro enfoque que tal vez quiera investigar que le quitará mucho trabajo de coordinación de sus manos para escenarios como este es mirar en TPL Dataflow. Específicamente en un escenario como este, puede usar un ActionBlock<T> y, cuando el mensaje proviene de la cola, simplemente Post la nueva información en el ActionBlock<T> y automáticamente lo procesará utilizando los hilos de trabajo TPL debajo de las cubiertas. Eso haría que su clase de Engine mirar un poco algo como esto:

ActionBlock<int> myActionBlock = new ActionBlock<int>(this.ProcessWorkItem); 

void mq_ReceiveCompleted(object sender, ReceiveCompletedEventArgs e)  
{  
    int input = (int)e.Message.Body; 

    // Post the data to the action block 
    this.myActionBlock.Post(input); 
} 

private void ProcessWorkItem(int workItemData) 
{ 
    // ActionBlock will hand each work item to this method for processing 
} 

Ahora, en lo que el paralelismo o la capacidad de controlar, se puede controlar fácil en estos aspectos específicos de la ActionBlock<T> pasando en un ExecutionDataflowBlockOptions al construir la ActionBlock<T>. Entonces, digamos que quiero asegurarme de que nunca tenga un paralelismo de más de cuatro y bloquear al productor para que no agregue más de cien elementos a la cola.Simplemente lo haría:

ActionBlock<int> myActionBlock = new ActionBlock<int>(
            this.ProcessWorkItem, 
            new ExecutionDataflowBlockOptions 
            { 
             MaxDegreeOfParallelism = 4, 
             BoundedCapacity = 100 
            });