우선순위 큐: 실전 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 헬퍼. 두 가지 박아둬: 최소-힙 이고 (가장 작은 게 먼저 pop), 평범한 리스트를 제자리에서 다뤄. 실전 90% 엔 그게 API 전부야.

늘 필요할 두 관용구

부정으로 최대-힙. heapq 는 최소만. 가장 큰 게 먼저 나오길 원해? -value 를 push 하고 나올 때 다시 부정해. ('top-k 가장 큰' 엔 heapq.nlargest(k, data) 가 바로 해줘.)
튜플로 우선순위 + 페이로드. (priority, item) 을 push 하면 힙이 우선순위로 정렬해. 근데 조심: 튜플은 원소별로 비교해, 그래서 두 우선순위가 동점이면 Python 이 item 을 비교하려 해 — 비교 불가능하면 크래시 (또는 의도 안 한 순서). 고침은 단조 카운터를 동점 깨기로: (priority, count, item) 을 push, count 가 고유해서 item 이 비교되기 전에 모든 동점을 깨.

우선순위 큐는 도착이 아니라 중요도로 모시고; 힙이 그걸 구현해 (O(log n) push/pop). Python 에선 heapq 를 써 — 리스트 위 최소-힙. 최대-힙엔 부정; 같은 우선순위가 페이로드 비교를 절대 강제 안 하게 (priority, counter, item) 튜플을 push.

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

OS 작업 스케줄러 (가장 높은 우선순위 준비 프로세스 실행), 이벤트 구동 시뮬레이션 (다음-가장-빠른 이벤트 처리), 대역폭/QoS 셰이핑, A* 와 Dijkstra 최단 경로 알고리즘 (다음 트랙이 가장 가까운 frontier 노드를 pop), 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 (동점은 도착순으로 깸)

# 부정으로 최대-힙 (heapq 는 최소만):
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 최소-힙 유지: 각 수 push, 힙이 3 초과면 가장 작은 걸 heappop. 최소-힙 루트가 현재 3번째로 큰 거라, 더 작은 건 싸게 퇴출. nlargest 는 전체 리스트가 메모리에 필요; 크기 3 힙은 O(1) 공간으로 스트림.

우선순위 큐 ADT

Python 의 heapq: 이미 다 됨

늘 필요할 두 관용구

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

피파의 고백

Code

External links

Exercise

Progress

댓글 0