Colas y Productor/Consumidor
Los pipelines de productor/consumidor desacoplan etapas con queue.Queue (hilos) o multiprocessing.Queue (procesos) de tamaño limitado. Los productores encolan el trabajo; los consumidores lo procesan a su propio ritmo, suavizando picos y simplificando la contrapresión.
Receta
from queue import Queue
from threading import Thread
q: Queue[str] = Queue(maxsize=10)
def producer() -> None:
for i in range(5):
q.put(f"job-{i}")
def consumer() -> None:
while True:
item = q.get()
try:
print("procesando", item)
finally:
q.task_done()
Thread(target=consumer, daemon=True).start()
producer()
q.join()Cuándo usar esto:
- Pipelines de descarga -> análisis -> almacenamiento
- Grupos de hilos alimentados por un único productor
- Contrapresión mediante el bloqueo de
putconmaxsize - Cierre elegante con centinelas o
join - Trabajadores de registro/métricas fuera de la ruta principal
Ejemplo de Trabajo
from queue import Queue
from threading import Thread, Event
STOP = object()
def worker(q: Queue, stop_event: Event) -> None:
while not stop_event.is_set():
try:
item = q.get(timeout=0.1
Lo que esto demuestra:
- La cola limitada bloquea al productor cuando está llena (
maxsize=3) task_done/joinrastrean el trabajo pendiente- Centinela STOP por consumidor para una salida limpia
- Múltiples consumidores compiten de manera justa en
get
Análisis Profundo
Tipos de Cola
| Cola | Ámbito |
|---|---|
queue.Queue | Hilos en el mismo proceso |
multiprocessing.Queue | IPC entre procesos |
asyncio.Queue | Corrutinas asíncronas |
Contrapresión
put(block=True, timeout=...)cuando está llena- Monitorizar
qsize()con cautela: aproximado en multiprocessing
Trampas Comunes
task_doneolvidado -joinse cuelga para siempre. Solución:try/finally: q.task_done()por elemento.- Memoria de cola ilimitada - los productores superan a los consumidores. Solución:
maxsize+putbloqueante. - Píldora envenenada por consumidor - un solo centinela detiene solo un hilo. Solución: un STOP por trabajador.
- Serialización de objetos grandes en colas MP - lento. Solución: memoria compartida o rutas a archivos.
- Procesamiento dentro de
getsin tiempo de espera - no se puede interrumpir el cierre. Solución: tiempos de espera + evento.
Alternativas
| Alternativa | Usar cuando | No usar cuando |
|---|---|---|
| asyncio.Queue | Pipeline asíncrono | Consumidores bloqueantes |
| Broker externo (Redis) | Entre servicios | Solo dentro del proceso |
| Fragmentos de iterador | Scripts de lotes simples | Muchos productores |
Preguntas Frecuentes
¿Significado de maxsize=0?
Cola ilimitada: put nunca se bloquea por capacidad.
¿PriorityQueue?
queue.PriorityQueue para trabajos ordenados: tuplas (prioridad, elemento).
¿Cuántos consumidores?
Igualar el rendimiento sostenido: a menudo similar al límite de paralelismo de E/S.
¿Diferencias de asyncio.Queue?
Esperar put/get; no es seguro para hilos con threading sin un puente.
¿Cómo probar pipelines?
Usar maxsize pequeño, elementos deterministas, join con tiempo de espera en las pruebas.
¿Cola de mensajes fallidos (dead letter queue)?
Cola secundaria para elementos fallidos después de N reintentos: patrón sobre la cola base.
¿Productor de registro?
QueueHandler en logging dirige los registros a un hilo de escucha: patrón de la biblioteca estándar.
¿Equidad?
La cola estándar es FIFO; múltiples consumidores obtienen una distribución aproximadamente equitativa.
¿Cola de procesos vs. hilos?
Nunca mezclar: elegir la cola de threading o multiprocessing que coincida con los trabajadores.
¿Orden de cierre?
Detener primero los productores, vaciar la cola, enviar centinelas, unir consumidores.
Relacionado
- threading - hilos de trabajo
- multiprocessing - colas de procesos
- concurrent.futures - alternativa de grupo
- Asyncio Basics - asyncio.Queue
Versiones de Stack: Esta página fue escrita para Python 3.14.0 (estable 3.14, mantenimiento 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, y uv 0.6+.