Python如何检测代码运行时长、超时终止并获取不可修改函数的返回值
解决方案
方案1:修改现有multiprocessing实现,用Queue获取返回值
你原来的实现拿不到返回值的核心原因是,子进程的返回值无法直接传递到主进程,通过multiprocessing.Queue作为中间容器存储执行结果即可解决:
from multiprocessing import Process, Queue def run_with_limited_time(func, args, timeout): res_queue = Queue(maxsize=1) # 封装wrapper将函数结果写入队列 def wrapper(): try: result = func(*args) res_queue.put((True, result)) except Exception as e: res_queue.put((False, e)) p = Process(target=wrapper) p.start() p.join(timeout) if p.is_alive(): # 超时终止进程 p.terminate() p.join() return (False, "timeout") else: return res_queue.get() if not res_queue.empty() else (False, "process exit without result")
测试示例
import time def test_func(sleep_sec): time.sleep(sleep_sec) return f"执行完成,耗时{sleep_sec}秒" if __name__ == "__main__": # 耗时小于15秒,正常返回结果 print(run_with_limited_time(test_func, (14,), 15)) # 输出:(True, '执行完成,耗时14秒') # 耗时超过15秒,触发超时 print(run_with_limited_time(test_func, (16,), 15)) # 输出:(False, 'timeout')
方案2:使用concurrent.futures.ProcessPoolExecutor(更简洁)
Python标准库自带的进程池封装了超时控制和结果传递逻辑,不需要手动管理进程和队列,代码量更少:
from concurrent.futures import ProcessPoolExecutor, TimeoutError import time def run_with_limited_time(func, args, timeout): with ProcessPoolExecutor(max_workers=1) as executor: future = executor.submit(func, *args) try: result = future.result(timeout=timeout) return (True, result) except TimeoutError: return (False, "timeout") except Exception as e: return (False, e) # 测试逻辑和方案1完全一致 def test_func(sleep_sec): time.sleep(sleep_sec) return f"执行完成,耗时{sleep_sec}秒" if __name__ == "__main__": print(run_with_limited_time(test_func, (14,), 15)) print(run_with_limited_time(test_func, (16,), 15))
注意事项
- Windows系统下运行多进程代码时,必须将调用逻辑放在
if __name__ == "__main__":代码块内,否则会出现进程重复启动的错误。 - 两种方案都要求函数返回结果是可序列化的对象,常用的字符串、数值、数组、字典等类型都满足要求,符合绝大多数图像分类库的返回场景。
内容的提问来源于stack exchange,提问作者popeye
相关产品推荐
相关产品推荐

