1. 다중 프로세스 개발
프로세스는 운영 체제에서 리소스 할당의 최소 단위이며, 각 프로세스는 독립적인 메모리 공간과 시스템 리소스를 가집니다. 하나의 프로세스 내에는 여러 스레드가 존재할 수 있으며, 이 스레드들은 프로세스의 리소스를 공유합니다. Python에서 다중 프로세스를 사용하면 CPU의 다중 코어 이점을 활용할 수 있어, 주로 계산 집약적인 작업에 적합합니다.
1.1. 프로세스 기본 사용법
Python의 multiprocessing 모듈을 사용하면 쉽게 새 프로세스를 생성하고 관리할 수 있습니다.
import multiprocessing
import time
def process_task(data_item):
"""자식 프로세스에서 실행될 함수."""
print(f"[{multiprocessing.current_process().name}] 작업 시작: {data_item}")
time.sleep(1) # 시뮬레이션
print(f"[{multiprocessing.current_process().name}] 작업 완료: {data_item}")
if __name__ == '__main__':
print(f"메인 프로세스 시작: {multiprocessing.current_process().name}")
# Process 객체 생성
# target: 새 프로세스에서 실행할 함수
# args: target 함수에 전달할 인자 튜플
proc1 = multiprocessing.Process(target=process_task, args=("데이터 A",))
proc2 = multiprocessing.Process(target=process_task, args=("데이터 B",))
# 프로세스 시작
proc1.start()
proc2.start()
# 프로세스가 완료될 때까지 대기
proc1.join()
proc2.join()
print("모든 프로세스 작업 완료.")
multiprocessing 모듈은 플랫폼에 따라 프로세스를 시작하는 세 가지 방식을 지원합니다. 이를 시작 방식(start methods)이라고 합니다:
fork:부모 프로세스가
os.fork()를 사용하여 Python 인터프리터를 포크합니다. 자식 프로세스는 부모 프로세스의 복사본처럼 시작되며, 부모의 거의 모든 리소스를 상속받습니다. 이는 Unix 계열 시스템에서 기본값이며, 빠르게 프로세스를 시작할 수 있습니다. 단, 멀티스레드 프로세스를 안전하게 포크하는 것은 어려울 수 있습니다.spawn:부모 프로세스는 완전히 새로운 Python 인터프리터 프로세스를 시작합니다. 자식 프로세스는
Process객체의run()메서드를 실행하는 데 필요한 리소스만 상속받습니다. 부모 프로세스의 불필요한 파일 디스크립터나 핸들은 상속되지 않습니다.fork방식에 비해 시작 속도가 느리지만, Unix 및 Windows 모두에서 사용 가능하며, Windows와 macOS의 기본값입니다.forkserver:프로그램이 시작될 때 서버 프로세스가 시작됩니다. 이후 새 프로세스가 필요할 때마다 부모 프로세스는 서버에 연결하여 새 프로세스를 포크하도록 요청합니다.
forkserver프로세스는 단일 스레드이므로os.fork()를 안전하게 사용할 수 있습니다. 불필요한 리소스는 상속되지 않습니다. 일부 Unix 플랫폼에서만 사용 가능합니다.
시작 방식은 multiprocessing.set_start_method() 함수를 사용하여 설정할 수 있습니다. 이 함수는 반드시 다른 프로세스를 생성하기 전에 호출해야 합니다.
import multiprocessing
# 'spawn' 방식으로 시작하도록 설정 (주로 Windows/macOS에서 명시적으로 사용)
multiprocessing.set_start_method("spawn", force=True)
프로세스 리소스 상속 예시
fork 방식은 부모 프로세스의 리소스를 복사하므로, 부모 프로세스에서 변경된 내용이 자식 프로세스에 영향을 미칠 수 있습니다. 반면 spawn 방식은 독립적인 환경을 제공합니다.
import multiprocessing
import time
import os
# --- 예시 1: 리스트 상속 (fork vs spawn) ---
def list_processor(data_list):
"""리스트에 데이터를 추가하는 작업"""
print(f"[자식 프로세스 {os.getpid()}] 시작. 현재 리스트: {data_list}")
data_list.append("자식 데이터")
print(f"[자식 프로세스 {os.getpid()}] 완료. 변경된 리스트: {data_list}")
if __name__ == '__main__':
# 'fork' 방식 시뮬레이션 (Unix 환경 가정)
# multiprocessing.set_start_method("fork", force=True) # Unix에서 테스트 시 주석 해제
print("--- fork 방식 시뮬레이션 ---")
shared_data_fork = [1, 2, 3]
print(f"부모 프로세스 초기 리스트 (fork): {shared_data_fork}")
p_fork = multiprocessing.Process(target=list_processor, args=(shared_data_fork,))
p_fork.start()
p_fork.join()
# fork 방식의 경우 자식 프로세스는 부모의 리스트를 복사하므로,
# 자식에서 변경된 내용은 부모 프로세스의 원본 리스트에 반영되지 않습니다.
print(f"부모 프로세스 최종 리스트 (fork): {shared_data_fork}")
# 결과: [1, 2, 3] (자식의 변경사항이 부모에 영향 없음)
print("\n--- spawn 방식 시뮬레이션 ---")
multiprocessing.set_start_method("spawn", force=True) # spawn 방식으로 강제 설정
shared_data_spawn = [10, 20, 30]
print(f"부모 프로세스 초기 리스트 (spawn): {shared_data_spawn}")
p_spawn = multiprocessing.Process(target=list_processor, args=(shared_data_spawn,))
p_spawn.start()
p_spawn.join()
# spawn 방식은 완전히 새로운 프로세스를 시작하므로, 자식은 부모의 리스트를 복사하지 않습니다.
# 인자로 전달된 리스트는 새로운 객체로 직렬화되어 전달됩니다.
print(f"부모 프로세스 최종 리스트 (spawn): {shared_data_spawn}")
# 결과: [10, 20, 30] (자식의 변경사항이 부모에 영향 없음)
# --- 예시 2: 파일 객체 상속 (fork만 가능, spawn/forkserver는 불가능) ---
# fork 방식은 파일 객체와 같은 OS 리소스도 상속합니다.
# spawn/forkserver는 불필요한 리소스 상속을 피합니다.
def file_writer_task(file_handle, content):
"""파일 객체를 통해 내용을 쓰는 작업"""
print(f"[자식 프로세스 {os.getpid()}] 파일에 '{content}' 작성 시도")
file_handle.write(content + "\n")
file_handle.flush() # 버퍼를 비워 즉시 파일에 쓰기
if __name__ == '__main__':
# x_output.txt 파일 초기화
with open('x_output.txt', 'w', encoding='utf-8') as f:
f.write("초기 내용\n")
# fork 방식 설정 (Unix 환경에서만 정상 작동)
# Windows에서는 spawn이 기본이며, set_start_method("fork")는 오류 발생.
try:
multiprocessing.set_start_method("fork", force=True)
print("\n--- 파일 객체 상속 (fork 방식) ---")
# 부모 프로세스에서 파일 열고 내용 작성
parent_file = open('x_output.txt', 'a+', encoding='utf-8')
parent_file.write("부모가 쓴 첫 번째 내용\n")
parent_file.flush() # 즉시 쓰기
# 자식 프로세스 생성 및 시작
p_file = multiprocessing.Process(target=file_writer_task, args=(parent_file, "자식이 쓴 내용"))
p_file.start()
p_file.join()
# 자식 프로세스가 완료된 후 부모가 추가로 작성
parent_file.write("부모가 쓴 두 번째 내용\n")
parent_file.flush()
parent_file.close()
print("파일 'x_output.txt' 내용 확인 필요.")
# 예상되는 x_output.txt 내용 (fork 방식):
# 초기 내용
# 부모가 쓴 첫 번째 내용
# 자식이 쓴 내용
# 부모가 쓴 두 번째 내용
except RuntimeError as e:
print(f"\n파일 객체 상속 테스트 실패: {e}")
print("fork 방식은 Windows에서 지원되지 않습니다. Unix/macOS에서 테스트하세요.")
1.2. 프로세스 주요 기능
multiprocessing.Process 객체는 프로세스를 제어하는 여러 메서드와 속성을 제공합니다.
p.start():프로세스를 시작하여 작업을 실행할 준비를 합니다. 실제 실행은 OS 스케줄러에 의해 결정됩니다.
p.join([timeout]):프로세스가 완료될 때까지 부모 프로세스를 블록(대기)합니다.
timeout을 지정하면 해당 시간 동안만 기다립니다.import time from multiprocessing import Process, set_start_method def worker_routine(task_id): print(f"[{task_id}] 작업 시작...") time.sleep(2) print(f"[{task_id}] 작업 완료!") if __name__ == '__main__': set_start_method("spawn", force=True) # macOS/Windows 기본값 print("메인 프로세스: 자식 프로세스 시작 전.") child_proc = Process(target=worker_routine, args=("ID-001",)) child_proc.start() print("메인 프로세스: 자식 프로세스 완료 대기 중...") child_proc.join() # 자식 프로세스가 끝날 때까지 기다립니다. print("메인 프로세스: 자식 프로세스 완료 후 계속 실행.")p.daemon = True/False:프로세스를 데몬 프로세스로 설정합니다.
start()메서드 호출 전에 설정해야 합니다.p.daemon = True: 자식 프로세스가 데몬이 됩니다. 부모 프로세스가 종료되면 자식 데몬 프로세스도 자동으로 종료됩니다. 따라서 부모가 자식의 완료를 기다리지 않습니다.p.daemon = False(기본값): 자식 프로세스가 비-데몬입니다. 부모 프로세스는 모든 비-데몬 자식 프로세스가 완료될 때까지 기다렸다가 종료됩니다.
import time from multiprocessing import Process, set_start_method def daemon_worker(name): print(f"[데몬 {name}] 작업 시작 (PID: {Process.pid}). 5초 슬립.") time.sleep(5) print(f"[데몬 {name}] 작업 완료.") if __name__ == '__main__': set_start_method("spawn", force=True) print("메인 프로세스: 데몬 프로세스 시작 시도.") d_proc = Process(target=daemon_worker, args=("Reporter",)) d_proc.daemon = True # 데몬 프로세스로 설정 d_proc.start() print("메인 프로세스: 2초 후 종료. 데몬은 자동으로 종료될 것임.") time.sleep(2) print("메인 프로세스: 종료.") # d_proc.join()을 호출하지 않았으므로 메인 프로세스는 2초 후 종료되고, # 데몬 프로세스는 강제로 종료됩니다. "데몬 작업 완료" 메시지는 출력되지 않을 수 있습니다.- 프로세스 이름 설정 및 가져오기:
프로세스의 이름을 설정하거나 현재 프로세스의 이름을 가져올 수 있습니다.
import os import time import multiprocessing import threading def sub_thread_task(): """자식 프로세스 내에서 실행될 스레드 작업""" time.sleep(1) # 스레드 작업 시뮬레이션 print(f"스레드 {threading.current_thread().name} 완료.") def sub_process_task(proc_name): """자식 프로세스에서 실행될 주 작업""" print(f"자식 프로세스 시작: PID={os.getpid()}, PPID={os.getppid()}") print(f" 프로세스 이름: {multiprocessing.current_process().name}") # 이 프로세스 내에서 여러 스레드 시작 for i in range(3): t = threading.Thread(target=sub_thread_task, name=f"SubThread-{i}") t.start() # 활성화된 스레드 수 확인 (메인 스레드 + 생성된 스레드) print(f" 현재 활성 스레드 수: {len(threading.enumerate())}") time.sleep(3) # 자식 프로세스 작업 시뮬레이션 print(f"자식 프로세스 {multiprocessing.current_process().name} 종료.") if __name__ == '__main__': multiprocessing.set_start_method("spawn", force=True) print(f"메인 프로세스 PID: {os.getpid()}") new_proc = multiprocessing.Process(target=sub_process_task, args=("MyWorker",)) new_proc.name = "CustomWorkerProcess" # 프로세스 이름 설정 new_proc.start() new_proc.join() # 자식 프로세스 완료 대기 print("메인 프로세스: 모든 자식 프로세스 완료.") - 사용자 정의 프로세스 클래스:
multiprocessing.Process를 상속받아 사용자 정의 프로세스 클래스를 만들 수 있습니다. 이 경우, 프로세스가 시작될 때run()메서드가 실행됩니다.import multiprocessing import time class CustomWorker(multiprocessing.Process): def __init__(self, task_data, *args, **kwargs): super().__init__(*args, **kwargs) self.task_data = task_data def run(self): """이 메서드는 프로세스가 시작될 때 자동으로 호출됩니다.""" print(f"[{self.name}] 커스텀 프로세스 실행: {self.task_data}") time.sleep(1.5) print(f"[{self.name}] 커스텀 프로세스 완료: {self.task_data}") if __name__ == '__main__': multiprocessing.set_start_method("spawn", force=True) print("메인 프로세스: 커스텀 프로세스 시작 전.") my_proc = CustomWorker(task_data="핵심 작업", name="CustomProcess-1") my_proc.start() my_proc.join() # 커스텀 프로세스 완료 대기 print("메인 프로세스: 모든 커스텀 프로세스 완료.") - CPU 코어 개수 확인:
multiprocessing.cpu_count()를 사용하여 시스템의 CPU 코어 개수를 알아낼 수 있습니다. 이를 바탕으로 최적의 프로세스 수를 결정할 수 있습니다.import multiprocessing if __name__ == '__main__': num_cpus = multiprocessing.cpu_count() print(f"시스템의 CPU 코어 개수: {num_cpus}") # 예를 들어, CPU 코어 수만큼 프로세스를 생성하여 작업을 분산 # for i in range(num_cpus): # p = multiprocessing.Process(target=some_cpu_bound_task, args=(i,)) # p.start()
2. 프로세스 간 데이터 공유
기본적으로 프로세스는 독립적인 메모리 공간을 가지므로 데이터를 직접 공유하지 않습니다. 자식 프로세스에 데이터를 전달하거나, 프로세스 간에 데이터를 공유하려면 특별한 메커니즘이 필요합니다.
import multiprocessing
def modify_list_in_child(some_list):
"""자식 프로세스에서 리스트를 수정"""
some_list.append(999)
print(f"[자식 프로세스] 리스트: {some_list}")
if __name__ == '__main__':
multiprocessing.set_start_method("spawn", force=True)
main_list = [10, 20]
print(f"[메인 프로세스] 초기 리스트: {main_list}")
child_p = multiprocessing.Process(target=modify_list_in_child, args=(main_list,))
child_p.start()
child_p.join()
print(f"[메인 프로세스] 최종 리스트: {main_list}") # 자식 프로세스의 변경사항이 반영되지 않음
위 예시에서 자식 프로세스는 main_list의 복사본을 받으므로, 원본 리스트는 변경되지 않습니다. 데이터를 공유하기 위한 방법은 다음과 같습니다.
2.1. 공유 메모리 (Shared Memory)
multiprocessing 모듈은 Value와 Array를 통해 공유 메모리에 데이터를 저장하고 조작할 수 있도록 지원합니다. 이는 ctypes를 기반으로 합니다.
Value(typecode, initial_value): 단일 값을 공유합니다.from multiprocessing import Process, Value import time def increment_value(shared_val): """공유된 Value 값을 증가시키는 작업""" print(f"[자식 프로세스] 시작. 현재 값: {shared_val.value}") time.sleep(0.5) shared_val.value += 10 print(f"[자식 프로세스] 완료. 변경된 값: {shared_val.value}") if __name__ == '__main__': multiprocessing.set_start_method("spawn", force=True) # 'i'는 signed int, 0은 초기값 shared_int = Value('i', 0) # 'd'는 double (float), 3.14는 초기값 shared_float = Value('d', 3.14) print(f"초기 공유 정수 값: {shared_int.value}") print(f"초기 공유 실수 값: {shared_float.value}") p1 = Process(target=increment_value, args=(shared_int,)) p2 = Process(target=increment_value, args=(shared_int,)) p1.start() p2.start() p1.join() p2.join() print(f"최종 공유 정수 값: {shared_int.value}") # 예상: 20 (경합 조건으로 인해 정확히 20이 아닐 수도 있음. 락 필요) print(f"최종 공유 실수 값: {shared_float.value}") # 변경 없음Array(typecode, size_or_list): 배열을 공유합니다. 요소의 타입은 모두 동일해야 합니다.from multiprocessing import Process, Array def update_array_element(shared_array, index, new_value): """공유된 Array의 특정 인덱스 값을 업데이트하는 작업""" print(f"[자식 프로세스] 시작. Array[{index}] 값: {shared_array[index]}") shared_array[index] = new_value print(f"[자식 프로세스] 완료. Array[{index}] 값: {shared_array[index]}") if __name__ == '__main__': multiprocessing.set_start_method("spawn", force=True) # 'i'는 signed int, [10, 20, 30, 40]은 초기값 리스트 shared_int_array = Array('i', [10, 20, 30, 40]) # 'c'는 char (바이트), 5는 배열의 크기 (ex: b'abc\0\0') shared_char_array = Array('c', b'Hello') print(f"초기 공유 정수 배열: {list(shared_int_array)}") print(f"초기 공유 문자 배열: {shared_char_array.value.decode('utf-8')}") pa = Process(target=update_array_element, args=(shared_int_array, 1, 200)) pb = Process(target=update_array_element, args=(shared_int_array, 3, 400)) pa.start() pb.start() pa.join() pb.join() print(f"최종 공유 정수 배열: {list(shared_int_array)}") print(f"최종 공유 문자 배열: {shared_char_array.value.decode('utf-8')}")
2.2. 매니저(Manager) 프로세스
Manager() 함수가 반환하는 매니저 객체는 파이썬 객체를 호스팅하는 서버 프로세스를 제어합니다. 다른 프로세스들은 프록시를 통해 이 서버 프로세스의 객체를 조작할 수 있습니다. 이를 통해 복잡한 파이썬 객체(리스트, 딕셔너리 등)를 프로세스 간에 공유할 수 있습니다.
from multiprocessing import Process, Manager
import time
def process_manager_data(managed_dict, managed_list):
"""매니저 객체를 통해 공유되는 데이터를 조작하는 작업"""
print(f"[자식 프로세스] 시작. 딕셔너리: {managed_dict}, 리스트: {managed_list}")
managed_dict["pid"] = Process.pid
managed_dict["status"] = "completed"
managed_list.append(f"자식 {Process.pid} 추가")
time.sleep(0.5)
print(f"[자식 프로세스] 완료.")
if __name__ == '__main__':
multiprocessing.set_start_method("spawn", force=True)
with Manager() as manager:
# 매니저를 통해 공유 가능한 딕셔너리와 리스트 생성
shared_dict = manager.dict()
shared_list = manager.list()
shared_dict["name"] = "공유 데이터"
shared_list.append("초기 항목")
print(f"[메인 프로세스] 초기 공유 딕셔너리: {shared_dict}")
print(f"[메인 프로세스] 초기 공유 리스트: {shared_list}")
p1 = Process(target=process_manager_data, args=(shared_dict, shared_list))
p2 = Process(target=process_manager_data, args=(shared_dict, shared_list))
p1.start()
p2.start()
p1.join()
p2.join()
print(f"[메인 프로세스] 최종 공유 딕셔너리: {shared_dict}")
print(f"[메인 프로세스] 최종 공유 리스트: {shared_list}")
2.3. 프로세스 간 통신 (IPC)
multiprocessing은 프로세스 간 통신을 위한 두 가지 유형의 채널을 지원합니다.
- 큐 (Queues):
Queue클래스는queue.Queue와 유사하며, 프로세스 간에 객체를 안전하게 전달하는 데 사용됩니다. FIFO (First-In, First-Out) 방식으로 데이터를 주고받을 수 있습니다.import multiprocessing import time def producer_task(data_queue, num_items): """큐에 데이터를 넣는 생산자 역할""" print(f"[생산자 프로세스] 시작.") for i in range(num_items): item = f"데이터-{i}" data_queue.put(item) print(f"[생산자 프로세스] 큐에 '{item}' 추가.") time.sleep(0.1) data_queue.put(None) # 작업 완료 신호 print(f"[생산자 프로세스] 완료.") def consumer_task(data_queue): """큐에서 데이터를 가져와 처리하는 소비자 역할""" print(f"[소비자 프로세스] 시작.") while True: item = data_queue.get() if item is None: # 작업 완료 신호 수신 break print(f"[소비자 프로세스] '{item}' 처리 중.") time.sleep(0.2) print(f"[소비자 프로세스] 완료.") if __name__ == '__main__': multiprocessing.set_start_method("spawn", force=True) data_queue = multiprocessing.Queue() producer_p = multiprocessing.Process(target=producer_task, args=(data_queue, 5)) consumer_p = multiprocessing.Process(target=consumer_task, args=(data_queue,)) producer_p.start() consumer_p.start() producer_p.join() consumer_p.join() print("모든 생산자/소비자 프로세스 완료.") - 파이프 (Pipes):
Pipe()함수는 기본적으로 양방향(duplex) 통신이 가능한 두 개의 연결 객체를 반환합니다.send()와recv()메서드를 사용하여 데이터를 주고받습니다.import multiprocessing import time def child_process_pipe(conn_end): """자식 프로세스에서 파이프를 통해 통신하는 작업""" print(f"[자식 프로세스] 시작. 부모의 메시지 대기.") received_msg = conn_end.recv() # 부모로부터 메시지 수신 (블록) print(f"[자식 프로세스] 부모로부터 메시지 수신: '{received_msg}'") response_msg = f"자식 {multiprocessing.current_process().pid} 응답: 잘 받았습니다!" conn_end.send(response_msg) # 부모에게 응답 메시지 전송 print(f"[자식 프로세스] 부모에게 응답 전송: '{response_msg}'") conn_end.close() # 연결 닫기 print(f"[자식 프로세스] 완료.") if __name__ == '__main__': multiprocessing.set_start_method("spawn", force=True) parent_conn, child_conn = multiprocessing.Pipe() # 파이프 생성 child_proc = multiprocessing.Process(target=child_process_pipe, args=(child_conn,)) child_proc.start() time.sleep(0.5) # 자식 프로세스 시작 대기 parent_msg = "안녕하세요, 자식 프로세스!" parent_conn.send(parent_msg) # 자식에게 메시지 전송 print(f"[메인 프로세스] 자식에게 메시지 전송: '{parent_msg}'") print(f"[메인 프로세스] 자식의 응답 대기.") child_response = parent_conn.recv() # 자식으로부터 응답 수신 (블록) print(f"[메인 프로세스] 자식으로부터 응답 수신: '{child_response}'") child_proc.join() parent_conn.close() # 연결 닫기 print("모든 프로세스 통신 완료.")
이러한 Python 내부의 메커니즘 외에도, 실제 프로젝트에서는 Redis, MySQL 등 외부 데이터베이스나 메시지 큐 시스템을 활용하여 프로세스 간 데이터를 공유하거나 교환하는 경우가 많습니다.
3. 프로세스 잠금 (Process Locks)
여러 프로세스가 동시에 공유 리소스(파일, 공유 메모리 등)를 조작할 때, 예측 불가능한 결과나 데이터 손상(경합 조건)이 발생할 수 있습니다. 이를 방지하기 위해 잠금(Lock)을 사용하여 한 번에 하나의 프로세스만 리소스에 접근하도록 제어할 수 있습니다.
문제 상황: 잠금 없이 공유 파일 조작
import time
import multiprocessing
import os
def update_counter_no_lock(file_name):
"""파일 내 카운터 값을 잠금 없이 업데이트"""
try:
# 파일에서 현재 카운터 읽기
with open(file_name, 'r', encoding='utf-8') as f:
current_count = int(f.read().strip())
print(f"[PID: {os.getpid()}] 현재 카운트: {current_count}")
time.sleep(0.1) # 경합 조건 시뮬레이션을 위해 잠시 대기
# 카운터 감소
new_count = current_count - 1
# 파일에 새 카운터 쓰기
with open(file_name, 'w', encoding='utf-8') as f:
f.write(str(new_count))
print(f"[PID: {os.getpid()}] 새 카운트: {new_count}")
except FileNotFoundError:
print(f"파일 {file_name}을 찾을 수 없습니다. 초기화하세요.")
except Exception as e:
print(f"[PID: {os.getpid()}] 오류 발생: {e}")
if __name__ == '__main__':
multiprocessing.set_start_method("spawn", force=True)
counter_file = 'counter.txt'
# 파일 초기화
with open(counter_file, 'w', encoding='utf-8') as f:
f.write('10') # 초기 카운터 값 10
processes = []
for _ in range(5): # 5개의 프로세스가 동시에 카운터 감소 시도
p = multiprocessing.Process(target=update_counter_no_lock, args=(counter_file,))
processes.append(p)
p.start()
for p in processes:
p.join()
with open(counter_file, 'r', encoding='utf-8') as f:
final_count = int(f.read().strip())
print(f"\n[메인 프로세스] 최종 카운트: {final_count}")
# 예상: 5, 실제: 8, 9 등 예측 불가능한 값 (경합 조건)
위 예시를 실행하면 최종 카운트가 예상치 못한 값(예: 8 또는 9)이 나올 수 있습니다. 이는 여러 프로세스가 동시에 파일을 읽고 쓰는 과정에서 데이터가 덮어씌워지기 때문입니다. 이 문제를 해결하기 위해 multiprocessing.Lock 또는 multiprocessing.RLock을 사용합니다.
해결책: 잠금을 사용하여 공유 파일 조작
import time
import multiprocessing
import os
def update_counter_with_lock(file_name, lock_obj):
"""파일 내 카운터 값을 잠금과 함께 업데이트"""
print(f"[PID: {os.getpid()}] 잠금 획득 대기...")
# 'with' 문을 사용하여 잠금을 자동으로 획득하고 해제
with lock_obj:
print(f"[PID: {os.getpid()}] 잠금 획득 성공.")
try:
with open(file_name, 'r', encoding='utf-8') as f:
current_count = int(f.read().strip())
print(f"[PID: {os.getpid()}] 현재 카운트: {current_count}")
time.sleep(0.1) # 경합 조건 시뮬레이션을 위해 잠시 대기
new_count = current_count - 1
with open(file_name, 'w', encoding='utf-8') as f:
f.write(str(new_count))
print(f"[PID: {os.getpid()}] 새 카운트: {new_count}")
except Exception as e:
print(f"[PID: {os.getpid()}] 오류 발생: {e}")
finally:
print(f"[PID: {os.getpid()}] 잠금 해제됨.")
if __name__ == '__main__':
multiprocessing.set_start_method("spawn", force=True)
counter_file = 'counter_locked.txt'
with open(counter_file, 'w', encoding='utf-8') as f:
f.write('10') # 초기 카운터 값 10
# 프로세스 잠금 객체 생성
process_lock = multiprocessing.RLock() # RLock은 같은 스레드에서 여러 번 획득 가능
processes = []
for _ in range(5):
p = multiprocessing.Process(target=update_counter_with_lock, args=(counter_file, process_lock))
processes.append(p)
p.start()
for p in processes:
p.join()
with open(counter_file, 'r', encoding='utf-8') as f:
final_count = int(f.read().strip())
print(f"\n[메인 프로세스] 최종 카운트 (잠금 사용): {final_count}")
# 예상: 5 (잠금 덕분에 정확한 결과)
잠금을 사용하면 각 프로세스가 공유 리소스에 접근하기 전에 잠금을 획득하고, 작업이 끝난 후 잠금을 해제하여 데이터 무결성을 보장할 수 있습니다.
4. 프로세스 풀 (Process Pool)
수많은 작업을 병렬로 처리해야 할 때, 매번 새 프로세스를 생성하고 관리하는 것은 오버헤드가 큽니다. 프로세스 풀(Process Pool)은 미리 정해진 수의 프로세스를 생성해 놓고, 이 프로세스들을 재사용하여 작업을 처리하는 방식입니다. 이는 concurrent.futures.ProcessPoolExecutor를 통해 구현됩니다.
import time
from concurrent.futures import ProcessPoolExecutor
import multiprocessing
def heavy_computation_task(value):
"""시간이 오래 걸리는 계산 작업 시뮬레이션"""
proc_name = multiprocessing.current_process().name
print(f"[{proc_name}] 작업 {value} 시작...")
time.sleep(2) # CPU-bound 작업 시뮬레이션
result = value * value
print(f"[{proc_name}] 작업 {value} 완료, 결과: {result}")
return result
if __name__ == '__main__':
multiprocessing.set_start_method("spawn", force=True)
# 최대 4개의 워커 프로세스를 사용하는 프로세스 풀 생성
with ProcessPoolExecutor(max_workers=4) as executor:
print("메인 프로세스: 작업 제출 중...")
# 10개의 작업을 풀에 제출
futures = [executor.submit(heavy_computation_task, i) for i in range(10)]
# 모든 작업이 완료될 때까지 기다림 (shutdown(True)과 유사)
for future in futures:
print(f"작업 결과 수신: {future.result()}")
# `with` 블록을 벗어나면 자동으로 shutdown(wait=True)이 호출됩니다.
print("메인 프로세스: 모든 작업 처리 및 풀 종료 완료.")
ProcessPoolExecutor의 주요 메서드:
executor.submit(fn, *args, **kwargs):실행할 함수
fn과 그 인자들을 풀에 제출합니다.Future객체를 즉시 반환하며, 이 객체를 통해 작업의 상태를 확인하거나 결과를 가져올 수 있습니다.executor.shutdown(wait=True):풀에 더 이상 새 작업을 제출할 수 없도록 합니다.
wait=True(기본값)이면 제출된 모든 작업이 완료될 때까지 기다렸다가 풀을 종료합니다.with문을 사용하면 이 호출은 자동으로 이루어집니다.future.add_done_callback(fn):Future객체가 완료될 때 호출될 콜백 함수를 등록합니다. 콜백 함수는Future객체를 인자로 받습니다.ProcessPoolExecutor의 콜백 함수는 보통 메인 프로세스에서 실행됩니다.import time from concurrent.futures import ProcessPoolExecutor import multiprocessing def processing_job(id_num): """데이터 처리 작업""" proc_name = multiprocessing.current_process().name print(f"[{proc_name}] Job {id_num} 시작.") time.sleep(1) # 작업 시뮬레이션 result = f"Job {id_num} 결과 (PID: {os.getpid()})" return result def completion_callback(future): """작업 완료 후 호출될 콜백 함수""" # 이 콜백은 메인 프로세스에서 실행됩니다! print(f"\n[콜백] 메인 프로세스 ({os.getpid()})에서 콜백 실행됨.") try: data_result = future.result() print(f"[콜백] 작업 완료. 결과: {data_result}") except Exception as exc: print(f"[콜백] 작업 중 예외 발생: {exc}") if __name__ == '__main__': multiprocessing.set_start_method("spawn", force=True) print(f"메인 프로세스 시작 (PID: {os.getpid()})") with ProcessPoolExecutor(max_workers=3) as executor: for i in range(5): task_future = executor.submit(processing_job, i) task_future.add_done_callback(completion_callback) print("\n메인 프로세스: 모든 작업 및 콜백 처리 완료.")
프로세스 풀 내에서 잠금 사용
프로세스 풀 내에서 공유 리소스에 대한 잠금을 사용하려면 multiprocessing.Manager()를 통해 생성된 잠금 객체를 사용해야 합니다. 일반 multiprocessing.RLock()은 풀 내에서 제대로 작동하지 않을 수 있습니다.
import time
import multiprocessing
from concurrent.futures import ProcessPoolExecutor
import os
def update_file_in_pool(file_path, manager_lock):
"""프로세스 풀 내에서 잠금을 사용하여 파일 업데이트"""
proc_name = multiprocessing.current_process().name
print(f"[{proc_name}] 잠금 획득 대기...")
with manager_lock: # 매니저를 통해 생성된 잠금 사용
print(f"[{proc_name}] 잠금 획득 성공. 파일 '{file_path}' 업데이트 시작.")
try:
with open(file_path, 'r+', encoding='utf-8') as f:
current_value = int(f.read().strip())
new_value = current_value + 1
f.seek(0)
f.truncate()
f.write(str(new_value))
print(f"[{proc_name}] 파일 업데이트 완료. 새 값: {new_value}")
except Exception as e:
print(f"[{proc_name}] 파일 업데이트 중 오류 발생: {e}")
time.sleep(0.5) # 작업 시뮬레이션
print(f"[{proc_name}] 잠금 해제됨.")
if __name__ == '__main__':
multiprocessing.set_start_method("spawn", force=True)
shared_file = 'shared_counter_pool.txt'
with open(shared_file, 'w', encoding='utf-8') as f:
f.write('0') # 초기 카운터 0
# Manager를 통해 공유 가능한 잠금 객체 생성
with multiprocessing.Manager() as manager:
pool_lock = manager.Lock() # 또는 manager.RLock()
with ProcessPoolExecutor(max_workers=3) as executor:
print("메인 프로세스: 풀에 작업 제출 중...")
futures = [executor.submit(update_file_in_pool, shared_file, pool_lock) for _ in range(10)]
for f in futures:
f.result() # 모든 작업 완료 대기
with open(shared_file, 'r', encoding='utf-8') as f:
final_count = int(f.read().strip())
print(f"\n[메인 프로세스] 최종 파일 카운트: {final_count}")
# 예상: 10 (잠금 덕분에 정확한 결과)
예시: 로그 파일 분석
여러 로그 파일에서 각 파일의 총 요청 수와 고유 IP 주소 수를 병렬로 계산하는 예시입니다.
import os
import time
from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Manager
# 예시 로그 파일 생성 (실제 환경에서는 이미 존재)
if not os.path.exists("log_samples"):
os.makedirs("log_samples")
with open("log_samples/access_2023-01-01.log", "w") as f:
f.write("192.168.1.1 - - [01/Jan/2023:10:00:00 +0000] \"GET /page1 HTTP/1.1\"\n")
f.write("192.168.1.2 - - [01/Jan/2023:10:00:05 +0000] \"GET /page2 HTTP/1.1\"\n")
f.write("192.168.1.1 - - [01/Jan/2023:10:00:10 +0000] \"GET /page1 HTTP/1.1\"\n")
with open("log_samples/access_2023-01-02.log", "w") as f:
f.write("192.168.1.3 - - [02/Jan/2023:11:00:00 +0000] \"GET /page3 HTTP/1.1\"\n")
f.write("192.168.1.1 - - [02/Jan/2023:11:00:05 +0000] \"GET /page4 HTTP/1.1\"\n")
f.write("192.168.1.3 - - [02/Jan/2023:11:00:10 +0000] \"GET /page3 HTTP/1.1\"\n")
def analyze_log_file(filename):
"""단일 로그 파일을 분석하여 총 요청 수와 고유 IP 수를 반환"""
file_path = os.path.join("log_samples", filename)
total_requests = 0
unique_ips = set()
try:
with open(file_path, 'r', encoding='utf-8') as f:
for line in f:
if not line.strip():
continue
parts = line.split(" - -", maxsplit=1)
if len(parts) > 0:
ip_address = parts[0].strip()
total_requests += 1
unique_ips.add(ip_address)
print(f"파일 '{filename}' 분석 완료. (요청: {total_requests}, 고유 IP: {len(unique_ips)})")
time.sleep(0.5) # 작업 시뮬레이션
return {"total_requests": total_requests, "unique_ips": len(unique_ips)}
except Exception as e:
print(f"파일 '{filename}' 분석 중 오류 발생: {e}")
return None
def main_log_analyzer():
multiprocessing.set_start_method("spawn", force=True)
log_directory = "log_samples"
if not os.path.exists(log_directory):
print(f"로그 디렉토리 '{log_directory}'를 찾을 수 없습니다.")
return
# Manager를 사용하여 결과를 저장할 공유 딕셔너리 생성
with Manager() as manager:
results_map = manager.dict() # {filename: {"total_requests": ..., "unique_ips": ...}}
with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
print("로그 파일 분석 시작...")
futures = []
for log_file in os.listdir(log_directory):
if log_file.endswith(".log"):
# 작업 제출 및 콜백 함수 등록 (결과를 results_map에 저장)
future = executor.submit(analyze_log_file, log_file)
# 콜백은 메인 프로세스에서 실행되며, future.result()를 통해 결과를 가져옴
def _callback(f, fname=log_file):
res = f.result()
if res:
results_map[fname] = res
future.add_done_callback(_callback)
futures.append(future)
# 모든 작업 완료 대기
for f in futures:
f.result()
print("\n--- 분석 결과 ---")
for filename, data in results_map.items():
print(f"파일: {filename}, 총 요청: {data['total_requests']}, 고유 IP: {data['unique_ips']}")
if __name__ == '__main_log_analyzer__': # 실제 실행 시 __name__ == '__main__'으로 변경
main_log_analyzer()
위 main_log_analyzer 함수는 if __name__ == '__main_log_analyzer__':로 되어 있어 직접 실행하려면 __main_log_analyzer__를 __main__으로 변경해야 합니다.
5. 코루틴 (Coroutines)
스레드와 프로세스가 운영 체제가 관리하는 실제 병렬/동시성 실행 단위라면, 코루틴(Coroutine)은 프로그래머가 코드 수준에서 구현하는 "마이크로스레드" 또는 "사용자 공간의 컨텍스트 전환" 기술입니다. 단일 스레드 내에서 여러 작업을 협력적으로 번갈아 실행하는 방식입니다.
예를 들어, 일반적인 함수 호출은 순차적으로 실행됩니다:
def process_step_one():
print("Step 1a")
print("Step 1b")
def process_step_two():
print("Step 2a")
print("Step 2b")
process_step_one()
process_step_two()
# 출력: Step 1a, Step 1b, Step 2a, Step 2b (순차적)
코루틴을 사용하면 함수 실행 중간에 제어권을 다른 코루틴으로 넘겨 "번갈아 가며" 실행하는 것이 가능합니다. 예를 들어, Step 1a -> Step 2a -> Step 1b -> Step 2b와 같은 실행 흐름을 만들 수 있습니다.
Python에서 코루틴을 구현하는 몇 가지 방법이 있습니다:
5.1. Greenlet (서드파티 라이브러리)
greenlet은 경량 코루틴을 위한 라이브러리입니다. switch() 메서드를 사용하여 수동으로 제어권을 전환합니다.
# pip install greenlet
from greenlet import greenlet
def routine_alpha():
print("알파 - 1단계")
yield_g.switch() # 제어권을 yield_g로 전환
print("알파 - 2단계")
yield_g.switch() # 다시 제어권을 yield_g로 전환
def routine_beta():
print("베타 - 1단계")
main_g.switch() # 제어권을 main_g (routine_alpha)로 전환
print("베타 - 2단계")
if __name__ == '__main__':
main_g = greenlet(routine_alpha)
yield_g = greenlet(routine_beta)
main_g.switch() # routine_alpha 실행 시작
# 예상 출력: 알파-1단계 -> 베타-1단계 -> 알파-2단계 -> 베타-2단계
5.2. Yield (파이썬 내장)
제너레이터의 yield 키워드도 코루틴과 유사하게 제어권을 넘길 수 있습니다. 특히 Python 3.3부터 도입된 yield from은 제너레이터를 위임하는 기능을 제공하여 코루틴 체이닝에 활용될 수 있습니다.
def sub_generator():
yield "서브젠 - A"
yield "서브젠 - B"
def main_coroutine():
yield "메인 - 시작"
yield from sub_generator() # sub_generator로 제어권 위임
yield "메인 - 끝"
if __name__ == '__main__':
coro_executor = main_coroutine()
for item in coro_executor:
print(item)
# 예상 출력: 메인-시작 -> 서브젠-A -> 서브젠-B -> 메인-끝
5.3. Asyncio (async/await)
Python 3.4부터 asyncio 모듈이 도입되었고, Python 3.5부터 async와 await 구문이 추가되면서 비동기 프로그래밍을 위한 표준 코루틴 지원이 강화되었습니다. 이 방식은 개발자가 수동으로 제어권을 전환하는 대신, 주로 I/O 작업(네트워크 요청, 파일 읽기/쓰기 등)이 발생할 때 자동으로 다른 코루틴으로 전환되도록 설계되었습니다. 이는 단일 스레드 내에서 높은 동시성을 달성하는 데 매우 효과적입니다.
import asyncio
import time
async def async_task_one():
print("[Task One] 시작. 2초 대기.")
await asyncio.sleep(2) # I/O 작업(대기) 시뮬레이션. 이때 제어권이 다른 코루틴으로 넘어갈 수 있음.
print("[Task One] 완료.")
async def async_task_two():
print("[Task Two] 시작. 1초 대기.")
await asyncio.sleep(1) # I/O 작업(대기) 시뮬레이션
print("[Task Two] 완료.")
async def main_async_program():
start_time = time.monotonic()
# 두 개의 비동기 작업을 동시에 실행하도록 예약
await asyncio.gather(
async_task_one(),
async_task_two()
)
end_time = time.monotonic()
print(f"모든 비동기 작업 완료. 총 소요 시간: {end_time - start_time:.2f}초")
if __name__ == '__main__':
asyncio.run(main_async_program())
# 예상 출력:
# [Task One] 시작. 2초 대기.
# [Task Two] 시작. 1초 대기.
# (약 1초 후)
# [Task Two] 완료.
# (약 1초 후, 즉 총 2초 후)
# [Task One] 완료.
# 모든 비동기 작업 완료. 총 소요 시간: 2.xx초
위 예시에서 asyncio.sleep()은 실제 대기가 아니라 I/O 작업처럼 동작하여, 대기하는 동안 다른 코루틴으로 제어권을 넘깁니다. 덕분에 두 작업이 총 3초가 아닌 약 2초 만에 완료됩니다.
비동기 HTTP 요청 예시
aiohttp와 같은 비동기 HTTP 클라이언트를 사용하면 여러 웹 요청을 단일 스레드에서 효율적으로 병렬 처리할 수 있습니다.
# pip install aiohttp
import aiohttp
import asyncio
import os
async def fetch_image(session, url, target_dir="downloaded_images"):
"""비동기로 이미지를 다운로드하고 저장"""
if not os.path.exists(target_dir):
os.makedirs(target_dir)
filename = url.split('/')[-1]
filepath = os.path.join(target_dir, filename)
print(f"다운로드 시작: {filename}")
try:
async with session.get(url, ssl=False) as response: # ssl=False는 테스트용, 실제 환경에서는 피해야 함
if response.status == 200:
content = await response.read()
with open(filepath, 'wb') as f:
f.write(content)
print(f"다운로드 완료: {filename}")
else:
print(f"다운로드 실패: {filename}, 상태 코드: {response.status}")
except aiohttp.ClientError as e:
print(f"다운로드 중 클라이언트 오류 발생: {filename} - {e}")
except Exception as e:
print(f"다운로드 중 알 수 없는 오류 발생: {filename} - {e}")
async def main_image_downloader():
image_urls = [
'https://picsum.photos/id/237/200/300', # 랜덤 이미지
'https://picsum.photos/id/238/200/300',
'https://picsum.photos/id/239/200/300'
]
async with aiohttp.ClientSession() as session:
# 각 URL에 대해 비동기 작업 생성
download_tasks = [fetch_image(session, url) for url in image_urls]
# 모든 작업이 완료될 때까지 기다림
await asyncio.gather(*download_tasks)
if __name__ == '__main__':
start_time = time.monotonic()
asyncio.run(main_image_downloader())
end_time = time.monotonic()
print(f"\n모든 이미지 다운로드 완료. 총 소요 시간: {end_time - start_time:.2f}초")
위 예시에서 세 개의 이미지를 거의 동시에 다운로드하여 총 소요 시간을 크게 줄일 수 있습니다.
코루틴, 스레드, 프로세스의 차이점
- 프로세스 (Process):
운영 체제가 리소스를 할당하는 최소 단위입니다. 각 프로세스는 독립적인 메모리 공간을 가지며 서로 격리됩니다. Python의 GIL(Global Interpreter Lock) 때문에 CPU 집약적 작업에 가장 적합합니다. 다중 코어를 완벽하게 활용할 수 있습니다.
- 스레드 (Thread):
CPU가 스케줄링하는 최소 실행 단위입니다. 동일 프로세스 내의 스레드들은 메모리와 리소스를 공유합니다. Python CPython 구현의 GIL로 인해 동시에 하나의 스레드만 Python 바이트코드를 실행할 수 있으므로, CPU 집약적 작업에는 적합하지 않습니다. 하지만 I/O 집약적 작업(네트워크 요청, 디스크 I/O)에서는 GIL이 해제되어 동시성 이점을 얻을 수 있습니다.
- 코루틴 (Coroutine):
사용자(프로그래머) 수준에서 구현되는 경량 동시성 단위입니다. 단일 스레드 내에서 협력적으로 작업을 전환하며 실행됩니다. 주로 I/O 작업 시 자동으로 컨텍스트를 전환하여 대기 시간을 활용함으로써 높은 동시성을 달성합니다. 스레드보다 컨텍스트 전환 오버헤드가 훨씬 적어 I/O 집약적 작업에 매우 효율적입니다. 개발 난이도가 스레드나 프로세스보다 높을 수 있습니다.
요약하자면, CPU 집약적 작업에는 프로세스를, I/O 집약적 작업에는 코루틴(또는 스레드)을 사용하는 것이 일반적입니다. 특히 asyncio 기반의 코루틴은 스레드보다 더 적은 오버헤드로 I/O 동시성을 제공하여 최신 웹 프레임워크(FastAPI, Starlette 등)나 웹 크롤링에서 널리 활용되고 있습니다.