本文介绍如何使用 threading.event 协调多线程，实现在用户输入特定指令（如 "r"）时立即停止 pygame 音频播放，解决主线程初始化、事件同步与资源清理等关键问题。
本文介绍如何使用 threading.event 协调多线程，实现在用户输入特定指令（如 "r"）时立即停止 pygame 音频播放，解决主线程初始化、事件同步与资源清理等关键问题。
在 Python 多线程音频控制场景中，直接在子线程中调用 pygame.init() 会导致未定义行为（Pygame 的 mixer 模块要求初始化必须在主线程完成），且原始代码中存在语法错误（如 break 出现在 while 循环外）、逻辑竞态（如对 deistvie[-1] 的无保护访问）以及资源未释放风险（如未调用 pygame.mixer.music.unload()）。以下是一个健壮、可运行的解决方案。
✅ 正确架构要点
-
pygame.init()必须在主线程调用：确保 mixer 系统正确初始化； -
使用
threading.Event实现线程间安全通信：避免轮询全局列表或共享变量； -
event.wait(timeout)实现非阻塞轮询：在音频播放循环中定期检查中断信号； -
及时释放资源：调用
pygame.mixer.music.stop()和unload()防止内存泄漏。
✅ 完整可运行代码
import pygame
from threading import Event, Thread
MP3 = "kukla-poet.mp3"
def f_music(event: Event) -> None:
"""音频播放线程：加载并播放 MP3，响应中断事件"""
pygame.mixer.music.load(MP3)
pygame.mixer.music.play()
# 每秒检查一次是否收到停止信号
while pygame.mixer.music.get_busy():
if event.wait(1.0): # 等待最多1秒；若事件被 set，则返回 True
pygame.mixer.music.stop()
break
pygame.mixer.music.unload() # 释放音频资源
def f_run(event: Event) -> None:
"""输入监听线程：等待用户输入 'r' 后触发中断"""
while True:
user_input = input("Vvedi deistvie: ").strip()
if user_input == "r":
print("Received 'r' — stopping music...")
event.set()
break
if __name__ == "__main__":
# ✅ 关键：Pygame 初始化必须在主线程
pygame.init()
pygame.mixer.init() # 显式初始化 mixer（更稳妥）
stop_event = Event()
# 启动两个守护线程
music_thread = Thread(target=f_music, args=(stop_event,), name="MusicThread")
input_thread = Thread(target=f_run, args=(stop_event,), name="InputThread")
music_thread.start()
input_thread.start()
# 等待两线程自然结束（推荐方式，避免忙等待）
music_thread.join()
input_thread.join()
print("Both threads finished. Exiting.")
⚠️ 注意事项与最佳实践
-
MP3 文件路径需有效：请确保
"kukla-poet.mp3"存在于当前工作目录，或使用绝对路径； -
避免
input()在非主线程中的潜在阻塞问题：本例中f_run运行于独立线程，input()是安全的，但不可用于 GUI 或异步环境； -
不要省略
pygame.mixer.unload()：尤其在重复运行程序时，防止音频句柄堆积； -
如需支持暂停/恢复而非硬停止，可改用
pygame.mixer.music.pause()/unpause()，但需额外状态管理； -
扩展性提示：若需响应多个命令（如
"p"暂停、"s"停止），建议改用queue.Queue或threading.Condition替代单一 Event。
该方案兼顾简洁性与鲁棒性，是 Python 多线程音频控制的推荐实现模式。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
