Python如何为函数设置最大运行时长，超时调用备选函数？

首选跨平台实现（支持Windows/Linux/macOS） #

基于concurrent.futures.ThreadPoolExecutor实现，适配所有操作系统，对原有业务代码侵入性最低：

import pandas as pd
from tqdm import tqdm
from concurrent.futures import ThreadPoolExecutor, TimeoutError

# 全局创建单线程池，避免循环内重复创建损耗性能
executor = ThreadPoolExecutor(max_workers=1)

for i in tqdm(range(891)):
    current_text = df['body_text'].iloc[i]
    try:
        # 提交summary任务到线程池，设置120秒超时
        future = executor.submit(summary, current_text)
        result = future.result(timeout=120)
        df_summary['body_text'].iloc[i] = result
    except TimeoutError:
        # 超时后执行兜底的preprocess逻辑
        df_summary['body_text'].iloc[i] = preprocess(current_text)

# 所有任务处理完成后关闭线程池
executor.shutdown()

注意事项 #

• 用.iloc[i]做索引是更规范的pandas写法，可避免DataFrame索引非默认整数序列时出现赋值错误
• 该方案对IO密集型的summary函数适配最好，如果是CPU密集型任务，GIL导致的超时判断延迟最多在秒级，针对120秒的超时阈值可以忽略不计

可选轻量实现（仅支持Linux/macOS） #

基于系统信号实现，性能消耗更低，但不兼容Windows系统：

import signal
from tqdm import tqdm

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

def timeout_handler(signum, frame):
    raise TimeoutException

for i in tqdm(range(891)):
    current_text = df['body_text'].iloc[i]
    # 注册超时信号，120秒后触发异常
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(120)
    try:
        result = summary(current_text)
        df_summary['body_text'].iloc[i] = result
    except TimeoutException:
        df_summary['body_text'].iloc[i] = preprocess(current_text)
    finally:
        # 取消闹钟，避免影响下一轮循环
        signal.alarm(0)

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