Colas en Python
Una cola (queue) es una estructura de datos que sigue el principio FIFO (First In, First Out), lo que significa que el primer elemento en entrar es el primero en salir.
Concepto básico
Imagina una fila de personas esperando:
Las nuevas personas llegan al final de la fila (enqueue)
Las personas son atendidas desde el frente de la fila (dequeue)
Implementaciones en Python
1. Usando collections.deque (Recomendado - más eficiente)
from collections import deque
# Crear una cola vacía
cola = deque()
# Agregar elementos al final (enqueue)
cola.append("Cliente A")
cola.append("Cliente B")
cola.append("Cliente C")
print(f"Cola: {list(cola)}") # Cola: ['Cliente A', 'Cliente B', 'Cliente C']
# Eliminar el primer elemento (dequeue)
primer_elemento = cola.popleft()
print(f"Atendiendo: {primer_elemento}") # Atendiendo: Cliente A
print(f"Cola después: {list(cola)}") # Cola después: ['Cliente B', 'Cliente C']
# Ver el primer elemento sin eliminarlo
print(f"Siguiente en la cola: {cola[0]}") # Siguiente en la cola: Cliente B2. Usando queue.Queue (Para programación multi-hilo)
from queue import Queue
# Crear una cola
cola = Queue()
# Agregar elementos
cola.put("Tarea 1")
cola.put("Tarea 2")
cola.put("Tarea 3")
# Eliminar elementos
print(cola.get()) # Tarea 1
print(cola.get()) # Tarea 2
# Verificar si está vacía
print(cola.empty()) # False
# Obtener tamaño
print(cola.qsize()) # 13. Usando listas (No recomendado para colas grandes)
# Crear cola con lista
cola = []
# Enqueue - agregar al final
cola.append(10)
cola.append(20)
cola.append(30)
print(f"Cola: {cola}") # Cola: [10, 20, 30]
# Dequeue - eliminar del frente (ineficiente para listas grandes)
elemento = cola.pop(0)
print(f"Elemento eliminado: {elemento}") # Elemento eliminado: 10
print(f"Cola después: {cola}") # Cola después: [20, 30]Operaciones principales de una cola
from collections import deque
# Crear cola
cola = deque()
# 1. Enqueue - agregar elemento al final
cola.append("primero")
cola.append("segundo")
cola.append("tercero")
print(f"Cola después de enqueue: {list(cola)}")
# 2. Dequeue - eliminar primer elemento
print(f"Dequeue: {cola.popleft()}") # primero
# 3. Front/Peek - ver primer elemento sin eliminar
print(f"Front: {cola[0]}") # segundo
# 4. Rear - ver último elemento
print(f"Rear: {cola[-1]}") # tercero
# 5. Tamaño de la cola
print(f"Tamaño: {len(cola)}") # 2
# 6. Verificar si está vacía
print(f"¿Está vacía? {len(cola) == 0}") # False
# 7. Limpiar la cola
cola.clear()
print(f"¿Está vacía después de clear? {len(cola) == 0}") # TrueEjemplo práctico: Sistema de atención al cliente
from collections import deque
import time
class SistemaAtencion:
def __init__(self):
self.cola_clientes = deque()
self.numero_turno = 1
def llegar_cliente(self, nombre):
"""Un cliente llega y toma un turno"""
turno = f"Turno {self.numero_turno}: {nombre}"
self.cola_clientes.append(turno)
print(f"✅ {nombre} tomó {turno}")
self.numero_turno += 1
def atender_cliente(self):
"""Atender al siguiente cliente en la cola"""
if self.cola_clientes:
cliente = self.cola_clientes.popleft()
print(f"🎯 Atendiendo a: {cliente}")
time.sleep(1) # Simular tiempo de atención
return cliente
else:
print("📭 No hay clientes en espera")
return None
def mostrar_estado(self):
"""Mostrar el estado actual de la cola"""
if self.cola_clientes:
print(f"👥 Clientes en espera ({len(self.cola_clientes)}):")
for i, cliente in enumerate(self.cola_clientes, 1):
print(f" {i}. {cliente}")
else:
print("🟢 No hay clientes en espera")
def siguiente_cliente(self):
"""Ver quién es el siguiente sin atenderlo"""
if self.cola_clientes:
return self.cola_clientes[0]
return "No hay clientes"
# Uso del sistema
sistema = SistemaAtencion()
# Llegan clientes
sistema.llegar_cliente("Ana García")
sistema.llegar_cliente("Carlos López")
sistema.llegar_cliente("Beatriz Martínez")
sistema.mostrar_estado()
# Atender clientes
print("\n--- Iniciando atención ---")
sistema.atender_cliente()
sistema.atender_cliente()
print(f"\n🔜 Siguiente: {sistema.siguiente_cliente()}")
sistema.mostrar_estado()Ejemplo: Procesamiento de tareas
class GestorTareas:
def __init__(self):
self.cola_tareas = deque()
def agregar_tarea(self, tarea, prioridad="normal"):
"""Agregar una tarea a la cola"""
tarea_con_prioridad = f"[{prioridad.upper()}] {tarea}"
self.cola_tareas.append(tarea_con_prioridad)
print(f"📝 Tarea agregada: {tarea_con_prioridad}")
def procesar_tarea(self):
"""Procesar la siguiente tarea en la cola"""
if self.cola_tareas:
tarea = self.cola_tareas.popleft()
print(f"⚙️ Procesando: {tarea}")
# Simular procesamiento
return tarea
else:
print("✅ No hay tareas pendientes")
return None
def urgencia(self, tarea):
"""Agregar una tarea urgente al frente (NO es FIFO puro)"""
tarea_urgente = f"[URGENTE] {tarea}"
self.cola_tareas.appendleft(tarea_urgente)
print(f"🚨 Tarea urgente agregada: {tarea_urgente}")
def estado_tareas(self):
"""Mostrar estado de las tareas"""
print(f"\n📊 Estado de tareas: {len(self.cola_tareas)} pendientes")
for i, tarea in enumerate(self.cola_tareas, 1):
print(f" {i}. {tarea}")
# Uso del gestor de tareas
gestor = GestorTareas()
gestor.agregar_tarea("Revisar correos")
gestor.agregar_tarea("Preparar informe")
gestor.agregar_tarea("Reunión equipo")
gestor.urgencia("Servidor caído") # Tarea urgente
gestor.estado_tareas()
print("\n--- Procesando tareas ---")
gestor.procesar_tarea() # Procesa la urgente primero
gestor.procesar_tarea()
gestor.estado_tareas()Colas especializadas
Cola de prioridad (usando heapq)
import heapq
class ColaPrioridad:
def __init__(self):
self.heap = []
def encolar(self, elemento, prioridad):
heapq.heappush(self.heap, (prioridad, elemento))
def desencolar(self):
if self.heap:
return heapq.heappop(self.heap)[1]
return None
def esta_vacia(self):
return len(self.heap) == 0
# Uso
cp = ColaPrioridad()
cp.encolar("Tarea baja", 3)
cp.encolar("Tarea alta", 1)
cp.encolar("Tarea media", 2)
print(cp.desencolar()) # Tarea alta (prioridad 1)
print(cp.desencolar()) # Tarea media (prioridad 2)Aplicaciones comunes de las colas
Sistemas de atención al cliente
Procesamiento de tareas en background
Manejo de solicitudes en servidores web
Algoritmo BFS (Breadth-First Search)
Buffers de datos
Mensajería entre procesos
Ejemplo: Búsqueda en anchura (BFS)
from collections import deque
def bfs(grafo, inicio):
"""Algoritmo de búsqueda en anchura usando cola"""
visitados = set()
cola = deque([inicio])
visitados.add(inicio)
while cola:
nodo_actual = cola.popleft()
print(f"Visitando: {nodo_actual}")
for vecino in grafo[nodo_actual]:
if vecino not in visitados:
visitados.add(vecino)
cola.append(vecino)
# Grafo ejemplo
grafo = {
'A': ['B', 'C'],
'B': ['A', 'D', 'E'],
'C': ['A', 'F'],
'D': ['B'],
'E': ['B', 'F'],
'F': ['C', 'E']
}
print("Recorrido BFS:")
bfs(grafo, 'A')¿Cuándo usar cada implementación?
deque: Para la mayoría de casos - eficiente en ambos extremosQueue: Programación multi-hilo - es thread-safeListas: Solo para colas pequeñas -
pop(0)es O(n)heapq: Cuando necesitas colas de prioridad
Diferencias clave entre implementaciones
| Implementación | Ventajas | Desventajas | Uso recomendado |
|---|---|---|---|
deque | O(1) en ambos extremos | No es thread-safe | Casos generales |
Queue | Thread-safe | Más lento | Programación concurrente |
| Lista | Simple | pop(0) es O(n) | Colas pequeñas |
Las colas son esenciales en programación para manejar elementos en el orden de llegada, perfectas para sistemas de tickets, procesamiento de solicitudes, y algoritmos de grafos como BFS.
Comentarios
Publicar un comentario