1️⃣ ¿Qué es una cola?

Una cola es una estructura de datos lineal que sigue el principio FIFO (First‑In‑First‑Out): el primer elemento que entra es el primero que sale. Este comportamiento la hace ideal para buffering, planificación de tareas, mensajería asíncrona y muchos patrones de DevOps como pipelines de CI/CD.

Operaciones básicas

• enqueue(item) – Inserta item al final de la cola.
• dequeue() – Elimina y devuelve el elemento al frente.
• peek() – Devuelve el elemento del frente sin eliminarlo.
• is_empty() – Indica si la cola está vacía.
• size() – Número de elementos almacenados.

2️⃣ Implementaciones nativas en Python

class ListQueue:
    def __init__(self):
        self._items = []                     # almacén interno
    def enqueue(self, item):
        self._items.append(item)             # inserción O(1)
    def dequeue(self):
        if self.is_empty():
            raise IndexError("dequeue from empty queue")
        return self._items.pop(0)            # eliminación O(n) → no recomendado en producción
    def peek(self):
        if self.is_empty():
            raise IndexError("peek from empty queue")
        return self._items[0]
    def is_empty(self):
        return len(self._items) == 0
    def size(self):
        return len(self._items)

from collections import deque
class DequeQueue:
    def __init__(self):
        self._queue = deque()
    def enqueue(self, item):
        self._queue.append(item)          # O(1)
    def dequeue(self):
        if self.is_empty():
            raise IndexError("dequeue from empty queue")
        return self._queue.popleft()      # O(1)
    def peek(self):
        if self.is_empty():
            raise IndexError("peek from empty queue")
        return self._queue[0]
    def is_empty(self):
        return not self._queue
    def size(self):
        return len(self._queue)

3️⃣ Colas seguras para hilos (thread‑safe)

Cuando varios hilos producen y consumen simultáneamente, es imprescindible garantizar la exclusión mutua. Python incluye queue.Queue, una implementación bloqueante con soporte de timeouts y maxsize.

import queue
import threading
import time
def producer(q, n):
    for i in range(n):
        item = f"msg-{i}"
        q.put(item)                     # bloquea si la cola está llena
        print(f"Producer → {item}")
        time.sleep(0.1)
def consumer(q):
    while True:
        try:
            item = q.get(timeout=2)     # lanza queue.Empty si no hay datos
            print(f"Consumer ← {item}")
            q.task_done()
        except queue.Empty:
            break
q = queue.Queue(maxsize=10)            # límite opcional
t1 = threading.Thread(target=producer, args=(q, 20))
t2 = threading.Thread(target=consumer, args=(q,))
t1.start(); t2.start()
t1.join(); q.join()                     # espera a que se procesen todos los ítems

Consideraciones de rendimiento

• Latencia: queue.Queue usa un Condition interno; la sobrecarga es mínima (
• Throughput: Para cargas intensas (> 10 k ops/s) evalúa multiprocessing.Queue o asyncio.Queue según el modelo de concurrencia.

4️⃣ Colas asíncronas con asyncio.Queue

En aplicaciones basadas en event‑loop (por ejemplo, servidores web con FastAPI o bots de Discord) la cola no bloquea el hilo principal. Cada await cede el control al loop, permitiendo alta concurrencia sin crear hilos.

import asyncio
import random
async def producer(q, count):
    for i in range(count):
        await asyncio.sleep(random.uniform(0.05, 0.2))
        item = f"task-{i}"
        await q.put(item)
        print(f"🔺 Produced {item}")
async def consumer(q, name):
    while True:
        item = await q.get()
        print(f"🔻 {name} processing {item}")
        await asyncio.sleep(random.uniform(0.1, 0.3))
        q.task_done()
async def main():
    q = asyncio.Queue(maxsize=5)
    prod = asyncio.create_task(producer(q, 15))
    cons = [asyncio.create_task(consumer(q, f"worker-{i}")) for i in range(3)]
    await prod                     # espera al productor
    await q.join()                 # espera a que todos los ítems se consuman
    for c in cons: c.cancel()      # termina los consumidores
asyncio.run(main())

Tips de producción

• Back‑pressure: define maxsize para evitar que el productor sature la memoria.
• Graceful shutdown: usa queue.join() y task.cancel() para cerrar cleanly.

5️⃣ Comparativa de opciones de cola en Python

6️⃣ Patrones de uso reales

7️⃣ Buenas prácticas, seguridad y troubleshooting

7.1. Evita list.pop(0) en producción

El coste lineal genera “GC pauses” y consumo de CPU innecesario. Prefiere deque o una cola bloqueante.

7.2. Limita el tamaño (maxsize)

Un maxsize razonable protege contra ataques de denegación de servicio que intentan saturar la memoria con miles de mensajes. En entornos críticos, combina maxsize con timeouts al put() y get().

7.3. Manejo de excepciones

• queue.Empty – siempre captura cuando uses get(timeout=…).
• queue.Full – útil al usar put_nowait() en productores críticos.

7.4. Registro y métricas

Integra prometheus_client o statsd para exponer:

• queue_size – número actual de ítems.
• queue_enqueue_seconds – latencia de inserción.
• queue_dequeue_seconds – latencia de consumo.

7.5. Seguridad en entornos multi‑tenant

Si diferentes usuarios comparten una misma cola (p.ej., en un servicio SaaS), usa namespaces o prefijos de clave y valida siempre los datos antes de enqueue. Evita la inyección de objetos maliciosos (por ejemplo, pickle no confiable) y considera serializar a JSON.