c# - Probar Dequeue en ConcurrentQueue

producer-consumer concurrent-queue

Question

El TryDequeue en ConcurrentQueue devolverá false si no hay elementos en la cola.

Si la cola está vacía, necesito que mi cola espere hasta que se agregue un nuevo elemento en la cola y elimine la cola de la nueva, y el proceso continuará así.

¿Debo usar monitor.enter, esperar, pulsar, o alguna otra opción mejor en C # 4.0?

Answer 1

¿No es esto para lo que está diseñado BlockingCollection ?

Como lo entiendo, puedes envolver tu ConcurrentQueue con uno de estos, y luego llamar a Take .

Answer 2

Periódicamente, puede verificar el número de elementos en la cola y cuando el número de elementos es mayor que cero, le da una señal usando, por ejemplo, ManualResetEvent al hilo que pone en cola el elemento (s) hasta que la cola esté vacía.

Aquí está el pseudo código para esto:

Revisar el hilo:

while(true)
{
  int QueueLength = 0;
  lock(Queue)
  {
    queueLength = Queue.Length;
  }

  if (Queue.Length > 0)
  {
    manualResetEvent.Set();
  }
  else
  {
    Thread.Sleep(...);
  }       
}    

Hilo de salida:

while(true)
{
  if(manualResetEvent.WaitOne(timeout))
  {
    DequeueUntilQueueEmpty();
  }
}

Considere usar el bloqueo en DequeueUntilQueueEmpty, también.

Answer 3

Puedes usar BlockingCollection .

Haz algo como eso:

private BlockingCollection<string> rowsQueue;
private void ProcessFiles() {
   this.rowsQueue = new BlockingCollection<string>(new ConcurrentBag<string>(), 1000);
   ReadFiles(new List<string>() { "file1.txt", "file2.txt" });


   while (!this.rowsQueue.IsCompleted || this.rowsQueue.Count > 0)
   {
       string line = this.rowsQueue.Take();

       // Do something
   }
}

private Task ReadFiles(List<string> fileNames)
{
    Task task = new Task(() =>
    {
        Parallel.ForEach(
        fileNames,
        new ParallelOptions
        {
            MaxDegreeOfParallelism = 10
        },
            (fileName) =>
            {
                using (StreamReader sr = File.OpenText(fileName))
                {
                    string line = String.Empty;
                    while ((line = sr.ReadLine()) != null)
                    {
                           this.rowsQueue.Add(line);
                    }
                }
            });

        this.rowsQueue.CompleteAdding();
    });

    task.Start();

    return task;
}