Python如何让一个模块的函数等待另一独立模块函数成功启动
问题根本原因
你当前代码不生效的核心原因是:通过subprocess.Popen启动的module3.py是完全独立的新进程,两个进程会分别加载一次module2.py,各自持有flag变量的内存副本,module3.py修改的只是自己进程内的flag值,完全不会影响module1.py进程里加载的module2.flag,所以你永远读到的都是初始的False。
解决方案:共享文件标记方案(改动最小)
不需要调整你的整体架构,只要把标志位的存储从内存变量换成磁盘临时文件即可,同时补充轮询+超时逻辑,实现成本极低。
修改后的module2.py
提供标志位读写/清除的通用函数:
import os # 用固定名称的文件做状态标记 FLAG_FILE = "module3_start_success.flag" def set_flag_success(): with open(FLAG_FILE, "w", encoding="utf-8") as f: f.write("1") def get_flag_status(): return os.path.exists(FLAG_FILE) and open(FLAG_FILE, encoding="utf-8").read().strip() == "1" def clear_flag(): if os.path.exists(FLAG_FILE): os.remove(FLAG_FILE)
修改后的module3.py
蓝牙连接成功后调用标记函数即可:
import module2 def run(): # do some stuff # try: connect.ToDevice() # 连接蓝牙设备 module2.set_flag_success() # 写入成功标记 doCoolStuff() except Exception as e: print('startup failed: ', e) if __name__ == "__main__": # do more cool stuff # try: loop.run_until_complete(run()) except KeyboardInterrupt: print('User has stopped run() from executing.')
修改后的module1.py
补充30秒超时轮询逻辑:
import module2 import subprocess import os import time def foo(): try: # 先清除上次运行残留的标记 module2.clear_flag() # 启动module3进程 subprocess.Popen(["python", "module3.py"], cwd=os.getcwd()) # 超时时间配置为30秒 timeout = 30 start_time = time.time() start_success = False # 每0.5秒检查一次标记,避免占用过高CPU while time.time() - start_time < timeout: if module2.get_flag_status(): start_success = True break time.sleep(0.5) if start_success: print('module3.py started successfully!') someOtherFunction() else: print('module3.py start timeout or failed') except Exception as e: print('something happened! ', e)
该方案跨平台兼容,调试成本低,完全符合你希望低改动实现的需求。
内容的提问来源于stack exchange,提问作者Augustus
相关产品推荐
相关产品推荐

