Python中如何为for循环单次迭代设置超时自动跳过机制

原代码的核心问题有两个：

1. 起始时间统计放在循环外部，计算的是整个循环的总运行时长，而非单次迭代的时长
2. 常规的顺序执行逻辑无法主动中断卡在任务内部的执行流程，即使时间到了也无法跳出当前迭代，必须借助可中断的执行机制实现需求

以下是两种可直接使用的实现方案：

方案1：信号量实现（仅适用于Unix/Linux/macOS平台） #

该方案轻量无额外依赖，利用系统信号实现超时中断，无需修改原有任务逻辑：

import time
import signal

# 自定义超时异常
class IterationTimeout(Exception):
    pass

# 信号处理函数，超时触发时抛出异常
def timeout_handler(signum, frame):
    raise IterationTimeout("当前迭代执行超时")

# 绑定SIGALRM信号的处理逻辑
signal.signal(signal.SIGALRM, timeout_handler)

profile = [p for p in range(1000, 2000)]
for item in profile:
    try:
        # 单次迭代开始前设置30秒超时闹钟
        signal.alarm(30)
        # 此处替换为你原本的任务逻辑
        print(f"正在处理：{item}")
        time.sleep(10) # 模拟正常执行，若替换为time.sleep(40)会触发超时
    except IterationTimeout:
        print(f"处理{item}超时，跳过当前迭代")
        continue
    finally:
        # 任务正常完成后关闭闹钟，避免影响下一次迭代
        signal.alarm(0)

注意：Windows系统不支持SIGALRM信号，该方案无法在Windows环境运行，需要跨平台请使用方案2。

方案2：线程池实现（全平台通用） #

该方案利用多线程将单次任务隔离执行，超时后直接取消当前任务，兼容所有操作系统，支持灵活调整并发数：

import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError

# 将单次迭代的任务逻辑封装为独立函数
def process_single_item(item):
    # 此处替换为你原本的任务逻辑
    print(f"正在处理：{item}")
    time.sleep(10) # 模拟正常执行，若替换为time.sleep(40)会触发超时
    return f"处理{item}完成"

if __name__ == "__main__":
    profile = [p for p in range(1000, 2000)]
    # max_workers设为1保持和原逻辑一致的串行执行，需要提速可自行调大并发数
    with ThreadPoolExecutor(max_workers=1) as executor:
        for item in profile:
            # 提交任务到线程池
            task = executor.submit(process_single_item, item)
            try:
                # 等待任务执行结果，设置30秒超时
                res = task.result(timeout=30)
                print(res)
            except TimeoutError:
                print(f"处理{item}超时，跳过当前迭代")
                # 取消超时未完成的任务释放资源
                task.cancel()
                continue

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