Python 3.7中能否统计函数运行时长并在超时时自动终止

可以实现，下面分两种常用场景给出可直接运行的代码示例：

方案1：Unix/Linux/macOS 专用（实现更轻量） #

基于系统SIGALRM信号实现，不需要额外依赖，性能损耗极低：

import signal
import time
from functools import wraps

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

def timeout(seconds):
    def decorator(func):
        # 超时触发的回调函数
        def _handle_timeout(signum, frame):
            raise TimeoutError(f"函数运行超时，设定阈值为{seconds}秒")
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 注册SIGALRM信号的处理函数
            signal.signal(signal.SIGALRM, _handle_timeout)
            # 设定闹钟，seconds秒后发送SIGALRM信号
            signal.alarm(seconds)
            start_time = time.perf_counter()
            try:
                result = func(*args, **kwargs)
            finally:
                # 取消闹钟
                signal.alarm(0)
                # 计算运行时长
                run_duration = time.perf_counter() - start_time
                print(f"函数{func.__name__}实际运行时长：{run_duration:.2f}秒")
            return result
        return wrapper
    return decorator

# 测试示例
@timeout(3) # 设定超时阈值为3秒
def test_func(delay):
    time.sleep(delay)
    return "运行完成"

if __name__ == "__main__":
    try:
        print(test_func(2)) # 延迟2秒，不会超时
        print(test_func(5)) # 延迟5秒，会触发超时
    except TimeoutError as e:
        print(e)

• 限制：Windows系统不支持SIGALRM信号，无法使用该方案

方案2：跨平台通用（兼容Windows） #

基于多进程实现，把目标函数放到子进程运行，主进程监控时长，超时直接终止子进程：

import time
import multiprocessing
from functools import wraps

class TimeoutError(Exception):
    pass

def timeout(seconds):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 定义子进程运行的包装，把返回值存在队列里
            def run_func(queue, *args, **kwargs):
                try:
                    res = func(*args, **kwargs)
                    queue.put(res)
                except Exception as e:
                    queue.put(e)
            
            queue = multiprocessing.Queue()
            start_time = time.perf_counter()
            p = multiprocessing.Process(target=run_func, args=(queue, *args), kwargs=kwargs)
            p.start()
            # 等待子进程运行，最多等seconds秒
            p.join(seconds)
            
            run_duration = time.perf_counter() - start_time
            print(f"函数{func.__name__}实际运行时长：{run_duration:.2f}秒")
            
            # 如果子进程还在运行，就强制终止
            if p.is_alive():
                p.terminate()
                p.join()
                raise TimeoutError(f"函数运行超时，设定阈值为{seconds}秒")
            # 从队列里取结果
            result = queue.get()
            if isinstance(result, Exception):
                raise result
            return result
        return wrapper
    return decorator

# 测试示例
@timeout(3)
def test_func(delay):
    time.sleep(delay)
    return "运行完成"

if __name__ == "__main__":
    try:
        print(test_func(2))
        print(test_func(5))
    except TimeoutError as e:
        print(e)

• 注意事项：被装饰的函数参数和返回值必须是可序列化的（Python pickle模块支持的类型），否则会运行报错
• 小缺点：进程启停会有毫秒级的额外开销，对计时精度要求极高的场景可以根据情况忽略

内容的提问来源于stack exchange，提问作者Crisóstomo Musa