python函数超时机制

### 实现 Python 中函数调用的超时控制

在 Python 编程环境中，当遇到长时间运行的任务或外部服务请求时，可能会导致程序卡住。为了避免这种情况发生，可以采用多种方式来设置函数调用的超时机制。

#### 使用 `signal` 模块处理超时 (仅适用于 Unix/Linux)

对于 Linux 和 macOS 平台上的应用程序来说，可以通过信号量中断正在执行的操作：

import signal

class TimeoutException(Exception): pass 

def handler(signum, frame):
    raise TimeoutException()

# 设置定时器，在指定秒数后触发 SIGALRM 信号
signal.signal(signal.SIGALRM, handler)
signal.alarm(5)  # 设定超时时长为5秒

try:
    result = some_long_running_function()
except TimeoutException:
    print("Function call timed out!")
finally:
    signal.alarm(0)  # 取消闹钟

此方法利用操作系统级别的特性实现了较为简单的超时检测功能[^1]。

#### 利用多线程或多进程实现非阻塞操作

另一种常见做法是通过启动新的线程或子进程去异步地完成任务，并设定最大等待时间。如果超过这个时间段还没有得到返回结果，则认为该次调用失败并终止它。

##### 基于 `threading` 的解决方案:

from threading import Thread
import queue as Queue

def run_with_timeout(func, args=(), kwargs={}, timeout_duration=10, default=None):
    class InterruptableThread(Thread):
        def __init__(self):
            super().__init__()
            self.result_queue = Queue.Queue()

        def run(self):
            try:
                ret = func(*args,**kwargs)
                self.result_queue.put(ret)
            except Exception as e:
                self.result_queue.put(e)

    it = InterruptableThread()
    it.start()
    it.join(timeout_duration)
    
    if not it.is_alive():
        try:
            res = it.result_queue.get_nowait()  
            if isinstance(res, BaseException):
                raise res
            return res
        except Queue.Empty:
            return default
    else:
        del it  # 清理未结束的线程资源
        return None

上述代码片段定义了一个辅助工具类 `InterruptableThread` 来封装目标函数及其参数，同时提供了一种安全的方式获取其计算成果或者异常信息。注意这里并没有强制杀死仍在工作的后台线程；相反，只是简单地忽略了它们的存在[^2]。

##### 运用 `multiprocessing` 库创建独立的工作单元:

from multiprocessing import Process, Manager

def target_func(shared_dict, key, *args, **kwds):
    shared_dict[key] = your_original_function(*args, **kwds)

manager = Manager()
shared_data = manager.dict()
p = Process(target=target_func, args=(shared_data,'result', arg1,arg2))
p.start()
p.join(timeout_seconds)

if p.is_alive(): 
    p.terminate()  # 明确停止活动中的子进程
    p.join()
print('Result:', shared_data.get('result'))

这种方法更加彻底地隔离了潜在的风险源——即使被监控的目标函数内部出现了致命错误也不会影响主线程的安全性[^4]。

#### 结合上下文管理器简化 API 调用场景下的超时逻辑

针对网络请求或其他 I/O 密集型任务，还可以考虑借助第三方库如 `requests` 或者标准库里的 `asyncio` 提供更高级别的抽象层次来进行优雅降级：

import requests

response = requests.get(url='https://example.com/api/data',
                        params={'key': 'value'},
                        timeout=5.0)  # 单位为秒
data = response.json()

在这个例子中，一旦 HTTP GET 请求超过了预设的时间限制就会抛出相应的异常而不是无休止地挂起整个应用流程[^3]。