queue — 동기화된 큐 클래스

소스 코드: Lib/queue.py

queue 모듈은 다중 생산자, 다중 소비자 큐를 구현해요. 특히 여러 스레드 간에 정보를 안전하게 교환해야 하는 스레드 프로그래밍에서 유용해요. 이 모듈의 Queue 클래스는 필요한 모든 잠금 의미(semantics)를 구현해요.

이 모듈은 항목이 검색되는 순서에서만 다른 세 가지 유형의 큐를 구현해요. FIFO 큐에서는 추가된 첫 작업이 가장 먼저 검색돼요. LIFO 큐에서는 가장 최근에 추가된 항목이 가장 먼저 검색돼요(스택처럼 동작). 우선순위 큐에서는 항목이 정렬된 상태로 유지되고(heapq 모듈 사용) 가장 낮은 값을 가진 항목이 가장 먼저 검색돼요.

내부적으로 이 세 가지 유형의 큐는 경쟁하는 스레드를 일시적으로 차단하기 위해 잠금을 사용해요. 다만 스레드 내 재진입(reentrancy)을 처리하도록 설계되지는 않았어요.

또한 이 모듈은 "단순한" FIFO 큐 유형인 SimpleQueue를 구현하는데, 그 구체적 구현은 더 작은 기능과 맞바꿔 추가 보장을 제공해요.

queue 모듈은 다음 클래스와 예외를 정의해요:

출처: Python 표준 라이브러리

class queue.Queue(maxsize=0)

FIFO 큐의 생성자예요. maxsize는 큐에 넣을 수 있는 항목 수의 상한을 설정하는 정수예요. 이 크기에 도달하면 큐 항목이 소비될 때까지 삽입이 차단돼요. maxsize가 0보다 작거나 같으면 큐 크기는 무한이에요.

class queue.LifoQueue(maxsize=0)

LIFO 큐의 생성자예요. maxsize는 큐에 넣을 수 있는 항목 수의 상한을 설정하는 정수예요. 이 크기에 도달하면 큐 항목이 소비될 때까지 삽입이 차단돼요. maxsize가 0보다 작거나 같으면 큐 크기는 무한이에요.

class queue.PriorityQueue(maxsize=0)

우선순위 큐의 생성자예요. maxsize는 큐에 넣을 수 있는 항목 수의 상한을 설정하는 정수예요. 이 크기에 도달하면 큐 항목이 소비될 때까지 삽입이 차단돼요. maxsize가 0보다 작거나 같으면 큐 크기는 무한이에요.

가장 낮은 값을 가진 항목이 가장 먼저 검색돼요(가장 낮은 값의 항목은 min(entries)이 돌려줄 값이에요). 항목의 전형적인 패턴은 (priority_number, data) 형태의 튜플이에요.

데이터 요소가 비교 불가능하면, 데이터 항목을 무시하고 우선순위 번호만 비교하는 클래스로 데이터를 감쌀 수 있어요:

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    item: Any=field(compare=False)

class queue.SimpleQueue

무한 FIFO 큐의 생성자예요. 단순 큐는 작업 추적 같은 고급 기능이 없어요. 단순 큐는 항목 유형에 대해 제네릭(generic)이에요.

버전 3.7에 추가.

exception queue.Empty

비어 있는 Queue 객체에서 비블로킹 get()(또는 get_nowait())이 호출될 때 발생하는 예외예요.

exception queue.Full

가득 찬 Queue 객체에서 비블로킹 put()(또는 put_nowait())이 호출될 때 발생하는 예외예요.

exception queue.ShutDown

종료된(shut down) Queue 객체에서 put() 또는 get()이 호출될 때 발생하는 예외예요.

버전 3.13에 추가.

본문

Queue 객체

Queue, LifoQueue, PriorityQueue 객체는 아래 설명된 공개 메서드를 제공해요.

Queue.qsize()

큐의 대략적인 크기를 돌려줘요. qsize() > 0이 이후 get()이 차단되지 않음을 보장하지 않으며, qsize() < maxsize가 put()이 차단되지 않음을 보장하지 않는다는 점에 주의하세요.

Queue.empty()

큐가 비어 있으면 True, 그렇지 않으면 False를 돌려줘요. empty()가 True를 돌려줘도 이후 put() 호출이 차단되지 않음을 보장하지 않아요. 마찬가지로 empty()가 False를 돌려줘도 이후 get() 호출이 차단되지 않음을 보장하지 않아요.

Queue.full()

큐가 가득 차면 True, 그렇지 않으면 False를 돌려줘요. full()이 True를 돌려줘도 이후 get() 호출이 차단되지 않음을 보장하지 않아요. 마찬가지로 full()이 False를 돌려줘도 이후 put() 호출이 차단되지 않음을 보장하지 않아요.

Queue.put(item, block=True, timeout=None)

item을 큐에 넣어요. 선택적 인자 block이 참이고 timeout이 None(기본)이면 필요할 때 자유 슬롯이 생길 때까지 차단해요. timeout이 양수면 최대 timeout 초 동안 차단하다가, 그 시간 안에 자유 슬롯이 없으면 Full 예외를 발생시켜요. 그렇지 않으면(block이 거짓) 자유 슬롯이 즉시 있으면 항목을 큐에 넣고, 아니면 Full 예외를 발생시켜요(그 경우 timeout은 무시돼요).

큐가 종료되었으면 ShutDown을 발생시켜요.

Queue.put_nowait(item)

put(item, block=False)와 같아요.

Queue.get(block=True, timeout=None)

큐에서 항목을 제거하고 돌려줘요. 선택적 인자 block이 참이고 timeout이 None(기본)이면 필요할 때 항목을 사용할 수 있을 때까지 차단해요. timeout이 양수면 최대 timeout 초 동안 차단하다가, 그 시간 안에 항목이 없으면 Empty 예외를 발생시켜요. 그렇지 않으면(block이 거짓) 즉시 사용할 수 있는 항목이 있으면 돌려주고, 아니면 Empty 예외를 발생시켜요(그 경우 timeout은 무시돼요).

POSIX 시스템의 3.0 이전과 Windows의 모든 버전에서, block이 참이고 timeout이 None이면 이 연산은 기본 잠금에 대해 중단할 수 없는 대기 상태로 들어가요. 이는 예외가 발생할 수 없고, 특히 SIGINT가 KeyboardInterrupt를 일으키지 않는다는 뜻이에요.

큐가 종료되었고 비어 있거나, 큐가 즉시 종료되었으면 ShutDown을 발생시켜요.

Queue.get_nowait()

get(False)와 같아요.

큐에 넣은(enqueued) 작업이 데몬 소비자 스레드에 의해 완전히 처리되었는지 추적을 지원하기 위해 두 메서드가 제공돼요.

Queue.task_done()

이전에 큐에 넣은 작업이 완료되었음을 나타내요. 큐 소비자 스레드가 사용해요. 작업을 가져오는 데 사용된 각 get()에 대해, 이후의 task_done() 호출은 그 작업에 대한 처리가 완료되었음을 큐에 알려줘요. 현재 join()이 차단 중이면 모든 항목이 처리되었을 때(즉, 큐에 put()된 모든 항목에 대해 task_done() 호출이 수신되었을 때) 재개돼요.

큐에 넣은 항목보다 더 많이 호출되면 ValueError를 발생시켜요.

Queue.join()

큐의 모든 항목이 가져와 처리될 때까지 차단해요. 항목이 큐에 추가될 때마다 미완료 작업 수가 올라가요. 소비자 스레드가 항목이 검색되었고 그에 대한 모든 작업이 완료되었음을 나타내는 task_done()을 호출할 때마다 수가 내려가요. 미완료 작업 수가 0이 되면 join()이 차단을 해제해요.

작업 완료 대기 (Waiting for task completion)

큐에 넣은 작업이 완료되기를 기다리는 예제예요:

import threading
import queue

q = queue.Queue()

def worker():
    while True:
        item = q.get()
        print(f'Working on {item}')
        print(f'Finished {item}')
        q.task_done()

# Turn-on the worker thread.
threading.Thread(target=worker, daemon=True).start()

# Send thirty task requests to the worker.
for item in range(30):
    q.put(item)

# Block until all tasks are done.
q.join()
print('All work completed')

큐 종료 (Terminating queues)

더 이상 필요하지 않으면 Queue 객체를 비울 때까지 종료하거나, 하드 셧다운으로 즉시 종료할 수 있어요.

Queue.shutdown(immediate=False)

Queue 인스턴스를 셧다운 모드로 넣어요. 큐는 더 이상 커질 수 없어요. 이후의 put() 호출은 ShutDown을 발생시켜요. 현재 차단된 put() 호출자는 차단 해제되고 이전에 차단된 스레드에서 ShutDown을 발생시켜요.

immediate가 거짓(기본)이면 큐는 get() 호출로 이미 로드된 작업을 추출해 정상적으로 종료될 수 있어요. 그리고 남은 각 작업에 대해 task_done()이 호출되면 보류 중인 join()이 정상적으로 차단 해제돼요. 큐가 비면 이후의 get() 호출은 ShutDown을 발생시켜요.

immediate가 참이면 큐는 즉시 종료돼요. 큐는 완전히 비워지고, 미완료 작업 수는 비워진 작업 수만큼 줄어들어요. 미완료 작업이 0이면 join() 호출자가 차단 해제돼요. 또한 차단된 get() 호출자도 차단 해제되고, 큐가 비어 있으므로 ShutDown을 발생시켜요.

immediate를 참으로 설정해 join()을 사용할 때는 주의하세요. 작업에 대한 작업이 수행되지 않아도 join을 해제하고, 일반적인 큐 결합 불변식을 위반해요.

버전 3.13에 추가.

SimpleQueue 객체

SimpleQueue 객체는 아래 설명된 공개 메서드를 제공해요.

SimpleQueue.qsize()

큐의 대략적인 크기를 돌려줘요. qsize() > 0이 이후 get()이 차단되지 않음을 보장하지 않는다는 점에 주의하세요.

SimpleQueue.empty()

큐가 비어 있으면 True, 그렇지 않으면 False를 돌려줘요. empty()가 False를 돌려줘도 이후 get() 호출이 차단되지 않음을 보장하지 않아요.

SimpleQueue.put(item, block=True, timeout=None)

item을 큐에 넣어요. 이 메서드는 절대 차단하지 않고 항상 성공해요(메모리 할당 실패 같은 잠재적 저수준 오류 외에는). 선택적 인자 block과 timeout은 무시되며 Queue.put()과의 호환성을 위해서만 제공돼요.

CPython 구현 세부 사항: 이 메서드에는 재진입 가능한 C 구현이 있어요. 즉, put() 또는 get() 호출이 같은 스레드의 다른 put() 호출에 의해 교란될 수 있지만 큐 내부 상태가 교착 상태가 되거나 손상되지는 않아요. 이는 __del__ 메서드나 weakref 콜백 같은 소멸자에서 사용하기에 적합하게 만들어줘요.

SimpleQueue.put_nowait(item)

put(item, block=False)와 같고, Queue.put_nowait()과의 호환성을 위해 제공돼요.

SimpleQueue.get(block=True, timeout=None)

SimpleQueue.get_nowait()

get(False)와 같아요.

더 알아보기

multiprocessing.Queue 클래스 — 다중 스레딩이 아닌 다중 프로세싱(multi-processing) 맥락에서 사용하는 큐 클래스.
collections.deque — 잠금이 필요 없고 인덱싱도 지원하는 빠른 원자적 append()와 popleft() 연산을 가진 무한 큐의 대안 구현.
queue (원문)