Colas en Python - Colas de prioridad

8.1 Concepto general

Una cola de prioridad atiende primero al elemento con mayor prioridad (o menor coste, según la convención elegida) sin importar el orden de llegada. Mantiene la eficiencia del modelo FIFO en la interfaz pero introduce una relación de orden parcial que requiere estructuras internas específicas.

Se usa en planificadores, routers, simuladores de eventos discretos y algoritmos de grafos como Dijkstra o Prim, donde es vital extraer constantemente el próximo elemento más relevante.

8.2 Interfaz y operaciones

insert o push: incorpora un elemento con su prioridad asociada.
extract o pop: remueve el elemento con mejor prioridad, devolviendo el par valor-prioridad.
peek: consulta el elemento ganador sin extraerlo.
update: opcional; ajusta la prioridad de un elemento existente (decrease-key/increase-key).

Para mantener complejidad logarítmica, la estructura interna suele ser un heap binario; en Python se utiliza el módulo heapq.

8.3 Alternativas de implementación

Array desordenado: inserción O(1) y extracción O(n). Útil cuando hay pocas operaciones pop.
Lista ordenada: inserción O(n), extracción O(1). Adecuado para colas muy pequeñas.
Heap binario: insert/extract en O(log n) y peek en O(1). Este es el balance más habitual.
Montículos especializados: binomial, Fibonacci o pairing heap se emplean cuando hay muchas operaciones decrease-key.

En Python, heapq ofrece operaciones O(log n) para insertar y extraer, apoyándose en una lista como almacenamiento.

8.4 Heap binario en arreglo

En un heap almacenado en una lista, el padre de un nodo en la posición i se encuentra en (i - 1) // 2 y los hijos en 2 * i + 1 y 2 * i + 2. El siguiente esqueleto funciona como min-heap (menor prioridad implica mayor urgencia):

import heapq


class PriorityQueue:
  def __init__(self):
    self._heap = []  # (prioridad, contador, valor)
    self._contador = 0  # rompe empates y preserva orden de llegada

  def push(self, prioridad, valor):
    heapq.heappush(self._heap, (prioridad, self._contador, valor))
    self._contador += 1

  def pop(self):
    if not self._heap:
      return None
    prioridad, _, valor = heapq.heappop(self._heap)
    return prioridad, valor

  def peek(self):
    if not self._heap:
      return None
    prioridad, _, valor = self._heap[0]
    return prioridad, valor

  def __len__(self):
    return len(self._heap)

Los helpers sift_up y sift_down mantienen el invariante del heap tras cada inserción o extracción.

8.5 Insertar y extraer

Insertar y extraer requieren validar la capacidad y reubicar nodos para preservar el orden de prioridad:

class PriorityQueue:
  ...

  def push(self, prioridad, valor):
    heapq.heappush(self._heap, (prioridad, self._contador, valor))
    self._contador += 1

  def pop(self):
    if not self._heap:
      return None
    prioridad, _, valor = heapq.heappop(self._heap)
    return prioridad, valor

  def peek(self):
    if not self._heap:
      return None
    prioridad, _, valor = self._heap[0]
    return prioridad, valor

El costo amortizado se mantiene en O(log n) gracias a las funciones de sift. Si se necesitan prioridades mínimas se intercambia la comparación (> por <).

8.6 Comparadores y estabilidad

Algunos escenarios requieren usar criterios compuestos (por ejemplo, mayor prioridad numérica pero, en caso de empate, menor tiempo de llegada). En esos casos conviene encapsular la comparación en una función:

# Ejemplo de comparador compuesto: menor prioridad y menor llegada primero
def cmp(a, b):
  if a["prioridad"] != b["prioridad"]:
    return -1 if a["prioridad"] < b["prioridad"] else 1
  return -1 if a["llegada"] < b["llegada"] else 1

# heapq no acepta comparadores directamente; se usa una tupla
# (prioridad, llegada, valor) para respetar el criterio.

El comparador define si la cola es estable. Guardar un contador de llegada facilita romper empates sin perder orden cronológico.

8.7 Casos de uso

Planificación preemptiva: los hilos con mayor prioridad se extraen del heap antes que el resto.
Redes: los paquetes con prioridad QoS alta se envían primero.
Algoritmos de grafos: Dijkstra y A* dependen de una cola de prioridad para seleccionar el próximo nodo con camino parcial más corto.

En todos los casos el diseño define si la prioridad es mutable y cómo se sincroniza el acceso concurrente.

8.8 Ejemplo completo en Python

El ejemplo siguiente crea una cola de prioridad de mínima prioridad numérica, inserta varias tareas y luego las procesa en orden:

import heapq


class PriorityQueue:
  def __init__(self):
    self._heap = []
    self._contador = 0

  def push(self, prioridad, valor):
    heapq.heappush(self._heap, (prioridad, self._contador, valor))
    self._contador += 1

  def pop(self):
    if not self._heap:
      return None
    prioridad, _, valor = heapq.heappop(self._heap)
    return prioridad, valor

  def peek(self):
    return None if not self._heap else (self._heap[0][0], self._heap[0][2])

  def __len__(self):
    return len(self._heap)


if __name__ == "__main__":
  pq = PriorityQueue()

  pq.push(5, "soporte estandar")
  pq.push(1, "soporte urgente")
  pq.push(10, "tarea batch")
  pq.push(3, "consulta general")

  print("peek inicial:", pq.peek())

  while len(pq):
    prioridad, valor = pq.pop()
    print(f"Atendiendo prioridad={prioridad} -> {valor}")

Modifica los valores de prioridad y observa cómo el montículo reorganiza los elementos para mantener siempre el máximo en la raíz.

Ejecutar en el editor online

Retornar

8 - Colas de prioridad (Priority Queue)