如何在Python 3中为函数设置限时执行（如5分钟内）

要实现model.predict(image=img)运行超过5分钟自动终止并跳过当前循环，有两种基于Python标准库的简便方案，无需额外安装第三方包：

方法1：使用concurrent.futures.ThreadPoolExecutor #

这是最直观的实现方式，利用线程池提交任务并设置超时时间：

import os
from concurrent.futures import ThreadPoolExecutor, TimeoutError
# 假设你的DataURI和model已完成导入

for filename in os.listdir():
    if filename.endswith(".jpg") or filename.endswith(".png"):
        print(filename)
        img = DataURI.from_file(filename)
        
        # 将预测任务提交到线程池
        with ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(model.predict, image=img)
            try:
                # 设置超时时间：5分钟=300秒
                output = future.result(timeout=300)
                # 此处可添加对output的后续处理逻辑
            except TimeoutError:
                print(f"{filename} 预测超时，跳过当前文件")
                continue

说明 #

• ThreadPoolExecutor是Python3.2+自带的标准库，无需额外安装。
• 通过将预测任务放入子线程执行，主线程等待future.result()时触发超时判断，超时后抛出TimeoutError，捕获后执行continue跳过当前文件处理。

方法2：使用threading模块手动控制 #

如果需要更底层的线程控制，可直接用threading.Thread结合join()超时机制：

import os
import threading
from queue import Queue
# 假设你的DataURI和model已完成导入

def predict_task(img, result_queue):
    try:
        output = model.predict(image=img)
        result_queue.put(output)
    except Exception as e:
        result_queue.put(e)

for filename in os.listdir():
    if filename.endswith(".jpg") or filename.endswith(".png"):
        print(filename)
        img = DataURI.from_file(filename)
        
        result_queue = Queue()
        thread = threading.Thread(target=predict_task, args=(img, result_queue))
        thread.start()
        # 等待线程执行完成，超时时间300秒
        thread.join(timeout=300)
        
        if thread.is_alive():
            print(f"{filename} 预测超时，跳过当前文件")
            continue
        else:
            result = result_queue.get()
            if isinstance(result, Exception):
                print(f"{filename} 预测执行出错: {result}")
                continue
            output = result
            # 此处可添加对output的后续处理逻辑

说明 #

• 借助Queue在主线程和子线程间传递预测结果或异常信息。
• thread.join(timeout=300)等待线程执行完毕，超时后若线程仍处于存活状态，说明预测任务未完成，直接跳过当前文件。

注意事项 #

• 若模型依赖GPU（如TensorFlow/PyTorch框架），部分场景下子线程运行可能存在资源冲突，此时可改用multiprocessing进程池替代线程池，但需注意模型在进程间的序列化问题，可能需要每个进程重新加载模型实例。

内容的提问来源于stack exchange，提问作者plshelpmeout