You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.01 07:45:01