Colas y pilas Producer-Consumer

1. Introducción

Empecemos por lo básico. Cola es una estructura de datos fundamental que funciona según el principio FIFO (First-In, First-Out), es decir "el primero que llega — el primero que sale". Imagina una cola en el supermercado: quien se puso primero, se atiende primero.

En programación multihilo el patrón Producer-Consumer (Producer-Consumer) es uno de los más comunes y poderosos.

• Productores (Producers) — hilos o partes de la aplicación que crean datos o tareas y las ponen en una cola compartida. Ellos "producen" trabajo.
• Consumidores (Consumers) — hilos o partes de la aplicación que toman datos o tareas de la cola y las procesan. Ellos "consumen" trabajo.

Este patrón ayuda a controlar el flujo de datos, desacopla componentes del sistema (el productor no necesita saber quién ni cómo procesará los datos), hace la app más reactiva y ayuda a distribuir la carga entre hilos de forma uniforme.

Ejemplo: ConcurrentQueue — añadir y extraer

Miremos cómo añadir y extraer elementos de ConcurrentQueue<T>.

using System.Collections.Concurrent;

ConcurrentQueue<string> tasks = new ConcurrentQueue<string>();

// Añadir elementos (productor)
tasks.Enqueue("Descargar archivo");
tasks.Enqueue("Procesar imagen");
Console.WriteLine($"Tareas en la cola: {tasks.Count}"); // Salida: Tareas en la cola: 2

// Extraer elementos (consumidor)
if (tasks.TryDequeue(out string task1))
{
    Console.WriteLine($"Tarea ejecutada: {task1}"); // Salida: Tarea ejecutada: Descargar archivo
}

if (tasks.TryDequeue(out string task2))
{
    Console.WriteLine($"Tarea ejecutada: {task2}"); // Salida: Tarea ejecutada: Procesar imagen
}

if (!tasks.TryDequeue(out string emptyTask))
{
    Console.WriteLine("La cola está vacía, no hay tareas nuevas."); // Salida: La cola está vacía, no hay tareas nuevas.
}

Fundamentos: Enqueue(), TryDequeue()

Enqueue(T item): Se usa para añadir un elemento al final de la cola. Esta operación es thread-safe. Puedes llamar Enqueue simultáneamente desde 10 hilos distintos y todos los elementos se añadirán correctamente.

TryDequeue(out T item): Se usa para intentar extraer un elemento del principio de la cola. Es el método clave para los consumidores. Devuelve true si el elemento se extrajo con éxito (el valor va al parámetro de salida item), y false si la cola está vacía. Importante: TryDequeue no bloquea el hilo si la cola está vacía.

2. Importancia de TryDequeue() y la atomicidad de las operaciones

El método TryDequeue() no sólo es conveniente; es críticamente importante para un correcto funcionamiento thread-safe. Es atómico: la comprobación de vacío y la extracción se realizan como una única operación indivisible.

Si tuviéramos métodos separados IsEmpty (comprobar si la cola está vacía) y Dequeue (extraer elemento), entre sus llamadas otro hilo podría vaciar la cola. Como resultado tu Dequeue lanzaría una excepción o devolvería datos incorrectos. TryDequeue elimina totalmente esa situación.

Ejemplo: Producer-Consumer con varios hilos

Aquí lanzamos dos hilos productores y un hilo consumidor.

using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

ConcurrentQueue<int> dataQueue = new ConcurrentQueue<int>();
bool producersDone = false; // Flag para señalizar al consumidor

void Producer(int start, int count)
{
    for (int i = 0; i < count; i++)
    {
        dataQueue.Enqueue(start + i);
        Console.WriteLine($"[P] Añadido: {start + i}");
        Thread.Sleep(10); 
    }
}

void Consumer()
{
    while (!producersDone || dataQueue.Count > 0) // Continuamos mientras haya datos o los productores estén trabajando
    {
        if (dataQueue.TryDequeue(out int item))
        {
            Console.WriteLine($"[C] Procesado: {item}");
        }
        else
        {
            Thread.Sleep(50); // Esperamos si la cola está vacía
        }
    }
    Console.WriteLine("[C] Terminó su trabajo.");
}

// Ejecutar ejemplo en Main:
// Task.Run(() => Producer(1, 5));
// Task.Run(() => Producer(100, 5)); // Segundo productor
// Task.Run(() => Consumer());
// Thread.Sleep(600); // Damos tiempo a los hilos para trabajar
// producersDone = true; // Señalamos que los productores terminaron
// Thread.Sleep(200); // Damos tiempo al consumidor para procesar el resto

Fíjate que en este ejemplo simple el flag producersDone y Thread.Sleep se usan para simular el cierre. En apps reales para una sincronización más robusta suelen usarse CancellationTokenSource o BlockingCollection<T>.

ConcurrentQueue<T> es ideal para escenarios donde:

• El orden de procesamiento es importante (FIFO).
• Muchos hilos añaden elementos y/o muchos hilos los consumen.
• Se necesita alto rendimiento sin manejar locks manualmente.

3. Pila para producer-consumer (LIFO)

Pila es otra estructura fundamental, que funciona según el principio LIFO (Last-In, First-Out), es decir "el último que llega — el primero que sale". Imagina una pila de platos: siempre coges el de arriba, y el nuevo plato siempre se pone arriba de la pila.

ConcurrentStack<T> es tan thread-safe como ConcurrentQueue<T>, y también puede usarse en el patrón producer-consumer, pero con orden invertido de procesamiento.

Ejemplo: ConcurrentStack — añadir y extraer

using System.Collections.Concurrent;

ConcurrentStack<string> commandStack = new ConcurrentStack<string>();

// Añadir comandos (productor)
commandStack.Push("Seleccionar texto");
commandStack.Push("Cambiar fuente");
commandStack.Push("Guardar documento");
Console.WriteLine($"Comandos en la pila: {commandStack.Count}"); // Salida: Comandos en la pila: 3

// Extraer comandos (consumidor)
if (commandStack.TryPop(out string cmd1))
{
    Console.WriteLine($"Comando deshecho: {cmd1}"); // Salida: Comando deshecho: Guardar documento
}

if (commandStack.TryPop(out string cmd2))
{
    Console.WriteLine($"Comando deshecho: {cmd2}"); // Salida: Comando deshecho: Cambiar fuente
}

if (!commandStack.TryPop(out string emptyCmd))
{
    Console.WriteLine("La pila de comandos está vacía."); // Salida: La pila de comandos está vacía.
}

4. Fundamentos: Push(), TryPop()

Push(T item): Se usa para añadir un elemento en la cima de la pila. La operación es thread-safe.

TryPop(out T item): Se usa para intentar extraer el elemento de la cima de la pila. Devuelve true si se extrajo con éxito, y false si la pila está vacía. Al igual que TryDequeue, es una operación atómica que evita race conditions.

Ejemplo: usar ConcurrentStack para un pool de objetos

La pila es excelente para implementar pools de objetos: coges — usas — devuelves.

using System.Collections.Concurrent;

class Connection { /* Simple stub */ public Guid Id { get; } = Guid.NewGuid(); }

ConcurrentStack<Connection> connectionPool = new ConcurrentStack<Connection>();

// Rellenamos el pool con conexiones iniciales
for (int i = 0; i < 3; i++)
{
    connectionPool.Push(new Connection());
}
Console.WriteLine($"Conexiones en el pool: {connectionPool.Count}"); // Salida: Conexiones en el pool: 3

void UseConnection()
{
    if (connectionPool.TryPop(out Connection conn))
    {
        Console.WriteLine($"[Pool] Usada conexión: {conn.Id}");
        // Simulación de trabajo con la conexión
        Thread.Sleep(50); 
        connectionPool.Push(conn); // Devolvemos al pool
        Console.WriteLine($"[Pool] Devuelta conexión: {conn.Id}. En el pool: {connectionPool.Count}");
    }
    else
    {
        Console.WriteLine("[Pool] El pool está vacío. Creando nueva conexión.");
        // Normalmente aquí se crea una nueva conexión si el pool está vacío
        connectionPool.Push(new Connection()); 
    }
}

// Ejecutar ejemplo en Main:
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Thread.Sleep(500);

En este ejemplo varios hilos pueden coger y devolver conexiones al pool compartido de forma segura.

5. Casos de uso y comparación con ConcurrentQueue

ConcurrentStack<T> se usa cuando:

• El orden LIFO es crítico (por ejemplo, historial de operaciones para "Undo").
• Se necesita acceso rápido a los elementos más recientes añadidos (a menudo "hot" en la caché de la CPU).
• Se implementan algoritmos basados en pilas (recorridos DFS, análisis sintáctico de expresiones).

Comparación y elección de la colección adecuada

ConcurrentQueue

ConcurrentStack

La elección entre ConcurrentQueue y ConcurrentStack depende totalmente del orden de procesamiento que necesites en tu escenario producer-consumer. Ambas colecciones ofrecen alto rendimiento y thread-safety "out of the box", liberándote de la sincronización manual y ayudándote a construir sistemas multihilo escalables.

2

Tarea

C# SELF, nivel 58, lección 1

Bloqueada

Implementación de Producer-Consumer usando una cola

Implementación de Producer-Consumer usando una cola

Abrir