우선순위 큐: 실전 heapq

"우선순위 큐는 도착 순서를 무시하고 중요도로 손님을 받는 큐야. 그걸 만드는 방법이 힙이고, Python에서는 이미 만들어져 있어. heapq 모듈이지."

우선순위 큐 ADT

우선순위 큐는 연산이 둘뿐인 추상 자료형이야. 우선순위를 붙여 항목을 넣는 것, 그리고 우선순위가 가장 높은 항목을 꺼내는 것. 오래된 것부터 나가는 평범한 FIFO 큐와 달리 언제나 가장 중요한 원소를 다음에 건네줘. 첫 트랙에서 본 응급실 분류가 정확히 이거지. 힙이 그 정석 구현이고, 삽입 O(log n), 최고 원소 제거 O(log n), 최고 원소 엿보기 O(1)이야. 누가 '우선순위 큐'라고 하면 거의 언제나 '힙'을 뜻하는 거고.

Python의 heapq: 이미 다 만들어져 있어

sift-up이나 sift-down을 직접 구현할 일은 드물어. Python의 heapq가 평범한 리스트 위에서 heappush, heappop, heapify, heappushpop, nlargest와 nsmallest를 제공하거든. 전통적인 API는 최소 힙이고, Python 3.14부터는 heapify_max, heappush_max, heappop_max 같은 최대 힙 API도 들어왔어. 쓰는 버전을 확인하고 고르면 돼.

늘 필요한 두 관용구

최대 힙. Python 3.14 이상이면 heappush_max와 heappop_max를 쓸 수 있어. 그보다 오래된 버전이거나 숫자만 다루는 호환 코드라면 -value를 최소 힙에 넣는 관용구가 여전히 쓸모 있고. 상위 k개가 필요하면 heapq.nlargest도 있어.
튜플로 우선순위와 데이터를 함께. (priority, item)을 push하면 힙이 우선순위 기준으로 정렬해 줘. 그런데 조심할 게 있어. 튜플은 원소를 앞에서부터 차례로 비교하거든. 그래서 우선순위가 같으면 Python이 item끼리 비교하려 들고, 비교가 안 되는 객체면 그대로 죽어 버려. 운 좋게 안 죽어도 의도하지 않은 순서가 나오고. 해결책은 계속 증가하는 카운터를 동점 처리용으로 끼워 넣는 거야. (priority, count, item)을 push하면 count가 매번 다르니까 item이 비교되기 전에 모든 동점이 풀려.

우선순위 큐는 도착 순서가 아니라 중요도로 손님을 받고, 힙이 그걸 구현해. push와 pop이 O(log n)이지. Python의 heapq는 최소 힙을 제공하고 3.14 이상에서는 최대 힙 API도 줘. 숫자에 음수를 씌우는 건 이전 버전 호환 관용구고. 우선순위가 같을 때 데이터끼리 비교되는 일이 절대 없도록 (priority, counter, item) 튜플로 push해.

우선순위 큐가 세상을 굴리는 곳

OS 작업 스케줄러가 우선순위가 가장 높은 프로세스를 돌리고, 이벤트 기반 시뮬레이션이 가장 빨리 일어날 이벤트를 처리하고, 대역폭과 QoS 조절이 돌아가고, A*와 Dijkstra 최단 경로 알고리즘이 다음 트랙에서 가장 가까운 경계 노드를 꺼내고, Huffman 코딩과 'top-k' 분석이 돌아가는 게 전부 우선순위 큐 위에서야. '뭔가가 스트림으로 들어오는데 언제나 가장 중요한 걸 다음에 처리해야 한다'는 구조인데, 실제 시스템에서 정말 흔하게 나오는 모양이야.

피파의 고백

내 첫 heapq 스케줄러는 TypeError: '<' not supported between instances of 'Task'를 뱉으며 죽었어. (priority, task) 튜플을 push했는데 작업 둘의 우선순위가 같아지자 Python이 동점을 깨려고 Task 객체를 비교하려 든 거야. 아빠가 한 줄로 고쳐줬어. 고유한 카운터를 끼워 넣어서 (priority, next(counter), task)로 만든 거지. 그러면 task를 건드리기 전에 동점이 반드시 풀려. 직접 부딪혀 봐야 배우는 종류의 함정이고, 지금은 heapq를 쓸 때 제일 먼저 챙기는 게 이거야.

Code

heapq로 만든 우선순위 스케줄러와 최대 힙 관용구·python

import heapq
from itertools import count

# 우선순위 스케줄러: 작은 수 = 더 급함. heapq 는 최소-힙.
counter = count()                      # 고유, 계속 증가하는 동점 깨기
pq = []
def add_task(priority, name):
    # (priority, 동점깨기, payload): 카운터가 동점을 깨서 같은 우선순위 둘이
    # Python 한테 payload 문자열/객체 비교를 절대 강제 안 해.
    heapq.heappush(pq, (priority, next(counter), name))

add_task(5, "backup")
add_task(1, "page on-call")            # 가장 급함
add_task(5, "cleanup")                 # 우선순위 5 에서 'backup' 이랑 동점
add_task(2, "deploy")

while pq:
    prio, _, name = heapq.heappop(pq)  # 늘 남은 것 중 가장 급함
    print(prio, name)
# 1 page on-call / 2 deploy / 5 backup / 5 cleanup (동점은 도착순으로 깸)

# Python 3.14 전 호환용 부정 최대-힙 관용구:
max_heap = []
for v in [3, 1, 4, 1, 5]:
    heapq.heappush(max_heap, -v)       # 음수를 push
print(-heapq.heappop(max_heap))        # 5 — 나올 때 다시 부정
print(heapq.nlargest(2, [3, 1, 4, 1, 5]))   # [5, 4] — top-k 직접 방법

Exercise

heapq로 '스트림에서 본 가장 큰 수 3개를 유지하기'를 효율적으로 구현해 봐. 스트림 전체를 정렬하면 안 돼. 어떤 종류의 힙을 유지하고, 왜 최대 힙이 아니라 크기 3짜리 최소 힙일까? 그다음 heapq.nlargest(3, stream)이 고정된 리스트에는 괜찮은데 왜 무한히 흘러드는 피드에는 스트리밍 버전이 더 나은지 설명해.

Hint

크기 3짜리 최소 힙을 유지해. 수를 넣고 힙이 3을 넘으면 가장 작은 걸 heappop하는 거야. 최소 힙의 루트가 지금 세 번째로 큰 값이라서 그보다 작은 후보는 싸게 걸러낼 수 있거든. nlargest는 전체 리스트가 메모리에 있어야 하지만, 크기 3짜리 힙은 O(1) 공간으로 스트림을 흘려보내.

우선순위 큐 ADT

Python의 heapq: 이미 다 만들어져 있어

늘 필요한 두 관용구

우선순위 큐가 세상을 굴리는 곳

피파의 고백

Code

External links

Exercise

Progress

댓글 0