[PYTHON] multiprocessing.Queue와 queue.Queue의 결정적 차이 5가지와 실무 해결 방법

multiprocessing.Queue와 queue.Queue의 결정적 차이

 

 

파이썬으로 고성능 애플리케이션을 설계할 때 개발자가 가장 먼저 마주하는 난관은 "데이터를 어떻게 안전하게 주고받을 것인가?"입니다. 특히 멀티스레딩(Multithreading)과 멀티프로세싱(Multiprocessing) 환경에서 큐(Queue)는 필수적인 자료구조입니다. 하지만 이름이 비슷하다고 해서 queue.Queue와 multiprocessing.Queue를 혼용하면 치명적인 성능 저하나 런타임 에러를 초래할 수 있습니다. 본 포스팅에서는 두 모듈의 아키텍처 수준의 차이점을 분석하고, 실제 운영 환경에서 발생할 수 있는 교착 상태(Deadlock) 해결 방법과 최적화된 활용 사례를 전문 엔지니어의 시각에서 심도 있게 다룹니다.

 

1. 개념적 정의 및 아키텍처 구조

기본적으로 queue.Queue는 동일 프로세스 내의 스레드 간 통신을 위해 설계되었습니다. 반면, multiprocessing.Queue는 서로 다른 프로세스 간 통신(IPC, Inter-Process Communication)을 위해 설계되었습니다. 이 근본적인 차이는 메모리 공유 방식과 직렬화(Serialization) 유무에서 갈립니다.

주요 비교 요약표

비교 항목	queue.Queue	multiprocessing.Queue
주요 목적	멀티스레드 간 데이터 공유	멀티프로세스 간 데이터 공유
메모리 영역	공유 메모리 (동일 주소 공간)	독립 메모리 (프로세스별 격리)
데이터 전달 방식	객체 참조 전달 (In-memory)	Pickle 직렬화 후 파이프(Pipe) 전달
Global Interpreter Lock (GIL)	영향을 받음	영향을 받지 않음 (병렬 처리 가능)
성능 특성	오버헤드가 매우 적음	직렬화/역직렬화 비용 발생
안전성	Thread-Safe	Process-Safe

 

2. 실무에서 직면하는 3가지 핵심 차이점

2.1 직렬화(Serialization)의 제약

multiprocessing.Queue에 데이터를 넣을 때, 파이썬은 내부적으로 pickle 모듈을 사용하여 객체를 바이트 스트림으로 변환합니다. 따라서 직렬화가 불가능한 객체(예: 로컬 함수, 일부 DB 커넥션, 오픈된 파일 객체)는 multiprocessing 큐를 통해 전달할 수 없습니다. 반면 queue.Queue는 메모리 주소를 공유하므로 모든 파이썬 객체를 그대로 전달할 수 있습니다.

2.2 종료 메커니즘과 Deadlock 위험

multiprocessing.Queue는 데이터를 전달하기 위해 내부 버퍼와 피더 스레드(Feeder Thread)를 사용합니다. 큐에 데이터가 가득 찬 상태에서 프로세스를 강제 종료하거나, join() 메소드를 잘못 호출하면 버퍼의 데이터가 비워지지 않아 부모 프로세스가 영원히 대기하는 데드락(Deadlock) 현상이 발생할 수 있습니다.

2.3 처리량(Throughput)과 지연 시간

스레드 기반 큐는 단순히 락(Lock)을 획득하고 포인터를 변경하는 수준이지만, 프로세스 기반 큐는 시스템 콜을 통한 IPC 비용이 추가됩니다. 대량의 작은 데이터를 빈번하게 주고받아야 한다면 multiprocessing.Queue보다는 공유 메모리(Value, Array)나 Manager를 검토해야 합니다.

 

3. 실무 적용을 위한 전문 코드 예제 (Example 7선)

개발자가 현업에서 즉시 활용할 수 있는 패턴 위주로 구성하였습니다.

Ex 1. 표준 멀티스레드 생산자-소비자 패턴 (queue.Queue)

import threading
import queue
import time

def worker(q):
    while True:
        item = q.get()
        if item is None: break
        print(f"Processing {item}")
        q.task_done()

q = queue.Queue()
threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(q,))
    t.start()
    threads.append(t)

for item in range(10):
    q.put(item)

q.join() # 모든 작업 완료 대기
for i in range(3):
    q.put(None) # 종료 신호
for t in threads:
    t.join()

Ex 2. 멀티프로세싱을 활용한 CPU 집약적 작업 처리 (multiprocessing.Queue)

import multiprocessing as mp

def cpu_bound_task(q_in, q_out):
    while True:
        data = q_in.get()
        if data == "DONE": break
        # 무거운 연산 시뮬레이션
        result = sum(i * i for i in range(data))
        q_out.put(result)

if __name__ == "__main__":
    request_q = mp.Queue()
    result_q = mp.Queue()
    p = mp.Process(target=cpu_bound_task, args=(request_q, result_q))
    p.start()

    request_q.put(10**6)
    print(f"Result: {result_q.get()}")
    
    request_q.put("DONE")
    p.join()

Ex 3. Poison Pill 패턴을 이용한 안전한 프로세스 종료 방법

# 여러 개의 워커 프로세스를 안전하게 종료하는 방법
import multiprocessing as mp

def robust_worker(q):
    while True:
        task = q.get()
        if task is None: # Poison Pill
            print("Worker shutting down")
            break
        # 작업 수행
        pass

if __name__ == "__main__":
    q = mp.Queue()
    workers = [mp.Process(target=robust_worker, args=(q,)) for _ in range(4)]
    for w in workers: w.start()
    
    # 작업 투입 후 워커 수만큼 None 투입
    for _ in range(4):
        q.put(None)
    for w in workers: w.join()

Ex 4. Timeout 설정을 통한 블로킹 방지 및 예외 처리

import queue
import multiprocessing as mp

def try_get_data(q):
    try:
        # 3초 동안 데이터가 없으면 Empty 예외 발생
        data = q.get(block=True, timeout=3)
        return data
    except (queue.Empty, mp.queues.Empty):
        print("Queue is empty, moving on...")
        return None

Ex 5. JoinableQueue를 이용한 작업 동기화 해결

# task_done()과 join()을 사용하여 프로세스 간 작업 완료를 보장
from multiprocessing import JoinableQueue, Process

def consumer(q):
    while True:
        task = q.get()
        if task is None: break
        print(f"Finished {task}")
        q.task_done() # 중요: 작업 완료 알림

if __name__ == "__main__":
    jq = JoinableQueue()
    p = Process(target=consumer, args=(jq,))
    p.daemon = True
    p.start()

    jq.put("Job A")
    jq.join() # 모든 task_done()이 호출될 때까지 대기
    print("All jobs are completed.")

Ex 6. 대규모 데이터 전송 시 직렬화 오류 해결 (Custom Object)

import multiprocessing as mp

class HeavyData:
    def __init__(self, data):
        self.data = data
    # pickle 가능하도록 설계되어야 함

def process_heavy(q):
    obj = q.get()
    print(f"Received data length: {len(obj.data)}")

if __name__ == "__main__":
    q = mp.Queue()
    data_obj = HeavyData([i for i in range(100000)])
    p = mp.Process(target=process_heavy, args=(q,))
    p.start()
    q.put(data_obj)
    p.join()

Ex 7. 하이브리드 구조: Thread Pool 내에서의 Queue 활용

from concurrent.futures import ThreadPoolExecutor
import queue

# 실무 I/O 바운드 작업 최적화 모델
def fetch_url(q):
    while not q.empty():
        url = q.get()
        # requests.get(url) 등의 작업 수행
        print(f"Fetching {url}")

url_queue = queue.Queue()
for u in ["url1", "url2", "url3"]: url_queue.put(u)

with ThreadPoolExecutor(max_workers=3) as executor:
    executor.submit(fetch_url, url_queue)

 

4. 결론: 어떤 상황에 무엇을 써야 하는가?

선택의 기준은 명확합니다. 프로그램의 병목 지점이 어디냐에 따라 결정하십시오.

1. I/O 바운드 작업(네트워크 요청, 파일 읽기/쓰기): queue.Queue와 threading 모듈을 사용하십시오. 메모리 소모가 적고 컨텍스트 스위칭 비용이 낮습니다.
2. CPU 집약적 작업(이미지 처리, 대량 연산, 데이터 분석): multiprocessing.Queue와 multiprocessing 모듈을 사용하십시오. GIL을 우회하여 멀티 코어를 온전히 활용할 수 있습니다.
3. 분산 시스템 또는 복잡한 상태 공유: 단순 큐를 넘어 multiprocessing.Manager나 Redis 같은 외부 브로커 도입을 고려해야 합니다.

전문가의 팁: multiprocessing.Queue를 사용할 때는 반드시 put 한 데이터가 모두 소비된 후에 process.join()을 호출하거나, queue.cancel_join_thread()를 적절히 활용하여 좀비 프로세스 생성을 방지하십시오.

 

출처 및 참고 문헌

• Python Software Foundation. "multiprocessing — Process-based parallelism." Python 3.12 Documentation.
• Python Software Foundation. "queue — A synchronized queue class." Python 3.12 Documentation.
• Luciano Ramalho. "Fluent Python: Clear, Concise, and Effective Programming." O'Reilly Media.
• Beazley, D. "Python Cookbook: Recipes for Mastering Python 3." O'Reilly Media.