파이썬/Basic
자료구조 : deque, Queue, heapq
코샵
2024. 12. 10. 10:37
반응형
소개
Python에서 제공하는 자료구조인 collections.deque, queue.Queue, heapq의 특징과 활용법을 자세히 알아보겠습니다. 각 자료구조의 성능 특성과 적합한 사용 사례를 살펴보겠습니다.
collections.deque
deque(double-ended queue)는 양쪽 끝에서 빠른 삽입과 삭제가 가능한 자료구조입니다.
시간 복잡도
| 연산 | 시간 복잡도 |
|---|---|
| 앞/뒤 삽입/삭제 | O(1) |
| 임의 접근 | O(n) |
| 길이 확인 | O(1) |
예시
from collections import deque
# 기본 사용법
d = deque([1, 2, 3, 4, 5])
d.append(6) # 오른쪽 끝에 추가
d.appendleft(0) # 왼쪽 끝에 추가
d.pop() # 오른쪽 끝에서 제거
d.popleft() # 왼쪽 끝에서 제거
# 최대 길이 제한
limited_deque = deque(maxlen=3)
for i in range(5):
limited_deque.append(i) # 최근 3개 항목만 유지
# 회전 연산
d = deque([1, 2, 3, 4, 5])
d.rotate(2) # [4, 5, 1, 2, 3]
d.rotate(-2) # [1, 2, 3, 4, 5]활용
- 슬라이딩 윈도우 알고리즘
- 최근 이력 관리
- BFS(너비 우선 탐색) 구현
# 슬라이딩 윈도우 예시
def sliding_window_max(nums, k):
result = []
window = deque()
for i, num in enumerate(nums):
# 윈도우 범위를 벗어난 인덱스 제거
if window and window[0] <= i - k:
window.popleft()
# 새로운 값보다 작은 값들 제거
while window and nums[window[-1]] < num:
window.pop()
window.append(i)
if i >= k - 1:
result.append(nums[window[0]])
return resultqueue.Queue
thread-safe한 FIFO(First In First Out) 큐 구현을 제공합니다.
시간 복잡도
| 연산 | 시간 복잡도 |
|---|---|
| 삽입/삭제 | O(1) |
| 크기 확인 | O(1) |
예시
from queue import Queue
import threading
import time
def producer(queue):
"""데이터 생산자"""
for i in range(5):
time.sleep(1)
queue.put(i)
def consumer(queue):
"""데이터 소비자"""
while True:
try:
item = queue.get(timeout=5)
print(f"Processed: {item}")
queue.task_done()
except queue.Empty:
break
# 사용 예시
q = Queue()
producer_thread = threading.Thread(target=producer, args=(q,))
consumer_thread = threading.Thread(target=consumer, args=(q,))
producer_thread.start()
consumer_thread.start()활용
- 멀티스레딩 환경에서의 작업 큐
- 생산자-소비자 패턴
- 이벤트 처리 시스템
heapq
최소 힙(Min Heap) 구현을 제공하는 모듈입니다.
시간 복잡도
| 연산 | 시간 복잡도 |
|---|---|
| 삽입/삭제 | O(log n) |
| 최솟값 조회 | O(1) |
| 힙 생성 | O(n) |
예시
import heapq
# 최소 힙 생성
heap = []
heapq.heappush(heap, 5)
heapq.heappush(heap, 3)
heapq.heappush(heap, 7)
smallest = heapq.heappop(heap) # 3
# 리스트를 힙으로 변환
numbers = [5, 3, 7, 1, 9]
heapq.heapify(numbers)
# 최대 힙 구현
max_heap = []
heapq.heappush(max_heap, (-5, 5))
heapq.heappush(max_heap, (-3, 3))
largest = heapq.heappop(max_heap)[1] # 5
# K번째 작은 수 찾기
def find_kth_smallest(nums, k):
heap = nums[:k]
heapq.heapify(heap)
for num in nums[k:]:
if num < heap[0]:
heapq.heapreplace(heap, num)
return heap[0]활용
- 우선순위 큐 구현
- 다익스트라 알고리즘
- K개 요소 관리
자료구조 선택 가이드
deque 사용이 좋은 경우
- 양쪽 끝에서의 빈번한 삽입/삭제
- 고정 크기의 히스토리 관리
- 메모리 효율성이 중요한 경우
# 최근 N개의 메시지 관리
class MessageHistory:
def __init__(self, max_size):
self.messages = deque(maxlen=max_size)
def add_message(self, message):
self.messages.append(message)
def get_recent(self, n):
return list(self.messages)[-n:]Queue 사용이 좋은 경우
- 멀티스레딩 환경
- 순서가 중요한 작업 처리
- 동기화가 필요한 경우
# 작업 처리 시스템
class TaskProcessor:
def __init__(self):
self.task_queue = Queue()
self.result_queue = Queue()
def add_task(self, task):
self.task_queue.put(task)
def process_tasks(self):
while not self.task_queue.empty():
task = self.task_queue.get()
result = self.process_single_task(task)
self.result_queue.put(result)
self.task_queue.task_done()heapq 사용이 좋은 경우
- 우선순위 기반 처리
- Top-K 문제
- 정렬된 순서로 처리가 필요한 경우
# 우선순위 작업 스케줄러
class PriorityScheduler:
def __init__(self):
self.tasks = []
def add_task(self, priority, task):
heapq.heappush(self.tasks, (-priority, task))
def get_next_task(self):
if self.tasks:
return heapq.heappop(self.tasks)[1]
return None성능 비교
import timeit
# 각 자료구조의 성능 비교
def compare_performance():
# deque vs list
deque_time = timeit.timeit(
'dq.appendleft(0)',
setup='from collections import deque; dq=deque(range(10000))'
)
list_time = timeit.timeit(
'lst.insert(0, 0)',
setup='lst=list(range(10000))'
)
print(f"deque vs list 삽입 시간: {deque_time:.6f} vs {list_time:.6f}")마치며
각 자료구조는 고유한 특징과 장단점을 가지고 있습니다. 사용 사례와 요구사항을 고려하여 적절한 자료구조를 선택하는 것이 중요합니다. 특히 성능이 중요한 애플리케이션에서는 각 자료구조의 시간 복잡도를 고려해야 합니다.