Python의 queue 모듈은 여러 스레드가 동시에 접근하더라도 데이터 손상 없이 안전하게 항목을 주고받을 수 있는 큐(Queue) 객체를 제공합니다. 큐는 생산자(Producer)-소비자(Consumer) 패턴을 구현할 때 특히 유용하며, 최대 저장 용량을 지정하여 생성할 수도 있습니다.
큐 객체를 제어하기 위해 사용되는 주요 메서드는 다음과 같습니다.
- get() – 큐에서 항목 하나를 꺼내어(제거하고) 반환합니다.
- put() – 큐에 새 항목을 추가합니다.
- qsize() – 현재 큐에 들어 있는 항목의 개수를 반환합니다.
- empty() – 큐가 비어 있으면 True, 그렇지 않으면 False를 반환합니다.
- full() – 큐가 가득 차 있으면 True, 그렇지 않으면 False를 반환합니다.
멀티스레드 큐 예제
아래 예제는 세 개의 워커 스레드가 하나의 공유 큐에서 데이터를 가져와 처리하는 과정을 보여줍니다. threading.Lock()을 사용해 여러 스레드가 동시에 큐에 접근하는 것을 방지하며, Python 3 기준 문법(import queue)으로 작성했습니다.
#!/usr/bin/python3
import queue
import threading
import time
exitFlag = 0
class myThread(threading.Thread):
def __init__(self, threadID, name, q):
threading.Thread.__init__(self)
self.threadID = threadID
self.name = name
self.q = q
def run(self):
print("Starting " + self.name)
process_data(self.name, self.q)
print("Exiting " + self.name)
def process_data(threadName, q):
while not exitFlag:
queueLock.acquire()
if not workQueue.empty():
data = q.get()
queueLock.release()
print("%s processing %s" % (threadName, data))
else:
queueLock.release()
time.sleep(1)
threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = threading.Lock()
workQueue = queue.Queue(10)
threads = []
threadID = 1
# 새로운 스레드 생성 및 시작
for tName in threadList:
thread = myThread(threadID, tName, workQueue)
thread.start()
threads.append(thread)
threadID += 1
# 큐에 작업 항목 채우기
queueLock.acquire()
for word in nameList:
workQueue.put(word)
queueLock.release()
# 큐가 비워질 때까지 대기
while not workQueue.empty():
pass
# 스레드에 종료 신호 전달
exitFlag = 1
# 모든 스레드가 끝날 때까지 대기
for t in threads:
t.join()
print("Exiting Main Thread")실행 결과
위 코드를 실행하면 다음과 같은 출력 결과를 얻을 수 있습니다. 실제 실행 순서는 스레드 스케줄링에 따라 달라질 수 있습니다.
Starting Thread-1 Starting Thread-2 Starting Thread-3 Thread-1 processing One Thread-2 processing Two Thread-3 processing Three Thread-1 processing Four Thread-2 processing Five Exiting Thread-3 Exiting Thread-1 Exiting Thread-2 Exiting Main Thread
코드 동작 원리
1. 스레드 생성과 시작
myThread 클래스는 threading.Thread를 상속받아 정의됩니다. 각 스레드는 고유한 ID와 이름, 그리고 공유할 큐 객체를 인자로 받으며, start() 호출 시 run() 메서드가 실행됩니다.
2. 락(Lock)을 이용한 동기화
여러 스레드가 동시에 큐에서 데이터를 꺼내려 하면 경쟁 조건(race condition)이 발생할 수 있습니다. 이를 방지하기 위해 queueLock.acquire()로 락을 획득한 후 큐를 확인하고, 작업이 끝나면 반드시 release()로 락을 해제합니다.
3. 종료 플래그 처리
메인 스레드는 큐가 비워질 때까지 기다린 후 exitFlag를 1로 설정하여 워커 스레드들에게 종료를 알립니다. 마지막으로 join() 메서드를 통해 모든 스레드가 완전히 종료될 때까지 대기한 뒤 프로그램이 마무리됩니다.
참고: Python 2에서는 import Queue로 모듈을 불러왔지만, Python 3부터는 소문자인 import queue를 사용합니다. 또한 queue.Queue 자체가 내부적으로 스레드 안전(thread-safe)하게 구현되어 있으므로, 단순한 경우에는 별도의 락 없이도 안전하게 사용할 수 있습니다.