如何为函数设置30秒超时并按指定参数列表依次重试？

方案选择 #

选multiprocessing而非multithreading：

• 你的get_frequent_items属于CPU密集型任务（从耗时差异能看出，底层是频繁项集计算逻辑，比如Apriori、FP-Growth这类），Python的全局解释器锁（GIL）会限制多线程在CPU密集场景下的并行能力，多进程能绕过GIL，让每个进程独立占用CPU核心。
• 多进程可以安全地终止超时任务，避免多线程强行终止时可能出现的资源泄漏问题。

代码实现 #

import multiprocessing
import time
import pandas as pd

# 替换为你实际的get_frequent_items函数
def get_frequent_items(support_threshold):
    # 示例模拟逻辑：support越小耗时越长
    if support_threshold <= 0.3:
        time.sleep(35)  # 模拟超时场景
    else:
        time.sleep(25)  # 模拟正常完成场景
    return pd.DataFrame({'itemset': [['A', 'B'], ['B', 'C']], 'support': [support_threshold, support_threshold]})

# 进程执行的包装函数，用于传递结果
def run_task(support, result_queue):
    try:
        df = get_frequent_items(support)
        result_queue.put(df)
    except Exception as e:
        result_queue.put(e)

def find_valid_frequent_items():
    var_support_options = [0.18, 0.2, 0.25, 0.3, 0.35]
    timeout = 30  # 超时阈值（秒）

    for support in var_support_options:
        print(f"尝试参数: {support}")
        result_queue = multiprocessing.Queue()
        # 创建并启动子进程
        process = multiprocessing.Process(target=run_task, args=(support, result_queue))
        process.start()
        
        # 等待进程结束，最多等待timeout秒
        process.join(timeout)
        
        if process.is_alive():
            # 超时终止进程
            print(f"参数{support}运行超时，终止并尝试下一个")
            process.terminate()
            process.join()  # 确保进程彻底退出
        else:
            # 正常结束，获取结果
            result = result_queue.get()
            if isinstance(result, Exception):
                print(f"参数{support}运行出错: {result}")
                continue
            print(f"参数{support}运行成功，返回结果")
            return result
    
    # 所有参数均失败的情况
    print("所有候选参数均超时或运行失败")
    return None

# 主执行逻辑
if __name__ == "__main__":
    frequent_itemsets = find_valid_frequent_items()
    if frequent_itemsets is not None:
        # 后续依赖frequent_itemsets的计算代码
        print("启动后续计算...")
        print(frequent_itemsets)

关键说明 #

• 按从小到大的顺序遍历候选参数，每次仅启动一个进程，避免资源过度占用。
• 用process.join(timeout)实现超时等待，超时后调用terminate()强制终止进程，再用join()确保进程彻底清理。
• 通过multiprocessing.Queue实现进程间结果传递，同时捕获执行异常。

注意事项 #

• 确保你的get_frequent_items函数支持序列化（可被pickle），否则无法在多进程间传递。
• 如果函数依赖全局变量或外部资源，需在子进程中重新初始化，或使用多进程管理器/共享内存传递。

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