Приоритетная очередь: что это и как реализовать на Python

Коротко

Обычная очередь выдаёт элементы в порядке поступления. Приоритетная — в порядке важности, независимо от того, когда элемент добавили.

Зачем она нужна

Житейская аналогия — приёмное отделение больницы. Люди приходят в разное время, но врач берёт не того, кто пришёл раньше, а того, кому хуже. Пациент с инфарктом обгонит очередь, даже если пришёл последним.

В коде это встречается постоянно:

• планировщик задач: сначала срочные;
• алгоритм Дейкстры: сначала ближайшая вершина;
• обработка событий в симуляции: сначала ближайшее по времени;
• сжатие Хаффмана: сначала самые редкие символы;
• балансировка нагрузки: сначала наименее загруженный сервер.

Базовое использование

В Python приоритетная очередь строится на модуле heapq. Элементы — кортежи (приоритет, данные).

import heapq

queue = []

heapq.heappush(queue, (3, 'помыть посуду'))
heapq.heappush(queue, (1, 'потушить пожар'))
heapq.heappush(queue, (2, 'ответить на письмо'))

while queue:
    priority, task = heapq.heappop(queue)
    print(f'{priority}: {task}')

# 1: потушить пожар
# 2: ответить на письмо
# 3: помыть посуду

Кортежи сравниваются поэлементно, поэтому сортировка идёт по первому элементу — приоритету.

Первая ловушка: несравнимые данные

Если приоритеты совпали, Python перейдёт к сравнению вторых элементов. Для строк это сработает, а для произвольных объектов — упадёт.

class Task:
    def __init__(self, name): self.name = name

queue = []
heapq.heappush(queue, (1, Task('a')))
heapq.heappush(queue, (1, Task('b')))
# TypeError: '<' not supported between instances of 'Task' and 'Task'

Решение — вставить между приоритетом и данными счётчик, который никогда не повторяется:

import heapq
import itertools


class PriorityQueue:
    def __init__(self) -> None:
        self._heap = []
        self._counter = itertools.count()   # 0, 1, 2, ...

    def push(self, item, priority: int) -> None:
        # счётчик разрывает ничьи и не даёт сравнивать item
        heapq.heappush(self._heap, (priority, next(self._counter), item))

    def pop(self):
        if not self._heap:
            raise IndexError('очередь пуста')
        priority, _, item = heapq.heappop(self._heap)
        return item

    def __len__(self) -> int:
        return len(self._heap)


pq = PriorityQueue()
pq.push(Task('первая'), 1)
pq.push(Task('вторая'), 1)
print(pq.pop().name)   # первая

Побочный эффект приятный: очередь становится устойчивой — элементы с одинаковым приоритетом выходят в порядке добавления, потому что счётчик у более раннего меньше.

Вторая ловушка: наибольший приоритет

heapq — это min-куча, она выдаёт наименьшее значение. Если «приоритет 10» должен означать «самый важный», нужен минус:

heapq.heappush(queue, (-priority, count, item))

Это источник трудноуловимых багов: код работает, задачи выполняются, просто в обратном порядке. Проговаривайте направление явно при написании.

Изменение приоритета

Классическая проблема: задача уже в очереди, а её приоритет изменился. Куча не умеет искать элемент — только за O(n).

Стандартный приём — ленивое удаление. Помечаем старую запись недействительной и кладём новую, а при извлечении пропускаем помеченные.

import heapq
import itertools

REMOVED = object()


class UpdatablePriorityQueue:
    def __init__(self) -> None:
        self._heap = []
        self._entries = {}                  # ключ -> запись в куче
        self._counter = itertools.count()

    def push(self, key, priority: int) -> None:
        if key in self._entries:
            self.remove(key)
        entry = [priority, next(self._counter), key]
        self._entries[key] = entry
        heapq.heappush(self._heap, entry)

    def remove(self, key) -> None:
        entry = self._entries.pop(key)
        entry[-1] = REMOVED                 # помечаем, из кучи не вынимаем

    def pop(self):
        while self._heap:
            priority, _, key = heapq.heappop(self._heap)
            if key is not REMOVED:
                del self._entries[key]
                return key
        raise IndexError('очередь пуста')


pq = UpdatablePriorityQueue()
pq.push('задача', 5)
pq.push('задача', 1)      # приоритет повышен
print(pq.pop())           # задача

Запись хранится как список, а не кортеж, — именно чтобы её можно было изменить на месте.

Цена подхода: куча может распухнуть от помеченных записей. Если обновлений много, стоит периодически пересобирать её целиком.

Готовые варианты в стандартной библиотеке

queue.PriorityQueue — потокобезопасная обёртка над heapq с блокировками. Нужна при работе из нескольких потоков, в однопоточном коде только замедляет.

from queue import PriorityQueue

pq = PriorityQueue()
pq.put((1, 'срочно'))
priority, task = pq.get()

heapq.nsmallest и nlargest — если нужно не поддерживать очередь, а разово взять k лучших:

import heapq

data = [(3, 'c'), (1, 'a'), (2, 'b')]
print(heapq.nsmallest(2, data))   # [(1, 'a'), (2, 'b')]

Для маленьких k это эффективнее полной сортировки: O(n log k) против O(n log n).

Практический пример: Дейкстра

Самое частое применение в алгоритмических задачах.

import heapq


def dijkstra(graph: dict, start):
    dist = {start: 0}
    pq = [(0, start)]        # (расстояние, вершина)

    while pq:
        d, node = heapq.heappop(pq)

        if d > dist.get(node, float('inf')):
            continue          # устаревшая запись — пропускаем

        for neighbor, weight in graph[node]:
            new_dist = d + weight
            if new_dist < dist.get(neighbor, float('inf')):
                dist[neighbor] = new_dist
                heapq.heappush(pq, (new_dist, neighbor))

    return dist


graph = {
    'A': [('B', 1), ('C', 4)],
    'B': [('C', 2), ('D', 5)],
    'C': [('D', 1)],
    'D': [],
}
print(dijkstra(graph, 'A'))   # {'A': 0, 'B': 1, 'C': 3, 'D': 4}

Строка if d > dist.get(node, ...) — то самое ленивое удаление в миниатюре. Вместо обновления приоритета мы просто кладём новую запись, а устаревшие отбрасываем при извлечении.

Частые ошибки

Забытый разделитель при равных приоритетах. Падение с TypeError на объектах без __lt__.

Путаница с направлением. Min-куча вместо max-кучи: код работает, порядок обратный.

Попытка изменить приоритет напрямую. Изменение элемента внутри кучи ломает её свойство — нужен heapify заново или ленивое удаление.

PriorityQueue в однопоточном коде. Блокировки дают заметные накладные расходы без всякой пользы.

Что запомнить

• Приоритетная очередь выдаёт элементы по важности, а не по времени добавления.
• Реализуется двоичной кучей: вставка и извлечение за O(log n).
• Храните кортежи (приоритет, счётчик, данные) — счётчик разрывает ничьи и обеспечивает устойчивость.
• heapq даёт только min-кучу; для max-кучи используйте отрицательный приоритет.
• Приоритет не меняют на месте — применяют ленивое удаление.

Пройди собеседование в топ-компанию

Платформа для подготовки

Решай алгоритмические задачи как профи

✓ Популярные алгоритмы✓ Разбор решений✓ AI помощь

Начать сейчас→