파이썬 스레드 간 통신 마스터하기: 동기화와 데이터 공유

스레드 간 통신(Inter-Thread Communication)은 파이썬 멀티스레드 프로그램에서 스레드 간 통신과 동기화를 가능하게 하는 프로세스를 말합니다.

일반적으로 파이썬의 스레드는 프로세스 내 동일한 메모리 공간을 공유하므로, 공유 변수, 객체, `threading` 모듈이 제공하는 특수 동기화 메커니즘을 통해 데이터를 교환하고 활동을 조정할 수 있습니다.

스레드 간 통신을 위해 `threading` 모듈은 Lock, Event, Condition, Semaphore 같은 다양한 동기화 기본 요소를 제공합니다. 이 튜토리얼에서는 Event와 Condition 객체를 사용해 멀티스레드 프로그램에서 스레드 간 통신을 제공하는 방법을 배웁니다.

Event 객체

Event 객체는 내부 플래그 상태를 관리해 스레드가 대기하거나 설정할 수 있게 합니다. Event 객체는 이 플래그 상태를 제어하는 메서드를 제공해 스레드가 공유 조건에 기반해 활동을 동기화할 수 있게 합니다.

플래그는 처음에 false이고 `set()`으로 true, `clear()`로 false가 됩니다. `wait()` 메서드는 플래그가 true가 될 때까지 블록합니다.

Event 객체 주요 메서드

• is_set() — 내부 플래그가 true면 True 반환
• set() — 내부 플래그를 true로 설정. true가 되길 기다리던 모든 스레드 깨움. 플래그 true 후 `wait()` 호출 시 블록하지 않음
• clear() — 내부 플래그를 false로 리셋. 이후 `wait()` 호출 시 `set()` 호출까지 블록
• wait(timeout=None) — 내부 플래그가 true가 될 때까지 블록. 진입 시 이미 true면 즉시 반환. 아니면 다른 스레드가 `set()` 호출하거나 선택적 타임아웃 발생까지 블록. timeout은 초 단위 부동소수점 수

예제: 교통 신호 시뮬레이션

교통 경찰이 GREEN/RED 신호로 교통 흐름을 제어하는 상황 시뮬레이션. 두 스레드 사용:

• `signal_state()`: 주기적으로 이벤트 설정/해제로 신호 변경
• `traffic_flow()`: 이벤트 설정 대기, 설정된 동안 루프 실행

from threading import Event, Thread
import time

terminate = False

def signal_state():
    global terminate
    while not terminate:
        time.sleep(0.5)
        print("교통 경찰: GREEN 신호")
        event.set()
        time.sleep(1)
        print("교통 경찰: RED 신호")
        event.clear()

def traffic_flow():
    global terminate
    num = 0
    while num < 10 and not terminate:
        print("GREEN 신호 대기 중")
        event.wait()
        print("GREEN 신호... 교통 이동 가능")
        while event.is_set() and not terminate:
            num += 1
            print(f"차량 번호: {num} 신호 통과")
            time.sleep(1)
        print("RED 신호... 교통 대기")

event = Event()
t1 = Thread(target=signal_state)
t2 = Thread(target=traffic_flow)
t1.start()
t2.start()

# 일정 시간 후 스레드 종료
time.sleep(5)
terminate = True
t1.join()
t2.join()
print("메인 스레드 종료")

출력 예시

GREEN 신호 대기 중
교통 경찰: GREEN 신호
GREEN 신호... 교통 이동 가능
차량 번호: 1 신호 통과
교통 경찰: RED 신호
RED 신호... 교통 대기
GREEN 신호 대기 중
교통 경찰: GREEN 신호
GREEN 신호... 교통 이동 가능
차량 번호: 2 신호 통과
차량 번호: 3 신호 통과
교통 경찰: RED 신호
교통 경찰: GREEN 신호
차량 번호: 4 신호 통과
교통 경찰: RED 신호
RED 신호... 교통 대기
교통 경찰: GREEN 신호
교통 경찰: RED 신호
메인 스레드 종료

Condition 객체

파이썬 `threading` 모듈의 Condition 객체는 더 고급 동기화 메커니즘을 제공합니다. 다른 스레드의 알림을 기다렸다가 진행할 수 있게 합니다. Condition 객체는 항상 락과 연관되어 있으며 스레드 간 신호 메커니즘을 제공합니다.

문법

threading.Condition(lock=None)

Condition 객체 주요 메서드

• acquire(*args) — 내부 락 획득
• release() — 내부 락 해제
• wait(timeout=None) — 내부 락 해제 후 다른 스레드의 `notify()`/`notify_all()` 호출이나 타임아웃까지 블록. 깨어나면 락 재획득 후 반환
• wait_for(predicate, timeout=None) — predicate 만족 또는 타임아웃까지 `wait()` 반복 호출. 반환값은 predicate 마지막 반환값, 타임아웃 시 False
• notify(n=1) — 조건 변수 대기 중인 최대 n개 스레드 깨움. 대기 스레드 없으면 no-op
• notify_all() — 이 조건에서 대기 중인 모든 스레드 깨움. 호출 스레드가 락 미보유 시 RuntimeError

예제 1: 기본 Condition 통신

스레드 A가 스레드 B의 알림을 기다리는 간단한 통신:

from threading import Condition, Thread
import time

c = Condition()

def thread_a():
    print("스레드 A 시작")
    with c:
        print("스레드 A 권한 대기 중...")
        c.wait()
        print("스레드 A 권한 획득!")
        print("스레드 A 종료")

def thread_b():
    print("스레드 B 시작")
    with c:
        time.sleep(2)
        print("스레드 A에 알림 전송...")
        c.notify()
        print("스레드 B 종료")

Thread(target=thread_a).start()
Thread(target=thread_b).start()

출력:

스레드 A 시작
스레드 A 권한 대기 중...
스레드 B 시작
스레드 A에 알림 전송...
스레드 B 종료
스레드 A 권한 획득!
스레드 A 종료

예제 2: 생산자-소비자 패턴

스레드 t2가 `taskB()`(소비자), t1이 `taskA()`(생산자) 실행. t1이 조건 획득 후 알림. t2는 대기 중이었다가 알림 받으면 생산된 난수 소비.

from threading import Condition, Thread
import time
import random

numbers = []

def taskA(c):
    for _ in range(5):
        with c:
            num = random.randint(1, 10)
            print(f"생성된 난수: {num}")
            numbers.append(num)
            print("알림 발송")
            c.notify()
            time.sleep(0.3)

def taskB(c):
    for i in range(5):
        with c:
            print("업데이트 대기 중")
            while not numbers:
                c.wait()
            print(f"획득한 난수: {numbers.pop()}")
            time.sleep(0.3)

c = Condition()
t1 = Thread(target=taskB, args=(c,))
t2 = Thread(target=taskA, args=(c,))
t1.start()
t2.start()
t1.join()
t2.join()
print("완료")

출력:

업데이트 대기 중
생성된 난수: 2
알림 발송
획득한 난수 2
생성된 난수: 5
알림 발송
업데이트 대기 중
획득한 난수 5
생성된 난수: 1
알림 발송
업데이트 대기 중
획득한 난수 1
생성된 난수: 9
알림 발송
업데이트 대기 중
획득한 난수 9
생성된 난수: 2
알림 발송
업데이트 대기 중
획득한 난수 2
완료