Python函数超时自动退出实现教程

为什么需要函数超时控制？

在Python开发中，经常会遇到需要限制函数执行时间的情况：

防止长时间运行的函数阻塞主程序
避免外部API调用等待时间过长
控制数据处理任务的最大执行时间
防止死循环或资源耗尽

本教程将介绍多种实现Python函数超时自动退出的方法。

方法一：使用signal模块（Unix/Linux系统）

signal模块提供进程间通信机制，可用于设置超时处理。


import signal

class TimeoutException(Exception):
    pass

def timeout_handler(signum, frame):
    raise TimeoutException("Function timed out")

def run_with_timeout(func, args=(), kwargs={}, timeout=5):
    # 设置信号处理函数
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout)  # 设置超时时间
    
    try:
        result = func(*args, **kwargs)
    finally:
        signal.alarm(0)  # 取消闹钟
    
    return result

# 使用示例
def long_running_task():
    import time
    time.sleep(10)
    return "任务完成"

try:
    result = run_with_timeout(long_running_task, timeout=3)
    print(result)
except TimeoutException as e:
    print(e)  # 输出: Function timed out

优点与限制

优点	限制
简单易用，代码量少	仅适用于Unix/Linux系统
执行效率高	不支持多线程环境
能有效中断阻塞操作	主线程中才能使用

方法二：使用multiprocessing模块（跨平台）

multiprocessing模块支持跨平台，使用进程代替线程实现超时控制。


from multiprocessing import Process, Queue
import time

def run_in_process(func, args=(), kwargs={}, timeout=5):
    def wrapper(q, *a, **kw):
        try:
            result = func(*a, **kw)
            q.put(result)
        except Exception as e:
            q.put(e)
    
    q = Queue()
    p = Process(target=wrapper, args=(q,)+args, kwargs=kwargs)
    p.start()
    
    # 等待超时时间
    p.join(timeout)
    
    if p.is_alive():
        p.terminate()  # 终止进程
        p.join()
        raise TimeoutError(f"函数执行超过 {timeout} 秒")
    
    result = q.get()
    if isinstance(result, Exception):
        raise result
    return result

# 使用示例
def long_calculation(n):
    time.sleep(n)
    return f"计算完成，耗时{n}秒"

try:
    result = run_in_process(long_calculation, args=(10,), timeout=3)
    print(result)
except TimeoutError as e:
    print(e)  # 输出: 函数执行超过 3 秒

优点与限制

优点	限制
真正的跨平台解决方案	进程间通信较复杂
能强制终止长时间运行的函数	进程启动开销较大
支持多线程环境	函数需要是可pickle的

方法三：装饰器实现（推荐方式）

结合装饰器与multiprocessing，创建更优雅的解决方案。


import functools
from concurrent.futures import ProcessPoolExecutor, TimeoutError as FutureTimeout

def timeout(seconds=5):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            with ProcessPoolExecutor(max_workers=1) as executor:
                future = executor.submit(func, *args, **kwargs)
                try:
                    result = future.result(timeout=seconds)
                    return result
                except FutureTimeout:
                    executor.shutdown(wait=False, cancel_futures=True)
                    raise TimeoutError(f"函数 {func.__name__} 执行超过 {seconds} 秒")
        return wrapper
    return decorator

# 使用示例
@timeout(seconds=3)
def fetch_data(url):
    import time
    import random
    # 模拟网络请求
    time.sleep(random.randint(1, 5))
    return f"从 {url} 获取数据成功"

try:
    print(fetch_data("https://example.com/api/data"))
except TimeoutError as e:
    print(e)

使用场景建议

网络请求：限制HTTP请求的最大等待时间
数据处理：确保数据处理任务不会无限期运行
外部命令：执行外部程序时设置超时
用户输入：限制用户输入等待时间

方法总结与选择建议

方法	适用平台	使用场景	推荐指数
signal模块	Unix/Linux	简单脚本，单线程环境	★★☆☆☆
multiprocessing	跨平台	复杂应用，需要强制终止	★★★★☆
装饰器实现	跨平台	生产环境，代码复用	★★★★★