Python使用multiprocessing实现可终止命令行加载动画问题排查
问题根因
- 终止方法失效:
multiprocessing.Process启动的子进程拥有独立内存空间,你传入的stopper对象会被深拷贝一份到子进程,主进程修改stopper.run的操作无法同步到子进程,子进程的循环永远不会终止。 - PowerShell 输出异常:Windows终端默认开启输出缓冲,
print输出的内容不会实时渲染到界面,同时你代码的循环内写了两次sleep(0.25),实际动画更新间隔是0.5秒,不符合需求。
实现方案
推荐优先用线程实现,没有跨进程通信开销,逻辑更简单:
from time import sleep import threading positions = ["\\", "|", "-", "/"] # 补全四个旋转符动画更流畅 def display_loading_until_event(message): stop_flag = False def animation_task(): pos = 0 while not stop_flag: # 加flush=True强制实时刷新输出,解决Windows终端缓冲问题 print(f"\r{positions[pos]}{message}", end="", flush=True) pos = (pos + 1) % len(positions) sleep(0.25) # 停止后输出完成标识 print(f"\r✔{message}", flush=True) # 启动守护线程,主进程退出时线程会自动销毁 t = threading.Thread(target=animation_task, daemon=True) t.start() def stop(): nonlocal stop_flag stop_flag = True t.join() # 等待动画线程退出 return stop # 测试代码 if __name__ == "__main__": endfnc = display_loading_until_event("Hello loading") sleep(4) endfnc()
如果必须用多进程实现,改用multiprocessing.Value做跨进程共享的停止信号即可:
from time import sleep from multiprocessing import Process, Value positions = ["\\", "|", "-", "/"] def display_loading_until_event(message): # 'b'表示布尔类型,初始值为True表示运行中 stop_flag = Value('b', True) def animation_task(stop_flag, message): pos = 0 while stop_flag.value: print(f"\r{positions[pos]}{message}", end="", flush=True) pos = (pos + 1) % len(positions) sleep(0.25) print(f"\r✔{message}", flush=True) p = Process(target=animation_task, args=(stop_flag, message), daemon=True) p.start() def stop(): stop_flag.value = False p.join() return stop # 测试代码 if __name__ == "__main__": endfnc = display_loading_until_event("Hello loading") sleep(4) endfnc()
注意多进程版本必须把
positions定义放在函数外,或者作为参数传入子进程,否则子进程无法访问父进程的变量。
内容的提问来源于stack exchange,提问作者Qrashi
相关产品推荐
相关产品推荐

