Eliminazione di oggetti da una ConcurrentQueue in C#

Question

Ehi, sto cercando di implementare ConcurrentQueue in C# per un server asincrono. Gli articoli vengono messi in coda non appena viene ricevuto un messaggio completo. Per deselezionare i messaggi, sto facendo un piccolo numero di thread per fare il lavoro di dequeueing e manutenzione delle richieste. Questo è inadeguato poiché ogni thread utilizza un ciclo while, che consuma piuttosto una grande quantità di tempo del processore, per ovvi motivi.

Qualcuno dovrebbe conoscere un metodo per eliminare i messaggi quando necessario, ma non consumare così tanto tempo di elaborazione.

{ 
    ... 

    for (int i = 0; i < 3; i++) 
    { 
     Thread t = new Thread(new ThreadStart(startParsingMessages)); 
     t.Start(); 
    } 

    ... 
} 

private void startParsingMessages() 
{ 
    QueueContainer dequeued = null; 
    Console.WriteLine("Trying"); 
    while (true) 
    { 
     if (queue.TryDequeue(out dequeued)) 
     { 
      Console.WriteLine("processing queue"); 
      ProcessMessage(dequeued.socket, dequeued.message); 
     } 
    } 
} 

Answer 1

Invece di usare direttamente ConcurrentQueue<T>, hai provato avvolgendolo in un BlockingCollection<T>? Quindi è possibile utilizzare TryTake(out T, TimeSpan) ecc. Credo che sia l'uso previsto: che le raccolte simultanee stesse sarebbero di solito essere lì solo così è possibile selezionare come funziona la raccolta di blocchi.

Questo non deve essere l'unica impiego per queste raccolte, naturalmente, ma in particolare per ConcurrentQueue<T>, lo scenario di produttore/consumatore di coda è la più comune - a questo punto BlockingCollection<T> è il modo per rendere più facile fare la cosa giusta

Answer 2

C'è un motivo abbastanza noto nella programmazione concorrente denominata Pattern di produzione parallela/Consumatore. Questo articolo potrebbe aiutarti a risolvere alcuni dei tuoi problemi. http://blogs.msdn.com/b/csharpfaq/archive/2010/08/12/blocking-collection-and-the-producer-consumer-problem.aspx

Answer 3

È possibile utilizzare un oggetto di blocco statico per i thread in attesa e quindi emetterlo a impulsi quando qualcosa è pronto per essere elaborato.

static readonly object queueLock = new object(); 

// Threads that enqueue an object 
void QueueMessage() { 
    lock (queueLock) { 
    queue.Enqueue(obj); 
    Monitor.Pulse(queueLock); 
    } 
} 
// Thread dequeuer 
private void startParsingMessages() { 
    QueueContainer dequeued = null; 
    Console.WriteLine("Trying"); 
    while (true) { 
    lock(queueLock) { 
     if (!queue.TryDequeue(out dequeued)) { 
     Console.WriteLine("No object to dequeue, waiting..."); 
     // Threads will wait here and only one will be released when .Pulse()d 
     Monitor.Wait(queueLock); 
     dequeued = queue.Dequeue(); 
     } 
     if (dequeued != null) { 
     Console.WriteLine("processing queue"); 
     ProcessMessage(dequeued.socket, dequeued.message); 
     } 
    } 
    } 
} 

Tenete a mente che questo può diventare piuttosto complicato con i più rami che si possiede, ma l'essenza di esso è, si blocca su un oggetto comune e chiamare Monitor.Wait aspettare su un oggetto per essere Pulse d. Quando questo accade, verrà rilasciato solo un thread che lo stava aspettando. Se accodate molti oggetti e volete che tutti vadano, potete chiamare Monitor.PulseAll che rilascerà tutti i thread che erano in attesa sull'oggetto.