Data Structures/Python
큐(Queue)
newclass
2025. 3. 26. 04:43
큐(Queue)
개념 소개
큐는 선입선출(FIFO: First-In-First-Out) 원칙을 따르는 자료구조입니다. 이는 실생활에서 줄을 서는 것과 같습니다. 가장 먼저 줄에 선 사람이 가장 먼저 서비스를 받게 됩니다. 큐에 새로운 요소는 항상 뒤쪽(rear)에 추가되고, 요소를 제거할 때는 항상 앞쪽(front)에서 이루어집니다.
큐의 주요 연산:
- enqueue: 큐의 뒤쪽(rear)에 항목을 추가
- dequeue: 큐의 앞쪽(front)에서 항목을 제거하고 반환
- peek/front: 큐의 앞쪽 항목을 제거하지 않고 확인
- isEmpty: 큐가 비어있는지 확인
- size: 큐의 크기 확인
시간 복잡도
연산 시간 복잡도
| Enqueue | O(1) |
| Dequeue | O(1)* |
| Peek/Front | O(1) |
| isEmpty | O(1) |
*일반적인 큐 구현에서는 O(1)이지만, 파이썬 리스트의 시작 부분에서 요소를 제거하는 경우(pop(0))에는 O(n) 시간이 걸릴 수 있습니다.
파이썬에서의 구현
리스트를 이용한 큐 구현 (비효율적)
class Queue:
def __init__(self):
"""큐 초기화"""
self.items = []
def is_empty(self):
"""큐가 비어있는지 확인"""
return len(self.items) == 0
def enqueue(self, item):
"""큐의 뒤쪽에 항목 추가"""
self.items.append(item)
def dequeue(self):
"""큐의 앞쪽에서 항목 제거 및 반환"""
if self.is_empty():
raise IndexError("Dequeue from an empty queue")
return self.items.pop(0) # O(n) 연산 - 비효율적
def peek(self):
"""큐의 앞쪽 항목 확인"""
if self.is_empty():
raise IndexError("Peek from an empty queue")
return self.items[0]
def size(self):
"""큐의 크기 반환"""
return len(self.items)
def __str__(self):
"""큐의 내용을 문자열로 반환 (앞쪽에서 뒤쪽으로)"""
return str(self.items)
# 사용 예
queue = Queue()
queue.enqueue(1)
queue.enqueue(2)
queue.enqueue(3)
print(queue) # [1, 2, 3]
print(queue.dequeue()) # 1
print(queue.peek()) # 2
print(queue.size()) # 2
이 구현은 dequeue 연산이 O(n) 시간 복잡도를 가지기 때문에 효율적이지 않습니다. 리스트의 첫 번째 요소를 제거하면 모든 나머지 요소가 한 칸씩 앞으로 이동해야 하기 때문입니다.
collections.deque를 이용한 큐 구현 (효율적)
파이썬의 collections 모듈에 있는 deque(양방향 큐)는 양쪽 끝에서 O(1) 시간에 요소를 추가하고 제거할 수 있어 큐 구현에 이상적입니다.
from collections import deque
class Queue:
def __init__(self):
"""deque를 사용하여 큐 초기화"""
self.items = deque()
def is_empty(self):
"""큐가 비어있는지 확인"""
return len(self.items) == 0
def enqueue(self, item):
"""큐의 뒤쪽에 항목 추가"""
self.items.append(item)
def dequeue(self):
"""큐의 앞쪽에서 항목 제거 및 반환"""
if self.is_empty():
raise IndexError("Dequeue from an empty queue")
return self.items.popleft() # O(1) 연산 - 효율적
def peek(self):
"""큐의 앞쪽 항목 확인"""
if self.is_empty():
raise IndexError("Peek from an empty queue")
return self.items[0]
def size(self):
"""큐의 크기 반환"""
return len(self.items)
def __str__(self):
"""큐의 내용을 문자열로 반환"""
return str(list(self.items))
# 사용 예
queue = Queue()
queue.enqueue(1)
queue.enqueue(2)
queue.enqueue(3)
print(queue) # [1, 2, 3]
print(queue.dequeue()) # 1
print(queue.peek()) # 2
print(queue.size()) # 2
collections.deque 직접 사용하기
클래스를 만들지 않고 collections.deque를 직접 사용하여 큐를 구현할 수도 있습니다:
from collections import deque
# 큐 초기화
queue = deque()
# 항목 추가(enqueue)
queue.append(1)
queue.append(2)
queue.append(3)
# 큐 내용 확인
print(queue) # deque([1, 2, 3])
# 앞쪽 항목 확인(peek)
if queue:
print(queue[0]) # 1
# 항목 제거(dequeue)
if queue:
print(queue.popleft()) # 1
# 큐가 비어있는지 확인
is_empty = len(queue) == 0
queue 모듈 사용하기
파이썬의 queue 모듈은 다중 생산자와 소비자 스레드를 지원하는 스레드 안전한 큐 구현을 제공합니다.
import queue
# 일반적인 FIFO 큐 생성
q = queue.Queue()
# 항목 추가
q.put(1)
q.put(2)
q.put(3)
# 항목 제거
while not q.empty():
print(q.get()) # 1, 2, 3 순서로 출력
# 우선순위 큐 생성
pq = queue.PriorityQueue()
# 항목 추가 (우선순위, 값)
pq.put((3, 'low priority'))
pq.put((1, 'high priority'))
pq.put((2, 'medium priority'))
# 항목 제거 (우선순위 순서대로)
while not pq.empty():
print(pq.get()[1]) # 'high priority', 'medium priority', 'low priority' 순서로 출력
큐의 특수한 형태
1. 원형 큐 (Circular Queue)
원형 큐는 선형 큐의 단점을 보완하기 위해 배열의 마지막과 첫 번째를 연결하는 형태입니다. 이를 통해 큐가 가득 찼을 때 더 이상 항목을 추가할 수 없는 문제를 해결합니다.
class CircularQueue:
def __init__(self, capacity):
"""원형 큐 초기화"""
self.capacity = capacity
self.items = [None] * capacity
self.front = -1 # 첫 번째 요소의 인덱스
self.rear = -1 # 마지막 요소의 인덱스
def is_empty(self):
"""큐가 비어있는지 확인"""
return self.front == -1
def is_full(self):
"""큐가 가득 찼는지 확인"""
return (self.rear + 1) % self.capacity == self.front
def enqueue(self, item):
"""큐에 항목 추가"""
if self.is_full():
raise IndexError("Queue is full")
if self.is_empty():
self.front = 0
self.rear = (self.rear + 1) % self.capacity
self.items[self.rear] = item
def dequeue(self):
"""큐에서 항목 제거 및 반환"""
if self.is_empty():
raise IndexError("Queue is empty")
item = self.items[self.front]
# 큐에 항목이 하나만 있는 경우
if self.front == self.rear:
self.front = -1
self.rear = -1
else:
self.front = (self.front + 1) % self.capacity
return item
def peek(self):
"""큐의 앞쪽 항목 확인"""
if self.is_empty():
raise IndexError("Queue is empty")
return self.items[self.front]
def size(self):
"""큐의 크기 반환"""
if self.is_empty():
return 0
elif self.front <= self.rear:
return self.rear - self.front + 1
else:
return self.capacity - (self.front - self.rear - 1)
def __str__(self):
"""큐의 내용을 문자열로 반환"""
if self.is_empty():
return "[]"
result = []
i = self.front
while True:
result.append(self.items[i])
if i == self.rear:
break
i = (i + 1) % self.capacity
return str(result)
# 사용 예
circular_queue = CircularQueue(5)
circular_queue.enqueue(1)
circular_queue.enqueue(2)
circular_queue.enqueue(3)
print(circular_queue) # [1, 2, 3]
print(circular_queue.dequeue()) # 1
circular_queue.enqueue(4)
circular_queue.enqueue(5)
print(circular_queue) # [2, 3, 4, 5]
2. 우선순위 큐 (Priority Queue)
우선순위 큐는 각 요소가 우선순위를 가지고 있어, 우선순위가 높은 요소가 먼저 처리되는 큐입니다. 파이썬에서는 heapq 모듈을 사용하여 우선순위 큐를 구현할 수 있습니다.
import heapq
class PriorityQueue:
def __init__(self):
"""우선순위 큐 초기화"""
self.heap = []
self.index = 0 # 같은 우선순위의 항목들을 구분하기 위한 인덱스
def is_empty(self):
"""큐가 비어있는지 확인"""
return len(self.heap) == 0
def enqueue(self, item, priority=0):
"""항목을 우선순위와 함께 추가"""
# (우선순위, 인덱스, 항목) 형태로 저장
# 우선순위가 같을 경우 인덱스가 작은 항목이 먼저 dequeue됨
heapq.heappush(self.heap, (priority, self.index, item))
self.index += 1
def dequeue(self):
"""우선순위가 가장 높은(숫자가 작은) 항목 제거 및 반환"""
if self.is_empty():
raise IndexError("Priority queue is empty")
# heap에서 (우선순위, 인덱스, 항목) 튜플을 pop하고 항목만 반환
return heapq.heappop(self.heap)[2]
def peek(self):
"""우선순위가 가장 높은 항목 확인"""
if self.is_empty():
raise IndexError("Priority queue is empty")
return self.heap[0][2]
def size(self):
"""큐의 크기 반환"""
return len(self.heap)
# 사용 예
priority_queue = PriorityQueue()
priority_queue.enqueue("Task 1", 3)
priority_queue.enqueue("Task 2", 1) # 우선순위가 높음 (숫자가 작음)
priority_queue.enqueue("Task 3", 2)
print(priority_queue.dequeue()) # "Task 2" (우선순위 1)
print(priority_queue.dequeue()) # "Task 3" (우선순위 2)
print(priority_queue.dequeue()) # "Task 1" (우선순위 3)
3. 덱 (Deque - Double-Ended Queue)
덱은 양쪽 끝에서 항목을 추가하고 제거할 수 있는 확장된 형태의 큐입니다. 파이썬의 collections.deque는 이러한 덱을 구현합니다.
from collections import deque
class Deque:
def __init__(self):
"""덱 초기화"""
self.items = deque()
def is_empty(self):
"""덱이 비어있는지 확인"""
return len(self.items) == 0
def add_front(self, item):
"""덱의 앞쪽에 항목 추가"""
self.items.appendleft(item)
def add_rear(self, item):
"""덱의 뒤쪽에 항목 추가"""
self.items.append(item)
def remove_front(self):
"""덱의 앞쪽에서 항목 제거 및 반환"""
if self.is_empty():
raise IndexError("Deque is empty")
return self.items.popleft()
def remove_rear(self):
"""덱의 뒤쪽에서 항목 제거 및 반환"""
if self.is_empty():
raise IndexError("Deque is empty")
return self.items.pop()
def peek_front(self):
"""덱의 앞쪽 항목 확인"""
if self.is_empty():
raise IndexError("Deque is empty")
return self.items[0]
def peek_rear(self):
"""덱의 뒤쪽 항목 확인"""
if self.is_empty():
raise IndexError("Deque is empty")
return self.items[-1]
def size(self):
"""덱의 크기 반환"""
return len(self.items)
def __str__(self):
"""덱의 내용을 문자열로 반환"""
return str(list(self.items))
# 사용 예
deque_obj = Deque()
deque_obj.add_rear(1) # [1]
deque_obj.add_rear(2) # [1, 2]
deque_obj.add_front(0) # [0, 1, 2]
print(deque_obj)
print(deque_obj.remove_front()) # 0, 덱은 [1, 2]
print(deque_obj.remove_rear()) # 2, 덱은 [1]
큐의 활용 사례
큐는 다양한 알고리즘과 실제 애플리케이션에서 널리 사용됩니다:
1. BFS(너비 우선 탐색)
그래프나 트리의 너비 우선 탐색에서 방문해야 할 노드를 저장하는 데 큐가 사용됩니다.
from collections import deque
def bfs(graph, start):
"""큐를 사용한 BFS 구현"""
visited = set()
queue = deque([start])
visited.add(start)
while queue:
vertex = queue.popleft()
print(vertex, end=' ')
# 인접 노드 중 방문하지 않은 노드를 큐에 추가
for neighbor in graph[vertex]:
if neighbor not in visited:
visited.add(neighbor)
queue.append(neighbor)
return visited
# 그래프 예시 (인접 리스트 표현)
graph = {
'A': set(['B', 'C']),
'B': set(['A', 'D', 'E']),
'C': set(['A', 'F']),
'D': set(['B']),
'E': set(['B', 'F']),
'F': set(['C', 'E'])
}
print("BFS:", end=' ')
bfs(graph, 'A') # A B C D E F
2. 작업 스케줄링
운영체제, 웹 서버, 프린터 등에서 작업을 관리하는 데 큐가 사용됩니다.
class PrintQueue:
def __init__(self):
"""프린터 큐 초기화"""
self.queue = deque()
def add_job(self, document):
"""인쇄 작업 추가"""
self.queue.append(document)
print(f"Added {document} to the print queue")
def process_job(self):
"""다음 인쇄 작업 처리"""
if not self.queue:
print("No jobs in the queue")
return None
document = self.queue.popleft()
print(f"Printing {document}")
return document
def jobs_pending(self):
"""대기 중인 작업 수 반환"""
return len(self.queue)
# 사용 예
printer = PrintQueue()
printer.add_job("Report.pdf")
printer.add_job("Resume.docx")
printer.add_job("Presentation.pptx")
print(f"{printer.jobs_pending()} jobs pending")
# 작업 처리
while printer.jobs_pending() > 0:
printer.process_job()
3. 메시지 큐 및 이벤트 처리
분산 시스템에서 메시지를 전달하거나 이벤트를 처리하는 데 큐가 사용됩니다.
class MessageQueue:
def __init__(self):
"""메시지 큐 초기화"""
self.queue = deque()
def publish(self, message):
"""메시지 발행"""
self.queue.append(message)
print(f"Published: {message}")
def consume(self):
"""메시지 소비"""
if not self.queue:
print("No messages in the queue")
return None
message = self.queue.popleft()
print(f"Consumed: {message}")
return message
def size(self):
"""큐의 크기 반환"""
return len(self.queue)
# 사용 예
message_queue = MessageQueue()
message_queue.publish({"type": "login", "user": "alice"})
message_queue.publish({"type": "purchase", "product": "book", "amount": 15.99})
message_queue.publish({"type": "logout", "user": "alice"})
# 메시지 처리
while message_queue.size() > 0:
message_queue.consume()
4. 버퍼 구현
키보드 입력, 네트워크 패킷 등 입출력 버퍼를 구현하는 데 큐가 사용됩니다.
class Buffer:
def __init__(self, capacity):
"""버퍼 초기화"""
self.capacity = capacity
self.buffer = deque(maxlen=capacity) # 최대 크기 제한
def write(self, data):
"""버퍼에 데이터 쓰기"""
if len(self.buffer) == self.capacity:
print("Buffer overflow, oldest data will be replaced")
self.buffer.append(data)
print(f"Written: {data}")
def read(self):
"""버퍼에서 데이터 읽기"""
if not self.buffer:
print("Buffer is empty")
return None
data = self.buffer.popleft()
print(f"Read: {data}")
return data
def peek(self):
"""다음 데이터 확인"""
if not self.buffer:
return None
return self.buffer[0]
def clear(self):
"""버퍼 비우기"""
self.buffer.clear()
def __str__(self):
"""버퍼의 내용을 문자열로 반환"""
return str(list(self.buffer))
# 사용 예
input_buffer = Buffer(3)
input_buffer.write("Hello")
input_buffer.write("World")
input_buffer.write("Python")
print(input_buffer) # ['Hello', 'World', 'Python']
# 버퍼가 가득 찼을 때 쓰기
input_buffer.write("Overflow") # 'Hello'가 삭제되고 'Overflow'가 추가됨
print(input_buffer) # ['World', 'Python', 'Overflow']
5. 레벨 순회
이진 트리를 레벨별로 순회하는 데 큐가 사용됩니다. 트리의 같은 레벨에 있는 노드들을 차례로 방문하는 방법입니다.
from collections import deque
class TreeNode:
def __init__(self, val=0, left=None, right=None):
self.val = val
self.left = left
self.right = right
def level_order_traversal(root):
"""이진 트리의 레벨 순회"""
if not root:
return []
result = []
queue = deque([root])
while queue:
level_size = len(queue)
level = []
for _ in range(level_size):
node = queue.popleft()
level.append(node.val)
if node.left:
queue.append(node.left)
if node.right:
queue.append(node.right)
result.append(level)
return result
# 트리 생성
# 1
# / \
# 2 3
# / \ \
# 4 5 6
root = TreeNode(1)
root.left = TreeNode(2)
root.right = TreeNode(3)
root.left.left = TreeNode(4)
root.left.right = TreeNode(5)
root.right.right = TreeNode(6)
print(level_order_traversal(root)) # [[1], [2, 3], [4, 5, 6]]
큐 실전 문제: 요세푸스 문제
요세푸스 문제는 n명이 원형으로 앉아있을 때, 순서대로 k번째 사람을 제거하는 과정을 반복하여 마지막에 남는 사람을 찾는 문제입니다. 큐를 사용하여 효율적으로 해결할 수 있습니다.
from collections import deque
def josephus(n, k):
"""
요세푸스 문제 해결 함수
n: 사람의 수
k: 제거할 순서 (k번째 사람)
반환값: 마지막 남는 사람의 위치 (1부터 시작)
"""
queue = deque(range(1, n + 1)) # 1부터 n까지의 사람
while len(queue) > 1:
# k-1명은 뒤로 보내고, k번째 사람은 제거
for _ in range(k - 1):
queue.append(queue.popleft())
# k번째 사람 제거
queue.popleft()
# 마지막 남은 사람 반환
return queue[0]
# 예시: 7명 중 3번째 사람을 계속 제거
print(josephus(7, 3)) # 4
큐의 시간 복잡도 비교
구현 방법 Enqueue Dequeue 공간 복잡도 비고
| 리스트(append/pop(0)) | O(1) | O(n) | O(n) | dequeue 시 모든 요소 이동 |
| collections.deque | O(1) | O(1) | O(n) | 양쪽에서 빠른 연산 가능 |
| 원형 큐 | O(1) | O(1) | O(n) | 고정 크기, 메모리 효율적 |
| 연결 리스트 | O(1) | O(1) | O(n) | 동적 크기 조정 가능 |
실생활 큐 시스템 예시
- 커피숍 주문 시스템: 고객이 주문한 순서대로 음료를 준비
- 프린터 대기열: 인쇄 요청이 들어온 순서대로 처리
- 콜센터 대기열: 먼저 전화한 고객부터 상담사에게 연결
- 운영체제의 프로세스 스케줄링: CPU 시간을 기다리는 프로세스 관리
- 네트워크 패킷 처리: 라우터에서 패킷을 받은 순서대로 처리
마무리
큐는 선입선출(FIFO) 원칙을 따르는 기본적이면서도 강력한 자료구조입니다. 적절한 구현 방법을 선택하면 모든 기본 연산이 O(1) 시간 복잡도를 가질 수 있습니다.
파이썬에서는 collections.deque가 가장 효율적인 큐 구현을 제공하며, 다양한 변형(원형 큐, 우선순위 큐, 덱 등)을 통해 다양한 문제를 해결할 수 있습니다.
큐는 특히 너비 우선 탐색(BFS), 작업 스케줄링, 버퍼 관리 등의 알고리즘에서 핵심적인 역할을 합니다. 큐의 개념과 활용법을 잘 이해하면 많은 실제 문제를 효율적으로 해결할 수 있습니다.