기쁘달

우선순위 큐 (Priority Queue)

10 min read

개념

우선순위 큐 - 우선순위가 높은 요소가 먼저 나가는 큐

각 요소는 우선순위 값을 가지며, 우선순위에 따라 처리

시간복잡도

연산힙 구현배열 구현연결 리스트
삽입O(log n)O(1)O(n)
삭제O(log n)O(n)O(1)
최댓값 조회O(1)O(n)O(1)

실무에서는 힙으로 구현

특징

일반 큐 vs 우선순위 큐

일반 큐 (FIFO):
삽입: A B C D
삭제: A B C D

우선순위 큐:
삽입: A(3) B(1) C(4) D(2)
삭제: C(4) A(3) D(2) B(1)

장점

  • 중요도 기반 처리: 우선순위 높은 작업 먼저 처리
  • 효율적 구현: 힙으로 O(log n) 보장
  • 동적 관리: 실시간 우선순위 변경 가능

단점

  • 순서 보장 X: FIFO 순서가 아님
  • 검색 비효율: 특정 원소 찾기 어려움

Python 구현

1. heapq 사용 (기본)

import heapq
 
# 우선순위 큐 생성
pq = []
 
# 삽입 (우선순위, 데이터)
heapq.heappush(pq, (3, 'task A'))
heapq.heappush(pq, (1, 'task B'))
heapq.heappush(pq, (4, 'task C'))
heapq.heappush(pq, (2, 'task D'))
 
# 삭제 (우선순위 낮은 순)
while pq:
    priority, task = heapq.heappop(pq)
    print(f"{task}: {priority}")
 
# 출력:
# task B: 1
# task D: 2
# task A: 3
# task C: 4
 
# 시간복잡도: O(log n)

2. 최대 우선순위 큐

import heapq
 
# 음수로 변환하여 최대 힙처럼 사용
max_pq = []
 
heapq.heappush(max_pq, (-3, 'task A'))
heapq.heappush(max_pq, (-1, 'task B'))
heapq.heappush(max_pq, (-4, 'task C'))
 
# 우선순위 높은 순으로 처리
while max_pq:
    priority, task = heapq.heappop(max_pq)
    print(f"{task}: {-priority}")
 
# 출력:
# task C: 4
# task A: 3
# task B: 1

3. queue.PriorityQueue 사용

from queue import PriorityQueue
 
# 우선순위 큐 생성
pq = PriorityQueue()
 
# 삽입
pq.put((3, 'task A'))
pq.put((1, 'task B'))
pq.put((4, 'task C'))
 
# 삭제
while not pq.empty():
    priority, task = pq.get()
    print(f"{task}: {priority}")
 
# 출력:
# task B: 1
# task A: 3
# task C: 4
 
# 특징: 스레드 안전 (멀티스레딩 환경)

4. 클래스로 구현

import heapq
 
class PriorityQueue:
    def __init__(self):
        self.heap = []
        self.index = 0  # 동일 우선순위 처리용
 
    def push(self, priority, item):
        # (우선순위, 삽입 순서, 데이터)
        heapq.heappush(self.heap, (priority, self.index, item))
        self.index += 1
 
    def pop(self):
        if self.heap:
            return heapq.heappop(self.heap)[2]  # 데이터만 반환
        return None
 
    def peek(self):
        if self.heap:
            return self.heap[0][2]
        return None
 
    def is_empty(self):
        return len(self.heap) == 0
 
    def size(self):
        return len(self.heap)
 
# 테스트
pq = PriorityQueue()
pq.push(3, 'A')
pq.push(1, 'B')
pq.push(2, 'C')
 
print(pq.pop())  # B
print(pq.pop())  # C
print(pq.pop())  # A

자주 나오는 문제 유형

1. CPU 작업 스케줄링

문제: 작업의 우선순위와 실행시간이 주어질 때 처리 순서

import heapq
 
def schedule_tasks(tasks):
    """
    tasks: [(priority, duration, name), ...]
    우선순위 높은 순으로 처리
    """
    pq = []
    for priority, duration, name in tasks:
        # 우선순위 낮은 값이 먼저 나오도록 음수
        heapq.heappush(pq, (-priority, duration, name))
 
    total_time = 0
    result = []
 
    while pq:
        priority, duration, name = heapq.heappop(pq)
        total_time += duration
        result.append((name, total_time))
 
    return result
 
# 테스트
tasks = [(3, 5, 'A'), (1, 2, 'B'), (4, 3, 'C'), (2, 4, 'D')]
print(schedule_tasks(tasks))
# [('C', 3), ('A', 8), ('D', 12), ('B', 14)]
 
# 시간복잡도: O(n log n)

2. K개의 가장 가까운 점

문제: 원점에서 가장 가까운 K개의 점 찾기

import heapq
 
def k_closest_points(points, k):
    """
    points: [(x, y), ...]
    """
    pq = []
 
    for x, y in points:
        distance = x * x + y * y
        # 최대 힙으로 K개 유지
        heapq.heappush(pq, (-distance, x, y))
 
        if len(pq) > k:
            heapq.heappop(pq)
 
    return [(x, y) for _, x, y in pq]
 
# 테스트
points = [(1, 3), (-2, 2), (5, 8), (0, 1)]
print(k_closest_points(points, 2))
# [(0, 1), (-2, 2)]
 
# 시간복잡도: O(n log k)

3. 실시간 순위표 (Top K)

문제: 스트리밍 데이터에서 상위 K개 유지

import heapq
 
class TopKTracker:
    def __init__(self, k):
        self.k = k
        self.heap = []
 
    def add(self, score):
        if len(self.heap) < self.k:
            heapq.heappush(self.heap, score)
        elif score > self.heap[0]:
            heapq.heapreplace(self.heap, score)
 
    def get_top_k(self):
        return sorted(self.heap, reverse=True)
 
# 테스트
tracker = TopKTracker(3)
for score in [10, 20, 15, 30, 5, 25]:
    tracker.add(score)
 
print(tracker.get_top_k())  # [30, 25, 20]
 
# 시간복잡도: O(log k) - 삽입

4. 이벤트 시뮬레이션

문제: 시간 순으로 이벤트 처리

import heapq
 
def simulate_events(events):
    """
    events: [(time, event_name), ...]
    """
    pq = []
    for time, name in events:
        heapq.heappush(pq, (time, name))
 
    result = []
    while pq:
        time, name = heapq.heappop(pq)
        result.append(f"Time {time}: {name}")
 
    return result
 
# 테스트
events = [(5, 'Meeting'), (2, 'Call'), (7, 'Lunch'), (2, 'Email')]
for event in simulate_events(events):
    print(event)
 
# 출력:
# Time 2: Call
# Time 2: Email
# Time 5: Meeting
# Time 7: Lunch
 
# 시간복잡도: O(n log n)

5. 회의실 배정 II

문제: 최소 회의실 개수 구하기

import heapq
 
def min_meeting_rooms(intervals):
    if not intervals:
        return 0
 
    # 시작 시간 기준 정렬
    intervals.sort(key=lambda x: x[0])
 
    # 종료 시간을 저장할 최소 힙
    heap = []
    heapq.heappush(heap, intervals[0][1])
 
    for i in range(1, len(intervals)):
        # 가장 빨리 끝나는 회의가 현재 회의 시작 전에 끝나면
        if heap[0] <= intervals[i][0]:
            heapq.heappop(heap)
 
        # 현재 회의 종료 시간 추가
        heapq.heappush(heap, intervals[i][1])
 
    return len(heap)
 
# 테스트
intervals = [(0, 30), (5, 10), (15, 20)]
print(min_meeting_rooms(intervals))  # 2
 
# 시간복잡도: O(n log n)

6. 합병 K개의 정렬된 배열

문제: K개의 정렬된 배열을 하나의 정렬된 배열로

import heapq
 
def merge_k_sorted_arrays(arrays):
    pq = []
 
    # 각 배열의 첫 원소를 힙에 추가
    for i, arr in enumerate(arrays):
        if arr:
            heapq.heappush(pq, (arr[0], i, 0))
 
    result = []
 
    while pq:
        val, array_idx, element_idx = heapq.heappop(pq)
        result.append(val)
 
        # 다음 원소를 힙에 추가
        if element_idx + 1 < len(arrays[array_idx]):
            next_val = arrays[array_idx][element_idx + 1]
            heapq.heappush(pq, (next_val, array_idx, element_idx + 1))
 
    return result
 
# 테스트
arrays = [[1, 4, 5], [1, 3, 4], [2, 6]]
print(merge_k_sorted_arrays(arrays))
# [1, 1, 2, 3, 4, 4, 5, 6]
 
# 시간복잡도: O(N log k) - N: 전체 원소 수, k: 배열 개수

추천 연습 문제

기초

중급

고급

핵심 요약

개념설명
우선순위 큐우선순위 높은 요소가 먼저 나감
구현힙으로 구현 (O(log n))
활용스케줄링, 실시간 순위, 이벤트 처리

heapq vs queue.PriorityQueue

특성heapqqueue.PriorityQueue
구현모듈 함수클래스
스레드 안전XO
성능빠름느림 (락 오버헤드)
사용 시기단일 스레드멀티스레딩

추천: 대부분의 경우 heapq 사용

우선순위 큐 구현 방법 비교

1. 배열 (정렬되지 않음)

  • 삽입: O(1)
  • 삭제: O(n) - 최댓값 찾기
  • 공간: O(n)

2. 배열 (정렬됨)

  • 삽입: O(n) - 정렬 유지
  • 삭제: O(1)
  • 공간: O(n)

3. 연결 리스트

  • 삽입: O(n) - 위치 찾기
  • 삭제: O(1)
  • 공간: O(n)

4. 힙 (최적)

  • 삽입: O(log n)
  • 삭제: O(log n)
  • 공간: O(n)

활용 분야

1. 운영체제

  • CPU 스케줄링
  • 프로세스 관리
  • 인터럽트 처리

2. 네트워크

  • 패킷 전송 우선순위
  • QoS (Quality of Service)
  • 라우팅 알고리즘

3. 게임 개발

  • 이벤트 시스템
  • AI 행동 결정
  • 애니메이션 우선순위

4. 데이터 처리

  • 실시간 순위
  • Top K 문제
  • 중앙값 찾기

주의사항

1. 동일 우선순위 처리

# 삽입 순서 보장
heapq.heappush(pq, (priority, index, data))

2. 비교 불가능한 객체

# 튜플의 첫 번째 요소로 비교
class Task:
    def __init__(self, priority, name):
        self.priority = priority
        self.name = name
 
# 사용 시
heapq.heappush(pq, (task.priority, task))

3. 우선순위 업데이트

# 힙에서 직접 업데이트 불가
# 방법 1: 삭제 후 재삽입
# 방법 2: lazy deletion (삭제 플래그)

4. 최대/최소 선택

# 최소 힙: 작은 값이 높은 우선순위
heapq.heappush(pq, (value, data))
 
# 최대 힙: 큰 값이 높은 우선순위
heapq.heappush(pq, (-value, data))

우선순위 큐 템플릿

기본 템플릿

import heapq
 
pq = []
 
# 삽입
heapq.heappush(pq, (priority, data))
 
# 삭제
priority, data = heapq.heappop(pq)
 
# 조회
priority, data = pq[0]
 
# 비어있는지 확인
if pq:
    # 처리

동일 우선순위 처리

import heapq
 
pq = []
index = 0
 
# 삽입
heapq.heappush(pq, (priority, index, data))
index += 1
 
# 삭제
priority, _, data = heapq.heappop(pq)

최대 우선순위 큐

import heapq
 
pq = []
 
# 삽입
heapq.heappush(pq, (-priority, data))
 
# 삭제
priority, data = heapq.heappop(pq)
priority = -priority  # 원래 값으로 복원

그래프 뷰

백링크