큐는 먼저 넣은 항목부터 꺼내는 FIFO 자료형이다. 원형 버퍼 구현과 BFS, 작업 큐의 역압·완료·종료·동시성 정책을 실행 예제와 4,000단계 모델 검사, 16,384개 최단 거리 대조로 설명한다.
개요: 들어온 순서를 처리 순서로 만든다
큐(queue)는 원소를 뒤에 넣고 앞에서 꺼내는 선입선출, 즉 FIFO 순서를 가진 추상 자료형이다. enqueue 또는 put은 대기열에 넣고, dequeue 또는 get은 가장 먼저 들어온 원소를 제거하며 반환한다. 맨 앞을 보기만 하는 peek와 빈 상태·길이 확인도 흔히 제공한다.
예를 들어 읽은 파일 조각을 처리 작업에 넘기거나 그래프에서 가까운 정점부터 방문할 때 대기 순서를 보관할 필요가 있다. 큐는 아직 처리하지 않은 항목을 명시적인 상태로 분리한다. 항목을 넣는 생산자와 꺼내는 소비자가 같은 함수일 수도 있고 서로 다른 스레드나 프로세스일 수도 있다.
이 문서는 FIFO 자료구조에서 시작해 원형 배열 구현, 너비 우선 탐색, 작업 큐의 용량·완료·종료 정책을 설명한다. 메모리 안의 큐와 서버 간 메시지 전달 시스템을 동일시하지 않으며, 순서 규칙과 저장 내구성·재시도 같은 보장은 구분한다.
예제는 Python 3.9.6에서 실행했다. RingQueue와 distances 등 앞에서 정의한 함수는 뒤 검증에서 재사용한다. Python의 최신 버전에만 있는 종료 API를 실행 예제로 사용하지 않고, 버전별 기능은 해당 문서와 대조해야 한다고 명시한다.
FIFO를 연산 순서로 확인하기
A,B,C를 차례로 넣으면 꺼내는 순서는 A,B,C다. A를 꺼낸 뒤 D를 넣더라도 이미 대기하던 B,C가 D보다 먼저 나온다. 반면 같은 목록의 오른쪽 끝에서 넣고 빼면 C부터 나오는 스택이 된다. 컨테이너 이름보다 사용한 끝이 실제 순서를 결정한다.
아래 deque는 오른쪽에 append하고 왼쪽에서 popleft한다. 맨 앞을 읽는 pending[0]은 제거하지 않으므로 첫 출력은 A와 길이 3이다. 이후 A를 꺼내고 D를 넣으면 남은 순서는 B,C,D다.
FIFO는 큐에서 꺼내는 순서의 규칙이다. 소비자가 여러 개이고 작업 시간이 서로 다르면 완료 순서는 달라질 수 있다. A를 먼저 시작했어도 오래 걸리고 B가 빨리 끝나면 결과 발표는 B,A가 될 수 있다. 순서가 필요한 것이 시작인지 완료인지 결과 반영인지 구분해야 한다.
예제 코드
from collections import deque
pending = deque(['A', 'B', 'C'])
print(pending[0], len(pending))
print(pending.popleft())
pending.append('D')
print(list(pending))
print([pending.popleft() for _ in range(3)])실행 결과
A 3
A
['B', 'C', 'D']
['B', 'C', 'D']추상 자료형과 라이브러리의 이름
자료구조 입문에서 큐는 보통 FIFO를 뜻하지만 라이브러리에서 Queue라는 이름이 항상 FIFO만 허용하는 것은 아니다. 우선순위 큐는 우선순위에 따라, LIFO 큐는 최근 항목부터 꺼낼 수 있다. 인터페이스가 공통이어도 구현마다 순서 보장을 확인해야 한다.
Java Queue 인터페이스도 각 구현이 정렬 규칙을 명시하도록 한다. 예외를 내는 add·remove·element와 특별한 결과를 반환하는 offer·poll·peek가 구분된다. 메서드의 실패 계약이 다른 API를 이름만 보고 바꾸면 빈 상태나 용량 초과 처리가 달라질 수 있다.
양방향 큐인 덱은 양 끝 삽입·삭제를 제공하여 FIFO 큐와 LIFO 스택을 모두 구현할 수 있다. 하지만 FIFO가 요구되는 모듈에서 중간 삭제나 앞 삽입을 외부에 마음대로 허용하면 순서 규칙을 지키기 어렵다. 필요한 연산만 감싼 인터페이스를 두면 의도가 명확해진다.
메시지 브로커와 작업 스케줄러도 큐라는 말을 쓰지만 영속 저장, 확인 응답, 분산 소비, 재시도와 처리 순서 같은 추가 기능이 있다. 메모리 컨테이너의 O(1) 삽입 분석을 그대로 네트워크 서비스의 지연 보장으로 옮기지 않는다.
앞 원소를 매번 당기는 구현의 비용
Python list에서 append와 pop(0)을 사용하면 FIFO 결과 자체는 맞다. 그러나 맨 앞을 제거할 때 뒤 원소의 참조를 앞으로 이동해야 한다. 길이 n의 목록을 끝까지 비우면 이동량은 (n-1)+(n-2)+…+0으로 이차가 된다.
아래 카운터는 실제 CPU 명령 수가 아니라 매 삭제에서 당겨야 하는 원소 수의 모델이다. 길이 4면 6개, 길이 8이면 28개를 이동한다. 작은 예제에서는 차이가 눈에 띄지 않아도 대기열이 길어지면 같은 자료를 반복 이동하는 비용이 커진다.
읽기 위치를 앞으로 옮기기만 하고 실제 저장소를 그대로 둘 수도 있다. 이 경우 매번 이동하는 비용은 줄지만 이미 소비한 참조가 남아 메모리를 붙잡거나 배열이 계속 커질 수 있다. 일정 시점에 압축하거나 원형 버퍼로 공간을 재사용하는 정책이 필요하다.
표준 deque는 양끝 처리를 위한 구현을 제공하므로 단일 스레드의 일반적인 FIFO 작업에 간단하게 사용할 수 있다. 임의 인덱스 접근·검색이 잦다면 그 연산의 비용도 함께 비교해야 한다. 한 구조가 모든 연산에서 동시에 최적인 것은 아니다.
예제 코드
def shifted_slots(n):
total = 0
while n:
total += n - 1
n -= 1
return total
for n in [4, 8]:
print(n, shifted_slots(n), n * (n - 1) // 2)실행 결과
4 6 6
8 28 28원형 배열: 논리 순서와 물리 위치
원형 배열은 배열 끝에 도달하면 인덱스를 처음으로 되돌려 사용한 공간을 재활용한다. 실제 메모리가 원 모양이라는 뜻은 아니다. 길이 C의 배열에서 논리 위치 i의 물리 인덱스를 (head+i)%C로 계산하는 규칙이다.
head는 다음에 꺼낼 원소의 위치이고 size는 현재 원소 수다. 새 항목의 위치 tail은 (head+size)%C다. 꺼낼 때 head를 한 칸 전진시키고 size를 줄이며, 넣을 때 tail에 쓰고 size를 늘린다. 다른 원소를 앞으로 당길 필요가 없다.
아래 용량 5,head 3,size 4이면 논리 원소의 위치는 3,4,0,1이다. 다음 삽입 위치는 2다. 물리 배열을 앞에서부터 출력한 순서와 사용자가 보는 큐 순서가 다를 수 있으므로 디버깅할 때 두 관점을 구분한다.
size를 따로 저장하면 head와 tail이 같아도 빈 상태인지 가득 찬 상태인지 구별할 수 있다. 다른 구현은 슬롯 하나를 비워 두거나 별도 full 플래그를 사용한다. 서로 다른 방식의 조건을 섞으면 용량을 한 칸 잘못 계산하거나 미처리 값을 덮는 오류가 생긴다.
예제 코드
capacity, head, size = 5, 3, 4
print([(head + i) % capacity for i in range(size)])
print((head + size) % capacity)실행 결과
[3, 4, 0, 1]
2고정 용량 FIFO 구현
아래 RingQueue는 양의 정수 용량을 받아 슬롯을 미리 확보한다. 빈 get·peek는 IndexError, 가득 찬 put은 OverflowError로 표현하며 실패한 put은 기존 원소를 바꾸지 않는다. None도 정상 데이터로 허용하므로 빈 슬롯 여부로 길이를 추측하지 않는다.
get은 값을 저장한 뒤 해당 슬롯을 None으로 비우고 head를 이동한다. 값을 제거한 뒤에도 참조를 남기면 논리적으로는 비어 있는 큐가 객체를 계속 붙잡을 수 있다. to_list는 물리 배열 대신 논리 순서로 복사한 목록을 반환하므로 O(size)다.
생성은 슬롯 수에 비례한 시간·공간이 필요하다. 생성 이후 put·get·peek는 고정된 수의 인덱스 계산과 대입을 수행한다. 여기서 상수 시간은 컨테이너 연산 모델이며 객체 정리, 메모리 접근과 시스템 스케줄링까지 포함한 실시간 지연 상한은 아니다.
이 코드는 교육용 단일 스레드 구현이다. 내부 필드를 외부에서 변경하거나 여러 스레드가 동시에 쓰는 상황을 지원하지 않는다. 실제 라이브러리에는 타입 계약, 동기화, 대기와 종료 정책을 더해야 하지만 그 기능과 FIFO의 핵심 불변식은 나누어 이해할 수 있다.
예제 코드
class RingQueue:
def __init__(self, capacity):
if capacity <= 0:
raise ValueError('capacity must be positive')
self._data = [None] * capacity
self._head = 0
self._size = 0
def __len__(self):
return self._size
def put(self, value):
if self._size == len(self._data):
raise OverflowError('queue full')
tail = (self._head + self._size) % len(self._data)
self._data[tail] = value
self._size += 1
def peek(self):
if self._size == 0:
raise IndexError('queue empty')
return self._data[self._head]
def get(self):
value = self.peek()
self._data[self._head] = None
self._head = (self._head + 1) % len(self._data)
self._size -= 1
return value
def to_list(self):
return [self._data[(self._head + i) % len(self._data)]
for i in range(self._size)]
q = RingQueue(3)
for item in ['A', 'B', 'C']:
q.put(item)
print(q.peek(), len(q))
print(q.get())
q.put('D')
print(q.to_list())실행 결과
A 3
A
['B', 'C', 'D']끝을 넘어 돌아오는 동작 추적
앞의 클래스 정의 뒤에 아래 코드를 실행한다. A,B,C를 넣은 뒤 A,B를 꺼내면 head는 2이고 C만 남는다. D,E를 넣으면 물리 배열은 [D,E,C]가 되지만 논리 순서는 [C,D,E]다. 처음 빈 두 칸을 재사용했기 때문이다.
이제 세 번 get하면 C,D,E 순서로 나온다. 물리 인덱스가 2,0,1 순으로 움직였어도 FIFO가 유지된다. 큐의 불변식은 유효 원소가 head에서 시작한 size개의 원형 구간에 순서대로 있다는 것이다.
가득 찼을 때 tail이 head와 같다는 이유로 빈 상태라고 판단하면 C를 덮어쓸 수 있다. size를 기준으로 빈 상태와 full을 검사해야 한다. 용량 1은 매번 같은 슬롯을 사용하므로 이런 조건 오류를 찾기 좋은 경계 사례다.
랩어라운드는 처음 한 번만 발생하는 특별한 경우가 아니다. 긴 운영에서는 여러 번 반복된다. 단순한 초기 삽입·삭제 테스트뿐 아니라 가득 참과 비어 있음, 경계 넘기를 섞은 연산열을 검증해야 한다.
예제 코드
q = RingQueue(3)
for item in ['A', 'B', 'C']:
q.put(item)
print(q.get(), q.get())
q.put('D')
q.put('E')
print(q._data, q._head, q.to_list())
print(q.get(), q.get(), q.get())
print(len(q), q._data)실행 결과
A B
['D', 'E', 'C'] 2 ['C', 'D', 'E']
C D E
0 [None, None, None]확장 가능한 원형 큐의 재배치
고정 용량으로 부족하면 더 큰 버퍼를 확보하는 정책을 추가할 수 있다. 이때 물리 배열의 앞부분부터 그대로 복사하면 논리 순서가 뒤틀릴 수 있다. head에서 시작한 유효 원소를 순서대로 새 배열의 0번부터 옮겨야 한다. 옮긴 뒤 head=0으로 초기화한다.
아래 grow는 앞의 RingQueue 내부 표현을 이용한 설명용 보조 함수다. [D,E,C]의 물리 배열과 head 2에서 확장하면 새 배열 앞부분은 [C,D,E]가 된다. 그 뒤 F를 추가해도 순서가 유지된다. 구현 제품에서는 이를 클래스 안에 감추고 자동 확장 조건을 정한다.
용량을 고정 비율로 늘리면 끝 추가의 재배치 비용을 분할 상환 분석할 수 있다. 한 번의 확장에는 현재 원소 수에 비례한 복사가 있으므로 개별 요청의 최대 지연은 남는다. 축소 정책도 임계점에 여유를 두어 추가·삭제마다 확장과 축소가 진동하지 않게 한다.
메모리 상한이 있는 작업 큐는 무조건 확장하는 것이 맞지 않을 수 있다. 소비자가 계속 느린 상태에서 버퍼만 늘리면 지연과 메모리 사용을 뒤로 미룰 뿐이다. 저장 구조의 성장 정책과 시스템의 과부하 정책을 함께 설계한다.
예제 코드
def grow(queue):
old_capacity = len(queue._data)
new_data = [None] * (old_capacity * 2)
for i in range(len(queue)):
new_data[i] = queue._data[(queue._head + i) % old_capacity]
queue._data = new_data
queue._head = 0
q = RingQueue(3)
for item in ['A', 'B', 'C']:
q.put(item)
q.get(); q.get()
q.put('D'); q.put('E')
grow(q)
q.put('F')
print(q.to_list(), len(q._data), q._head)실행 결과
['C', 'D', 'E', 'F'] 6 0연결 큐와 양끝 포인터
단일 연결 리스트로 큐를 구현할 때 앞 노드와 뒤 노드를 모두 유지하면 앞 제거와 뒤 추가를 빠르게 할 수 있다. 뒤 포인터 없이 매번 마지막 노드를 찾으면 추가할 때 전체 순회가 필요해진다. 어떤 보조 정보를 유지하는지가 연산 비용을 바꾼다.
빈 큐에 첫 노드를 넣을 때는 앞과 뒤가 같은 노드를 가리킨다. 마지막 노드를 꺼내면 앞뿐 아니라 뒤도 빈 상태로 되돌려야 한다. 이 처리를 빠뜨리면 이미 제거한 노드를 통해 다음 노드를 연결하거나 사용이 끝난 객체를 붙잡는 오류가 생길 수 있다.
연결 구현은 원형 인덱스 계산이 필요 없고 고정 연속 공간을 미리 확보하지 않아도 된다. 대신 노드별 연결 정보와 할당 비용, 메모리 지역성의 차이가 생긴다. 단순히 복사가 없다는 이유만으로 배열 기반 큐보다 항상 빠르다고 결론 내릴 수 없다.
실제 선택은 최대 대기량, 원소 크기, 할당자, 순회 여부와 참조 안정성에 달려 있다. 많은 작은 노드를 만드는 방식과 큰 버퍼를 재사용하는 방식의 메모리 부담은 다르다. 추상 FIFO 규칙과 구현의 비용을 분리하면 환경에 맞는 선택을 할 수 있다.
너비 우선 탐색에서 큐가 필요한 이유
가중치가 없는 그래프의 너비 우선 탐색은 시작점에서 간선 수가 적은 정점부터 방문한다. 현재 거리 d의 정점을 꺼내면 처음 발견한 이웃의 거리는 d+1이다. FIFO 큐가 앞서 발견한 같은 층과 다음 층의 순서를 유지하므로 가까운 후보를 뒤의 먼 후보보다 먼저 처리한다.
아래 distances는 result 사전을 거리와 방문 표시로 함께 사용한다. 이웃을 처음 큐에 넣는 순간 거리를 기록하여 여러 부모가 같은 정점을 중복해서 넣지 않게 한다. 나중에 꺼낼 때까지 방문 표시를 미루면 대기 항목이 불필요하게 늘 수 있다.
A에서 B,C는 거리 1, D는 거리 2다. D가 다시 A를 가리키는 순환이 있어도 이미 result에 있는 정점은 다시 넣지 않는다. 입력은 정점에서 이웃 목록으로 가는 매핑이며 정점은 해시 가능한 값이라는 전제를 사용한다.
인접 목록에서 도달 가능한 정점과 간선을 각각 한정된 횟수 처리하므로 기대 상수 시간 사전 연산을 가정하면 시간 O(V+E)다. 거리 사전과 큐는 O(V) 공간이다. 그래프 전체를 복사하거나 입력을 만드는 비용은 이 함수 호출의 별도 전처리 비용이다.
예제 코드
from collections import deque
def distances(graph, start):
result = {start: 0}
pending = deque([start])
while pending:
vertex = pending.popleft()
for neighbor in graph.get(vertex, []):
if neighbor not in result:
result[neighbor] = result[vertex] + 1
pending.append(neighbor)
return result
graph = {'A': ['B', 'C'], 'B': ['D'], 'C': ['D'], 'D': ['A']}
print(distances(graph, 'A'))실행 결과
{'A': 0, 'B': 1, 'C': 1, 'D': 2}최단 경로 복원과 도달 불가능한 경우
거리뿐 아니라 실제 경로를 구하려면 정점을 처음 발견한 부모를 저장한다. 목표에 도착한 뒤 부모를 역으로 따라가면 목표에서 시작점으로 가는 순서가 나오므로 마지막에 뒤집는다. 정점마다 경로 목록 전체를 복사해 큐에 넣는 것보다 중복 저장을 줄일 수 있다.
아래 예제는 문자열 정점을 사용하고 None을 시작점 부모의 종료 표시로 사용한다. 따라서 None 자체를 정점으로 사용하는 그래프는 이 구현의 지원 범위 밖이다. 일반적인 정점 값을 모두 지원하려면 고유한 센티널 객체나 시작점 비교 등으로 계약을 바꾼다.
A에서 D까지 여러 최단 경로가 있으면 이웃 목록 순서에 따라 하나를 반환한다. 모든 최단 경로를 열거하는 문제는 별도의 부모 집합과 출력량을 고려해야 한다. 시작점과 목표가 같으면 시작점 하나인 경로이며, 도달할 수 없으면 None이다.
함수의 반환값은 경로의 길이가 최소라는 뜻이지 정점 이름의 사전순이 최소라는 뜻은 아니다. 결과가 여러 개일 수 있는 알고리즘에서는 어떤 동률 규칙을 보장하는지 구분해야 테스트를 불필요하게 특정 경로에 묶지 않는다.
예제 코드
def shortest_path(graph, start, goal):
from collections import deque
parent = {start: None}
pending = deque([start])
while pending:
vertex = pending.popleft()
if vertex == goal:
route = []
while vertex is not None:
route.append(vertex)
vertex = parent[vertex]
return route[::-1]
for neighbor in graph.get(vertex, []):
if neighbor not in parent:
parent[neighbor] = vertex
pending.append(neighbor)
return None
graph = {'A': ['B', 'C'], 'B': ['D'], 'C': ['D'], 'D': []}
print(shortest_path(graph, 'A', 'D'))
print(shortest_path(graph, 'A', 'A'))
print(shortest_path(graph, 'A', 'Z'))실행 결과
['A', 'B', 'D']
['A']
NoneFIFO가 모든 최단 경로를 해결하지는 않는다
일반 BFS가 최소화하는 것은 지나간 간선 수다. 간선마다 비용이 다르면 간선 수가 적은 경로가 비용도 적다는 보장은 없다. A에서 B로 바로 가는 비용이 10이고 A→C→B가 각각 1이라면 한 간선 경로의 비용 10보다 두 간선 경로의 비용 2가 작다.
이 경우 단순 FIFO에서 처음 발견한 B의 값을 최종 답으로 고정하면 잘못될 수 있다. 비음수 일반 가중치에는 거리 우선순위를 관리하는 방법이 필요하고, 0과 1만 있는 경우에는 덱을 이용하는 특수한 0-1 BFS가 가능하다. 각 알고리즘의 전제를 확인해야 한다.
자료구조 선택은 정확성과 연결된다. FIFO 큐를 우선순위 큐로 바꾸기만 하고 거리 갱신·오래된 항목 처리·방문 확정 시점은 그대로 두면 또 다른 오류가 생길 수 있다. 다음 항목을 고르는 규칙과 상태 갱신의 근거를 함께 바꿔야 한다.
아래 숫자 비교는 코드의 속도 문제가 아니라 목적함수의 차이를 보여준다. 문제의 길이가 간선 수인지 거리·시간·요금의 합인지 먼저 정의하고 적절한 탐색 방법을 선택한다.
예제 코드
direct_edges, direct_cost = 1, 10
via_edges, via_cost = 2, 1 + 1
print(direct_edges < via_edges)
print(direct_cost > via_cost)실행 결과
True
True다중 시작점과 격자 탐색
여러 시작점에서 가장 가까운 거리를 구할 때는 모든 시작점을 거리 0으로 먼저 큐에 넣고 탐색할 수 있다. 시작점마다 전체 BFS를 별도로 실행하는 대신 한 번의 파동처럼 층을 확장한다. 이미 등록한 시작점은 중복해서 넣지 않는다.
격자에서는 각 칸을 정점, 허용한 이동을 간선으로 볼 수 있다. 네 방향 이동인지 대각선도 허용하는지, 장애물이 무엇인지에 따라 그래프가 달라진다. 좌표가 범위 안인지 확인하고 방문 표시를 넣는 시점을 일관되게 정한다.
아래는 길이 7의 일렬 칸에서 양 끝 0과 6을 시작점으로 삼아 가장 가까운 끝까지의 거리를 구한다. 결과 0,1,2,3,2,1,0은 가운데로 갈수록 거리가 늘어나는 모습을 보여준다. 모든 칸의 이동 비용이 같다는 전제를 사용한다.
최단 거리 외에 어느 시작점에서 왔는지 소유자 표시를 함께 전파할 수도 있다. 거리가 같은 경우 시작점 삽입 순서나 이웃 순서에 따라 선택이 달라질 수 있으므로 특정 동률 규칙이 필요하면 명시적으로 구현한다.
예제 코드
from collections import deque
length = 7
starts = [0, 6]
distance = {s: 0 for s in starts}
pending = deque(distance)
while pending:
cell = pending.popleft()
for neighbor in [cell - 1, cell + 1]:
if 0 <= neighbor < length and neighbor not in distance:
distance[neighbor] = distance[cell] + 1
pending.append(neighbor)
print([distance[i] for i in range(length)])실행 결과
[0, 1, 2, 3, 2, 1, 0]라운드 로빈: 남은 작업을 뒤로 보내기
작업을 한 번에 모두 처리하지 않고 작은 시간 조각만 처리한 뒤 남은 작업을 큐 뒤에 넣으면 라운드 로빈 방식의 순환 처리를 만들 수 있다. 이미 기다리던 다른 작업에게 차례를 주기 위해 아직 끝나지 않은 작업을 맨 뒤로 돌려보낸다.
아래 A의 작업량은 3,B는 1,C는 2이고 한 번에 1씩 줄인다. 실행 순서는 A,B,C,A,C,A다. B는 한 번에 끝나고 C는 두 번째 차례에 끝나며 A는 세 번 필요하다. FIFO는 작업 전체의 완료 순서가 아니라 각 차례를 배분하는 데 사용된다.
실제 CPU 스케줄러를 그대로 구현한 예제는 아니다. 문맥 전환 비용, 우선순위, I/O 대기와 새 작업 도착을 생략한 결정적 모델이다. 시간 조각이 너무 작으면 전환 비용이 커지고 너무 크면 다른 작업의 대기가 길어질 수 있다는 설계 관점을 익히는 예다.
남은 작업량이 0인 항목을 다시 넣으면 끝나지 않는 반복을 만들 수 있다. 양의 작업량, 처리량과 종료 조건을 명시하고, 외부 작업이 실패하거나 취소되는 경우 어떤 상태로 큐에 돌아가는지도 정의해야 한다.
예제 코드
from collections import deque
pending = deque([('A', 3), ('B', 1), ('C', 2)])
order = []
while pending:
name, remaining = pending.popleft()
order.append(name)
remaining -= 1
if remaining:
pending.append((name, remaining))
print(order)실행 결과
['A', 'B', 'C', 'A', 'C', 'A']대기열은 처리 능력을 만들어내지 않는다
생산자가 초당 100개를 넣고 소비자가 초당 80개를 처리하는 상황이 계속되면 대기량은 대략 초당 20개씩 늘어난다. 큐를 추가했다고 이 지속적인 처리량 차이가 사라지는 것은 아니다. 큐는 일시적인 폭주를 흡수하고 구성 요소를 분리하지만 무한한 과부하를 해결하지 않는다.
긴 대기열은 메모리와 응답 지연을 늘린다. 이미 사용자가 포기한 작업을 뒤늦게 처리하는 문제가 생길 수도 있다. 최대 대기량, 작업 만료 시간과 취소 정책을 함께 정해야 한다. 작은 작업 수라도 각 항목이 큰 객체나 파일 내용을 붙잡으면 실제 메모리는 클 수 있다.
아래 모델은 매 단계 5개가 들어오고 최대 3개를 처리한다고 가정한다. 대기량은 2,4,6,8로 증가한다. 도착·처리 순서와 초기 상태를 고정한 교육용 계산이며 확률적인 실제 트래픽의 지연 예측 모델은 아니다.
처리량 개선에는 소비자 병렬화, 작업당 비용 감소, 입력 제한과 불필요한 작업 제거 같은 접근이 있다. 소비자를 늘려도 공통 데이터베이스나 잠금이 병목이면 효과가 제한될 수 있다. 큐의 길이만 보지 말고 어디서 시간이 쓰이는지 관찰한다.
예제 코드
waiting = 0
for _ in range(4):
waiting += 5
completed = min(waiting, 3)
waiting -= completed
print(waiting)실행 결과
2
4
6
8용량 제한과 역압 정책
용량이 가득 찼을 때 생산자를 기다리게 하면 소비자의 처리 속도가 생산자에게 전달된다. 이를 역압이라고 부른다. 모든 데이터를 받아 메모리에 계속 쌓는 대신 앞단에서 대기하거나 요청을 거절하여 자원을 제한하는 선택이다.
차단, 제한 시간 후 실패, 즉시 거절, 오래된 항목 버림, 새 항목 버림은 서로 다른 정책이다. 결제 요청과 최신 센서 상태는 잃어도 되는 정보의 의미가 다르므로 같은 버림 정책을 적용할 수 없다. 큐 구현은 정책을 제공할 수 있지만 어떤 것을 선택할지는 업무 요구사항이다.
아래 queue.Queue(maxsize=2)는 대기 항목을 두 개로 제한하고 put_nowait는 공간이 없으면 Full을 발생시킨다. 세 번째 항목이 조용히 사라지는 것이 아니라 호출자가 실패를 처리할 수 있다. 이미 꺼내 처리 중인 항목은 보통 이 대기 용량에 포함되지 않으므로 전체 동시 작업량 상한과 구별한다.
차단 put을 사용하면 생산자도 멈출 수 있다. 같은 스레드가 큐를 채운 뒤 나중에 소비자를 시작하는 구조에서는 가득 찬 시점에 진행이 막힐 수 있다. 생산·소비의 실행 순서와 종료 경로를 함께 설계한다.
예제 코드
from queue import Queue, Full
jobs = Queue(maxsize=2)
jobs.put_nowait('A')
jobs.put_nowait('B')
try:
jobs.put_nowait('C')
except Full:
print('full')
print(jobs.get_nowait(), jobs.get_nowait())실행 결과
full
A Bempty 확인과 get 사이의 경쟁
동시에 여러 소비자가 접근하면 empty가 False였다는 사실이 다음 get의 성공을 보장하지 않는다. 확인 직후 다른 소비자가 마지막 항목을 가져갈 수 있기 때문이다. qsize 같은 관찰 결과도 이후 연산이 수행될 때까지 유지되는 예약이 아니다.
이럴 때는 get_nowait를 직접 호출하고 Empty를 처리하거나, 필요한 제한 시간을 가진 차단 get을 사용한다. 확인과 제거를 한 번에 수행하는 API의 계약을 이용하는 것이다. 직접 만든 RingQueue는 이러한 동기화를 제공하지 않으므로 같은 패턴을 그대로 스레드 안전 구현이라고 부를 수 없다.
락을 사용하더라도 개별 put·get을 보호하는 것과 여러 단계의 업무 절차를 원자적으로 보호하는 것은 다르다. 항목을 꺼낸 뒤 데이터베이스를 갱신하고 결과를 알리는 과정은 큐의 내부 잠금 밖에서 별도의 실패 처리가 필요하다.
아래 예제는 대기하지 않는 제거의 빈 상태 예외를 보여준다. 실제 경쟁 상황을 재현하는 테스트는 아니며 결과가 결정적인 단일 스레드 예제다. 동시성 문제를 설명할 때 실행하지 않은 경쟁 테스트를 수행한 것처럼 표현하지 않는다.
예제 코드
from queue import Queue, Empty
jobs = Queue()
try:
item = jobs.get_nowait()
except Empty:
print('nothing available')실행 결과
nothing available꺼냄과 완료는 다르다: task_done과 join
큐에서 항목을 꺼내면 대기열의 길이는 줄지만 작업이 성공적으로 끝났다는 뜻은 아니다. 소비자는 꺼낸 후 계산·저장·전송을 수행할 수 있고 그 사이에 실패할 수 있다. 대기 항목 수와 미완료 작업 수를 구별해야 한다.
Python Queue의 작업 추적은 put마다 미완료 수를 늘리고 task_done마다 줄인다. join은 이 수가 0이 될 때까지 기다린다. get만 호출하고 task_done을 빠뜨리면 큐가 비어 있어도 join이 끝나지 않을 수 있다. 반대로 한 항목에 task_done을 두 번 호출하면 계산이 틀어진다.
task_done은 라이브러리의 완료 계수에 대한 신호다. 업무 성공·실패를 자동 기록하거나 재시도를 예약해 주는 기능은 아니다. 실패를 결과 채널에 남기고 완료를 표시할지, 다른 큐로 재시도할지 정책을 정해야 한다.
작업 중 예외가 나더라도 정상적으로 꺼낸 항목에 대응하는 계수 정리가 필요하면 try/finally를 사용한다. 다만 finally에서 완료를 알렸다는 사실을 성공으로 해석해서는 안 된다. 실패 정보와 완료 계수는 별도로 관리한다.
실제 스레드 생산자·소비자 예제
다음 예제는 소비자 스레드 하나와 용량 2의 큐를 사용한다. 생산자는 정수 1부터 4까지 넣고 마지막에 고유한 STOP 객체를 넣는다. 소비자는 각 값을 제곱해 결과에 기록하고 STOP을 만나면 종료한다. None도 정상 데이터일 수 있는 API에서는 고유 객체가 종료 표시로 더 명확하다.
소비자를 먼저 시작하므로 생산자가 가득 찬 큐에서 기다리더라도 소비자가 공간을 만들 수 있다. 모든 항목과 종료 표시에 대해 task_done을 정확히 한 번 호출한다. jobs.join은 작업 계수가 0이 되기를 기다리고 worker.join은 스레드 자체의 종료를 기다린다는 차이가 있다.
소비자가 하나라 결과 순서는 [1,4,9,16]이다. 여러 소비자로 바꾸면 완료 순서는 달라질 수 있고 보통 소비자 수만큼 종료 표시를 보내는 등의 추가 종료 설계가 필요하다. 실패가 일어난 소비자가 조용히 사라지지 않도록 오류 전달도 마련해야 한다.
이 코드는 실제로 스레드를 실행해 확인한 작은 예제다. 장시간 운영에서의 취소·재시도·예외 복구·충돌 입력까지 검증한 서비스 구현은 아니다. 업무 함수를 일반화할 때는 정상 종료뿐 아니라 생산자 실패와 소비자 실패 경로를 별도로 검토한다.
예제 코드
from queue import Queue
from threading import Thread
jobs = Queue(maxsize=2)
STOP = object()
results = []
def consume():
while True:
item = jobs.get()
try:
if item is STOP:
return
results.append(item * item)
finally:
jobs.task_done()
worker = Thread(target=consume)
worker.start()
for value in [1, 2, 3, 4]:
jobs.put(value)
jobs.put(STOP)
jobs.join()
worker.join()
print(results, worker.is_alive())실행 결과
[1, 4, 9, 16] False비동기 큐와 스레드 큐를 혼용하지 않기
asyncio.Queue는 async/await 작업 흐름을 위한 큐다. await put과 await get은 코루틴이 공간이나 데이터를 기다리는 동안 이벤트 루프가 다른 작업을 진행할 수 있게 한다. 스레드 간 공유를 위한 queue.Queue와 목적이 다르며 asyncio 큐 자체는 스레드 안전 API가 아니다.
이벤트 루프 안에서 오래 차단되는 동기 get을 직접 호출하면 다른 코루틴도 진행하지 못할 수 있다. 반대로 일반 스레드에서 asyncio 큐의 메서드를 임의로 호출한다고 안전한 스레드 간 전달이 되는 것은 아니다. 실행 모델에 맞는 도구와 연결 방법을 선택한다.
아래는 소비자 코루틴 하나와 용량 1의 큐다. 생산자가 연속으로 넣더라도 공간이 없으면 await가 소비자에게 진행 기회를 준다. STOP을 만나면 소비자가 끝나고 join과 task 대기로 마무리한다. 외부 네트워크나 인위적인 sleep 없이 동작 순서만 확인한다.
취소와 제한 시간은 별도 정책이다. 기다리던 작업이 취소되면 무엇을 처리한 것으로 볼지, 꺼낸 항목을 다시 넣을지 정해야 한다. Python 버전에 따라 큐 종료 API가 추가되었으므로 사용하는 버전의 문서를 확인하고 예제의 센티널 방식과 혼동하지 않는다.
예제 코드
import asyncio
async def main():
jobs = asyncio.Queue(maxsize=1)
stop = object()
results = []
async def consume():
while True:
item = await jobs.get()
try:
if item is stop:
return
results.append(item + 10)
finally:
jobs.task_done()
task = asyncio.create_task(consume())
for value in [1, 2, 3]:
await jobs.put(value)
await jobs.put(stop)
await jobs.join()
await task
return results
print(asyncio.run(main()))실행 결과
[11, 12, 13]종료·시간 제한·취소의 설계
작업이 더 오지 않는다는 사실은 큐가 잠시 비어 있다는 사실과 다르다. 소비자가 empty를 보고 종료하면 생산자가 나중에 넣을 작업을 놓칠 수 있다. 생산 종료를 나타내는 별도 신호나 라이브러리 종료 API가 필요하다.
정상 종료는 이미 받은 작업을 모두 처리하고 새 입력을 막는 방식일 수 있고, 즉시 중단은 대기 작업을 포기하는 방식일 수 있다. 두 의미를 구분하고 미완료 작업 계수와 결과 보고를 맞춘다. 큐를 비웠다는 사실만으로 모든 작업이 처리되었다고 보고하지 않는다.
제한 시간 초과도 작업이 실행되지 않았다는 확정은 아닐 수 있다. 호출자가 기다리기를 포기한 시점에 소비자가 이미 시작했을 수 있다. 작업 ID와 상태를 추적하여 취소 요청을 관찰하도록 설계하거나, 중복 요청이 들어와도 결과를 일관되게 처리하는 방법이 필요하다.
여러 생산자·소비자가 있으면 종료 신호를 누가 언제 보낼지 정한다. 생산자가 모두 끝나기 전에 STOP을 넣으면 남은 작업보다 먼저 소비자가 종료할 수 있다. 단순 예제를 병렬화할 때 가장 먼저 검토해야 하는 부분이다.
재시도·중복 처리·내구성은 추가 보장이다
메모리 큐에서 값을 꺼낸 직후 프로세스가 종료되면 그 작업은 큐에서도 사라지고 처리 결과도 남지 않을 수 있다. 자료구조 FIFO만으로 장애 이후의 복구나 정확히 한 번 처리를 보장하지 않는다. 저장 내구성과 처리 완료 확인은 별도 프로토콜이 필요하다.
재시도는 실패를 줄이지만 같은 작업이 두 번 실행될 가능성을 만든다. 작업 ID로 중복을 확인하거나 같은 요청을 여러 번 적용해도 결과가 같도록 멱등성을 설계할 수 있다. 외부 결제나 이메일처럼 되돌릴 수 없는 부수 효과에서는 처리 경계와 확인 기록의 관계가 특히 중요하다.
실패한 항목을 맨 앞에 무한히 돌려놓으면 다른 항목이 계속 밀릴 수 있다. 최대 재시도 수, 지연과 별도 실패 보관 경로를 두어 정상 흐름과 분리할 수 있다. 오류 유형을 구별하지 않고 모든 실패를 즉시 재시도하면 이미 과부하인 시스템을 더 악화시킬 수 있다.
이 절은 설계상의 구분을 설명하며 특정 메시지 브로커의 보장을 선언하지 않는다. 실제 제품에서는 설정, 확인 응답 시점, 복제와 저장 정책을 공식 문서로 확인해야 한다. 단순 RingQueue 예제에 이러한 분산 보장이 있다고 읽어서는 안 된다.
관찰 지표와 처리 순서의 실제 의미
운영에서 유용한 지표는 현재 대기량뿐 아니라 가장 오래 기다린 항목의 나이, 입력·처리율, 처리 시간 분포, 거절·실패·재시도 수다. 대기량이 작아도 한 항목이 오랫동안 막혀 있을 수 있고, 대기량이 커도 빠르게 소진되는 일시적 폭주일 수 있다.
FIFO는 특정 큐에 들어온 순서를 보존하지만 여러 생산자가 동시에 넣으면 외부 사건의 실제 발생 순서와 큐가 받아들인 순서가 다를 수 있다. 여러 큐나 파티션을 사용하면 전체 순서를 다시 정할 규칙이 필요하다. 시간 기록만으로 완전한 전역 순서를 보장한다고 가정하지 않는다.
작업 완료 순서를 입력 순서로 맞추려면 일련번호로 결과를 모으고 앞선 결과가 준비될 때까지 기다릴 수 있다. 이때 느린 작업 하나가 뒤의 빠른 결과 공개를 막는 대기 문제가 생긴다. 처리 병렬성과 결과 순서 사이의 교환을 요구사항에 맞게 선택한다.
측정값을 기록할 때 대기 시간과 실행 시간을 구분한다. 큐에 오래 머문 것인지 소비자가 실제로 오래 실행한 것인지 알아야 용량·병렬도·작업 비용 중 무엇을 바꿔야 할지 판단할 수 있다. 숫자 하나보다 원인을 설명할 수 있는 관찰이 중요하다.
모델 기반 검사: 원형 큐의 연산열
다음 검사는 용량 1,2,3,7의 RingQueue를 Python deque와 비교한다. 각 용량에서 고정 시드로 1,000회 넣기와 꺼내기를 시도한다. 꽉 찬 put과 빈 get의 오류도 확인하고 매 단계마다 전체 순서·길이·head 범위를 비교한다.
지원한 값에는 None이 포함되므로 빈 슬롯 표식과 정상 값이 충돌하지 않는지 확인한다. 단순히 정수만 넣는 검사보다 내부 표현에 대한 잘못된 가정을 찾는 데 유리하다. 실패한 put 이후 기존 원소가 보존되는지도 전체 상태 비교에 포함된다.
검사 수는 총 4,000단계다. 같은 Python 환경에서 시드를 고정하여 재현할 수 있으며 버전이 달라지면 난수 도구의 보장을 확인한다. 이 검사는 단일 스레드 구현의 경계 처리이며 동시성이나 메모리 부족을 검증한 것은 아니다. 앞의 RingQueue 정의 뒤에 실행한다.
예제 코드
from collections import deque
from random import Random
steps = 0
for capacity in [1, 2, 3, 7]:
rng = Random(capacity)
q = RingQueue(capacity)
expected = deque()
for _ in range(1000):
if rng.randrange(2):
value = rng.choice([None, -1, 0, 8])
if len(expected) == capacity:
try:
q.put(value)
except OverflowError:
pass
else:
raise AssertionError('full queue accepted input')
else:
q.put(value)
expected.append(value)
elif expected:
assert q.get() == expected.popleft()
else:
try:
q.get()
except IndexError:
pass
else:
raise AssertionError('empty queue returned a value')
assert q.to_list() == list(expected)
assert len(q) == len(expected)
assert 0 <= q._head < capacity
steps += 1
print(steps)실행 결과
4000BFS를 독립적인 최단 거리 계산과 대조하기
BFS 구현은 작은 방향 그래프를 완전 열거해 다른 방식의 거리 계산과 비교할 수 있다. 정점 4개에서 자기 자신을 제외한 방향 간선은 12개이고, 각 간선의 존재 여부 조합은 2^12=4,096개다. 각 그래프의 네 시작점에서 BFS를 실행한다.
기준 계산은 거리 행렬에서 중간 정점을 하나씩 허용하는 Floyd–Warshall 방식이다. BFS와 다른 제어 흐름으로 모든 쌍의 거리를 계산하므로 큐의 순서나 방문 표시 실수를 독립적으로 점검할 수 있다. 도달 불가능한 거리는 충분히 큰 값 99로 둔다. 이 작은 그래프의 유한 최단 경로 길이를 넘는 값이다.
총 16,384개의 시작점별 거리 사전을 비교한다. 순환, 단절, 여러 경로와 빈 간선 그래프가 포함된다. 이 검사 통과가 가중 그래프나 변경 중인 그래프까지 지원한다는 뜻은 아니다. 앞의 distances 정의 뒤에 실행하며 일반 입력의 정확성은 BFS 층별 불변식으로 별도 설명한다.
예제 코드
edges = [(a, b) for a in range(4) for b in range(4) if a != b]
checks = 0
for mask in range(1 << len(edges)):
graph = {a: [] for a in range(4)}
matrix = [[0 if a == b else 99 for b in range(4)] for a in range(4)]
for bit, (a, b) in enumerate(edges):
if mask & (1 << bit):
graph[a].append(b)
matrix[a][b] = 1
for middle in range(4):
for a in range(4):
for b in range(4):
matrix[a][b] = min(matrix[a][b], matrix[a][middle] + matrix[middle][b])
for start in range(4):
expected = {b: matrix[start][b] for b in range(4) if matrix[start][b] < 99}
assert distances(graph, start) == expected
checks += 1
print(checks)실행 결과
16384선택과 구현을 마무리하는 기준
먼저 필요한 순서가 FIFO인지, 가장 최근 항목인지, 우선순위인지 정한다. 다음으로 최대 대기량과 가득 찬 상태의 행동, 비었을 때 대기 여부, 데이터 손실 허용 여부를 정한다. 같은 큐라는 이름 아래 서로 다른 계약을 섞지 않는 것이 출발점이다.
단일 실행 흐름의 알고리즘에는 deque가 간단한 선택이 될 수 있고, 일정 용량의 버퍼를 세밀하게 관리하려면 원형 배열을 사용할 수 있다. 스레드·코루틴·프로세스 간 전달에는 해당 실행 모델을 지원하는 도구를 선택한다. 저장 내구성이 필요하면 단순 메모리 컨테이너를 넘어선 설계가 필요하다.
구현에서는 논리 순서와 물리 위치, 빈 상태와 가득 참, 꺼냄과 완료, 대기와 종료를 각각 구분한다. 작은 용량과 반복 랩어라운드를 테스트하고, 실제 업무에서는 실패·취소·초과 입력이 데이터와 계수에 어떤 영향을 주는지 확인한다.
큐는 작업을 잠시 보관하는 곳인 동시에 시스템의 처리 능력과 과부하 정책이 드러나는 경계다. 빠른 put·get만 구현하는 데서 끝나지 않고, 어떤 항목을 언제까지 처리할지와 실패를 어떻게 드러낼지를 설명할 수 있어야 실제 프로그램에서 신뢰할 수 있게 사용할 수 있다.