queue en Python: coordina hilos

El módulo queue en Python ofrece colas sincronizadas para intercambiar tareas y resultados de forma segura entre varios hilos. Encapsula locks y condiciones, de modo que los productores puedan esperar espacio y los consumidores puedan esperar trabajo sin implementar sincronización manual.

Una cola no crea paralelismo por sí sola. Organiza la comunicación, el orden y el backpressure. Los hilos son especialmente útiles para entrada y salida, como archivos, redes y APIs. Para entender el modelo, consulta la guía de threading en Python.

Queue FIFO

from queue import Queue

cola = Queue(maxsize=100)
cola.put({"id": 1, "url": "https://ejemplo.com"})
tarea = cola.get()
try:
    procesar(tarea)
finally:
    cola.task_done()

maxsize limita el trabajo pendiente. Cuando la cola está llena, put() espera hasta que exista espacio, evitando que productores rápidos consuman memoria sin límite.

Productor y consumidores

import queue
import threading

cola = queue.Queue(maxsize=50)

def consumidor():
    while True:
        try:
            tarea = cola.get(timeout=1)
        except queue.Empty:
            continue
        try:
            ejecutar(tarea)
        except Exception as error:
            registrar_fallo(tarea, error)
        finally:
            cola.task_done()

hilos = [threading.Thread(target=consumidor, daemon=True) for _ in range(4)]
for hilo in hilos:
    hilo.start()

for tarea in cargar_tareas():
    cola.put(tarea, timeout=5)

cola.join()

Cada get() exitoso necesita exactamente un task_done(), incluso cuando el procesamiento falla. Colócalo en finally. join() espera a que el contador de tareas pendientes llegue a cero.

Backpressure con maxsize

Una cola ilimitada puede crecer hasta agotar memoria. Selecciona el límite según coste de la tarea, cantidad de workers y latencia aceptable. Si está llena, el productor puede esperar, fallar o guardar el trabajo en un almacenamiento persistente.

import queue

try:
    cola.put(tarea, timeout=2)
except queue.Full:
    guardar_para_despues(tarea)

El timeout evita una espera infinita y crea un punto claro para métricas y fallback.

No decidas con qsize

qsize(), empty() y full() son aproximados. Otro hilo puede cambiar el estado inmediatamente. Ejecuta put() o get() con timeout y maneja Full o Empty.

Shutdown de la cola

Desde Python 3.13, Queue.shutdown() detiene el crecimiento y despierta llamadas bloqueadas:

import queue

cola.shutdown(immediate=False)

try:
    cola.put(nueva_tarea)
except queue.ShutDown:
    print("la cola ya no acepta trabajo")

Con immediate=False, los consumidores pueden vaciar las tareas existentes y join() conserva su garantía normal. Con immediate=True, la cola se vacía y join() puede regresar aunque el trabajo no se haya ejecutado. Reserva el modo inmediato para cancelaciones de emergencia.

Consumidor compatible con shutdown

import queue

while True:
    try:
        tarea = cola.get()
    except queue.ShutDown:
        break
    try:
        ejecutar(tarea)
    finally:
        cola.task_done()

Este patrón expresa la intención directamente. En versiones anteriores, utiliza una estrategia de sentinelas documentada.

Sentinelas

PARAR = object()

def consumidor():
    while True:
        item = cola.get()
        try:
            if item is PARAR:
                return
            ejecutar(item)
        finally:
            cola.task_done()

for _ in hilos:
    cola.put(PARAR)

Envía una sentinela por consumidor. No uses un valor válido como None si podría ser una tarea real.

LifoQueue

LifoQueue funciona como una pila: el elemento más reciente sale primero. Puede servir en búsquedas en profundidad o cuando las tareas nuevas son más urgentes. Una carga continua puede dejar tareas antiguas esperando; define expiración o usa FIFO cuando la justicia temporal importe.

PriorityQueue

from queue import PriorityQueue

cola = PriorityQueue()
cola.put((10, "informe normal"))
cola.put((1, "incidente crítico"))
prioridad, tarea = cola.get()

El menor valor sale primero. Cuando las prioridades son iguales, se compara el siguiente elemento, lo que falla con objetos no comparables. Añade un contador o wrapper:

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class ItemPrioritario:
    prioridad: int
    secuencia: int
    item: Any = field(compare=False)

SimpleQueue

SimpleQueue es FIFO ilimitada sin tracking, maxsize, join() ni shutdown. Úsala cuando otro mecanismo controle el volumen y solo necesites transporte.

En CPython, su implementación es reentrante y puede utilizarse en destructores o callbacks de referencias débiles. Para ese ciclo de vida, revisa weakref en Python.

No es multiprocessing.Queue

queue.Queue sincroniza hilos del mismo proceso. Los procesos separados requieren multiprocessing.Queue u otro IPC, con serialización y fallos diferentes.

Errores y reintentos

Cuando una tarea falla, elige entre descartar, reintentar con límite, enviar a una dead-letter queue o detener el worker. Evita reintentos infinitos.

from dataclasses import replace

if tarea.intentos < 3:
    cola.put(replace(tarea, intentos=tarea.intentos + 1))
else:
    fallos.put(tarea)

Añade retraso para que una dependencia caída no produzca un bucle intenso.

Resultados y correlación

Usa una segunda cola para resultados e incluye un ID de correlación. No esperes una respuesta mientras mantienes locks que el consumidor necesita.

Observabilidad

Mide espera en cola, duración de procesamiento, errores, reintentos y profundidad aproximada. Usa la profundidad como métrica, no como garantía. Para perfilar workers, consulta pstats en Python y trace en Python.

Evita deadlocks

• Equilibra get() con task_done().
• Inicia consumidores antes de join().
• No mantengas locks externos durante put() bloqueante.
• Usa timeouts para servicios externos.
• Define una estrategia de shutdown.
• Evita ciclos donde un consumidor espera trabajo que él mismo debe producir.

Pruebas

Prueba cola llena y vacía, timeouts, excepciones, shutdown normal e inmediato, sentinelas, prioridades iguales y cancelación. Coordina con eventos en lugar de sleep() arbitrario.

Buenas prácticas

• Usa colas limitadas para backpressure.
• Coloca task_done() en finally.
• Maneja Empty, Full y ShutDown.
• No bases lógica en empty().
• Documenta reintentos.
• Incluye IDs de correlación.
• Mide espera y procesamiento.
• Elige FIFO, LIFO o prioridad conscientemente.

Conclusión

queue en Python simplifica la comunicación entre hilos y ofrece backpressure, tracking, prioridad y terminación coordinada. Es más predecible que compartir listas con locks improvisados.

Consulta la documentación oficial de queue y la documentación de threading. Para configurar workers y límites, revisa configparser en Python.