threading.Event() 객체의 활용 예제 및 백그라운드 실행 정리

1. threading.Event() 객체의 사용

 

🎯  threading.Event()란?

threading.Event()는 스레드 간의 동기화를 도와주는 객체 
스레드가 특정 이벤트(신호)를 기다리도록 만들고, 다른 스레드가 신호를 보내서 진행을 제어할 수 있음.

---

🎯 threading.Event()의 주요 기능

1. event.set() → 이벤트 신호를 발생(켜기)
2. event.clear() → 이벤트 신호를 리셋(끄기)
3. event.wait() → 이벤트가 발생할 때까지 기다림

🎯 threading.Event()를 사용할 때의 장점

✅ 스레드 간 동기화
✅ 불필요한 CPU 사용 방지 (while 루프 대신 wait() 사용)
✅ 메인 프로세스가 특정 작업이 끝날 때까지 기다릴 수 있음

 

 

하나의 파일에서 여러 스레드를 컨트롤한다면 아래 예제로도 무리가 없겠지만, threading.Event()객체를 여러 파일에서 글로벌하게 공유해야한다고 할때는 global 방식으로 해야할까, arg로 객체를 넘겨줘야할까

 

 

import pygame
import threading
import time

# 이벤트 객체 생성
tuning_done = threading.Event()

def play_tuning_sound():
    """튜닝 소리를 재생하는 스레드"""
    pygame.mixer.init()
    pygame.mixer.music.load("tuning_sound.mp3")
    pygame.mixer.music.play()
    
    print("🔧 튜닝 사운드 재생 중...")
    time.sleep(5)  # 튜닝 사운드가 5초 동안 재생된다고 가정
    
    pygame.mixer.music.stop()
    print("✅ 튜닝 완료!")
    tuning_done.set()  # 이벤트 신호 발생

def start_main_audio():
    """튜닝 완료 후, 메인 오디오를 재생하는 스레드"""
    print("🚀 메인 오디오 대기 중...")
    tuning_done.wait()  # 튜닝 완료될 때까지 대기

    pygame.mixer.music.load("main_audio.mp3")
    pygame.mixer.music.play()
    print("🎵 메인 오디오 재생 시작!")

# 스레드 실행
tuning_thread = threading.Thread(target=play_tuning_sound)
main_audio_thread = threading.Thread(target=start_main_audio)

tuning_thread.start()
main_audio_thread.start()

tuning_thread.join()
main_audio_thread.join()

 

 

✅ 결론: global을 피하고 인자로 전달하는 것이 더 좋은 이유

❌ global 사용✅ threading.Event() 인자로 전달

audio_player.py가 main.py의 변수를 직접 참조 → 모듈 간 의존성 증가	main.py에서 직접 tuning_done을 전달 → 독립성 유지
순환 참조 위험 (from main import tuning_done 하면 main.py가 먼저 실행되지 않으면 오류 발생)	순환 참조 문제 없음
global 변수 변경이 어디서든 일어날 수 있어 버그 발생 가능성 증가	tuning_done이 명확하게 어디서 전달되는지 알 수 있어 코드 유지보수 용이

 

 

main.py에서 변수 선언

import threading
import time
from audio.audio_player import play_tuning_audio

# ✅ `tuning_done` 선언 (전역 변수 아님)
tuning_done = threading.Event()

def main():
    """메인 실행 함수"""
    print("🚀 튜닝 사운드 재생 시작!")

    # ✅ `tuning_done`을 인자로 전달
    tuning_thread = threading.Thread(target=play_tuning_audio, args=(tuning_done,))
    tuning_thread.start()

    # ✅ 5초 후 튜닝 완료
    time.sleep(5)
    print("✅ 튜닝 완료! 오디오 중단")
    tuning_done.set()  # ✅ 이벤트 발생 → 튜닝 종료 신호

    tuning_thread.join()
    print("🎵 모든 작업 완료!")

if __name__ == "__main__":
    main()

 

 

audio/audio_player.py (인자로 전달)

import pygame
import time
import os

def get_random_tuning_audio():
    return "tuning_audio.mp3"

def play_tuning_audio(tuning_done):
    """튜닝 사운드를 계속 재생하는 스레드"""
    pygame.mixer.init()
    count = 0
    tuning_audio = get_random_tuning_audio()

    while not tuning_done.is_set():  # ✅ 튜닝 완료될 때까지 반복
        count += 1
        if count % 3 == 0:
            tuning_audio = get_random_tuning_audio()

        if tuning_audio and os.path.exists(tuning_audio):
            sound = pygame.mixer.Sound(tuning_audio)
            sound.play(-1)
            print(f"📡 튜닝 사운드 재생 중: {tuning_audio}")

        time.sleep(1)

    pygame.mixer.stop()
    print("🎵 튜닝 사운드 중단됨")

모듈 간의 의존성을 줄이고, 유지보수가 쉬운 구조로 유지하자..

 

 

---

🔥 Q1:  event.clear()를 하면 tuning_done 객체는 다시 사용할 수 없는 건가?

 event.clear()는 메모리를 해제하는 게 아니라 이벤트 상태를 "OFF"로 변경하는 역할을 함.
즉, 다시 set()을 호출하면 언제든지 재사용 가능!

 

✅ 이벤트는 ON/OFF 스위치처럼 동작

• event.set() → ON (True) 상태 → wait()을 호출한 스레드가 진행됨.
• event.clear() → OFF (False) 상태 → wait()을 호출한 스레드가 다시 대기 상태로 변함.

 

🔥 Q2: event.set() 상태 검사 vs event.wait() 후 코드 실행 차이점?

🔹 비교event.is_set()   vs event.wait()

💡 의미	현재 상태를 즉시 검사 (True/False)	이벤트가 set() 될 때까지 대기
⏳ 블로킹 여부	❌ 비동기 (즉시 실행 가능)	✅ 블로킹 (이벤트가 발생할 때까지 멈춤)
🎯 언제 사용?	이벤트 상태를 수시로 확인해야 할 때	특정 이벤트가 발생할 때까지 대기해야 할 때
🛠 코드 흐름	if event.is_set(): 사용	event.wait() 사용

✅ event.is_set() → 비동기적으로 상태를 확인하고 싶을 때 사용.
✅ event.wait() → 특정 시점까지 코드 실행을 멈추고 기다리고 싶을 때 사용

 

---

2. 스레드 arg 중 daemon=True/False 의 의미

 

🔥 데몬 스레드와 일반(Non-daemon) 스레드 차이

🔹 구분🔥 일반 스레드 (daemon=False, 기본값)🚀 데몬 스레드 (daemon=True)

실행 방식	메인 프로그램이 종료돼도 스레드가 계속 실행됨	메인 프로그램이 종료되면 자동으로 스레드 종료
언제 사용?	반드시 끝까지 실행해야 하는 작업 (예: 파일 저장, 데이터베이스 업데이트)	백그라운드에서 실행되는 작업 (예: 로그 기록, 음악 재생)
수동 종료 필요?	thread.join()을 사용해야 안전하게 종료됨	메인 프로그램이 종료되면 자동 종료

import threading
import time
from audio.audio_player import play_tuning_audio, stop_tuning_audio

# ✅ `tuning_done`을 전역적으로 선언
tuning_done = threading.Event()

def main():
    """메인 실행 함수"""
    print("🚀 튜닝 사운드 재생 시작!")
    tuning_thread = threading.Thread(target=play_tuning_audio, args=(tuning_done,))
    tuning_thread.start()

    # ✅ 5초 후 튜닝 완료 (예제)
    time.sleep(5)
    print("✅ 튜닝 완료! 오디오 중단")
    tuning_done.set()  # ✅ 이벤트 발생 → 튜닝 종료 신호

    tuning_thread.join()
    print("🎵 모든 작업 완료!")

if __name__ == "__main__":
    main()