前言

在实际开发过程中，我们可能会遇到一些场景，需要对函数的执行时间进行限制。例如，当一个函数执行时间过长时，可能会导致程序卡顿、资源占用过高，甚至影响整个系统的稳定性。因此，在某些情况下，我们希望限制函数调用的最大时间，以确保程序能够在合理的时间范围内完成任务，或者在超时的情况下采取其他措施。
为了实现这一目标，可以通过多种方式来控制函数的执行时间。例如，可以使用多线程或异步编程技术，在指定的时间范围内监控函数的执行情况。如果函数在规定时间内未能完成执行，则可以中断该函数的运行，并返回一个超时提示或执行备用逻辑。这种方式不仅能够提高程序的健壮性，还能有效避免因单个函数执行时间过长而导致的系统性能问题。
限制函数调用的最大时间是一种非常实用的技术手段，能够帮助开发者更好地控制程序的行为，提升用户体验，同时确保系统的稳定性和可靠性。

func-timeout

func-timeout 是一个 Python 库，允许为函数设置超时时间，防止代码长时间运行或无限阻塞。它适用于需要强制限制执行时间的场景，例如网络请求、计算密集型任务或可能出现死循环的代码。

1. 安装 func-timeout

可以使用 pip 安装：

1. pip install func-timeout

复制代码

2. 基本用法

最常用的方式是 func_timeout，它允许在指定的时间内运行一个函数，超时则抛出异常。

1. from func_timeout import func_timeout, FunctionTimedOut
   
2. import time
   
3. 
   
4. def long_running_task():
   
5.     time.sleep(5)  # 模拟长时间运行的任务
   
6.     return "Task completed"
   
7. 
   
8. try:
   
9.     result = func_timeout(3, long_running_task)  # 设置3秒超时
   
10.     print(result)
    
11. except FunctionTimedOut:
    
12.     print("Function execution timed out!")

复制代码

解释：

• func_timeout(3, long_running_task)：尝试在 3 秒内运行 long_running_task
  
• FunctionTimedOut 异常表示函数超时未完成
  

也可以使用装饰器方式为函数设定超时：

1. from func_timeout import func_set_timeout
   
2. import time
   
3. 
   
4. @func_set_timeout(2)  # 限制该函数的运行时间为2秒
   
5. def long_task():
   
6.     time.sleep(5)  # 任务实际需要5秒
   
7.     return "Finished"
   
8. 
   
9. try:
   
10.     print(long_task())
    
11. except FunctionTimedOut:
    
12.     print("Function execution timed out!")

复制代码

这种方式适用于需要多次调用的函数，避免每次调用都手动设置超时。
func-timeout 本质上还是依赖 多线程 或 多进程 实现超时控制，在某些情况下可能不适用于主线程（如 Jupyter Notebook）。它也不能用于 main 线程内的 while True 死循环，因为 Python 的 GIL 可能会影响信号处理。

自定义进程

除了使用上面的库，也可以自己使用一个进程来计时和检测超时，另一个进程来调用 Python 函数。以下是具体实现代码：

1. import time
   
2. from itertools import count
   
3. from multiprocessing import Process
   
4. 
   
5. def inc_forever():
   
6.     print('Starting function inc_forever()...')
   
7.     while True:
   
8.         time.sleep(1)
   
9.         print(next(counter))
   
10. 
    
11. def return_zero():
    
12.     print('Starting function return_zero()...')
    
13.     return 0
    
14. 
    
15. if __name__ == '__main__':
    
16.     # counter 是一个无限迭代器
    
17.     counter = count(0)
    
18. 
    
19.     p1 = Process(target=inc_forever, name='Process_inc_forever')
    
20.     p2 = Process(target=return_zero, name='Process_return_zero')
    
21.     p1.start()
    
22.     p2.start()
    
23.     p1.join(timeout=5)
    
24.     p2.join(timeout=5)
    
25.     p1.terminate()
    
26.     p2.terminate()
    
27. 
    
28. if p1.exitcode is None:
    
29.     print(f'Oops, {p1} timeouts!')
    
30. 
    
31. if p2.exitcode == 0:
    
32.     print(f'{p2} is luck and finishes in 5 seconds!')

复制代码

运行结果如下：

1. Starting function inc_forever()...
   
2. Starting function return_zero()...
   
3. 0
   
4. 1
   
5. 2
   
6. 3
   
7. 4
   
8. Oops, <Process(Process_inc_forever, started)> timeouts!
   
9. <Process(Process_return_zero, stopped)> is luck and finishes in 5 seconds!

复制代码

从退出码可以看出，

1. inc_forever()

复制代码

函数超时了（退出码为

1. None

复制代码

），而

1. return_zero()

复制代码

函数在 5 秒内成功完成。

subprocess 参数设置超时

从 Python 3.5 开始，

1. subprocess

复制代码

模块提供了一个便捷且推荐使用的

1. run()

复制代码

API，它内置了超时支持。以下是示例代码：

1. import subprocess
   
2. 
   
3. r = subprocess.run(['echo', 'hello timeout'], timeout=5)
   
4. print(
   
5.     f'''type(r)={type(r)},
   
6.     r.args={r.args},
   
7.     r.returncode={r.returncode},
   
8.     r.stdout={r.stdout},
   
9.     r.stderr={r.stderr}'''
   
10. )
    
11. 
    
12. try:
    
13.     r = subprocess.run(['ping', 'www.google.com'], timeout=5)
    
14. except subprocess.TimeoutExpired as e:
    
15.     print(e)

复制代码

运行结果如下：

1. hello timeout
   
2. type(r)=<class 'subprocess.CompletedProcess'>,
   
3.     r.args=['echo', 'hello timeout'],
   
4.     r.returncode=0,
   
5.     r.stdout=None,
   
6.     r.stderr=None
   
7. PING www.google.com (216.58.194.164) 56(84) bytes of data.
   
8. 64 bytes from ...: icmp_seq=1 ttl=54 time=10.4 ms
   
9. 64 bytes from ...: icmp_seq=2 ttl=54 time=5.90 ms
   
10. 64 bytes from ...: icmp_seq=3 ttl=54 time=6.19 ms
    
11. 64 bytes from ...: icmp_seq=4 ttl=54 time=9.04 ms
    
12. 64 bytes from ...: icmp_seq=5 ttl=54 time=16.7 ms
    
13. Command '['ping', 'www.google.com']' timed out after 5 seconds

复制代码

当超时时，会抛出一个

1. TimeoutExpired

复制代码

异常。

信号（Signals）

对于 UNIX 系统，还可以使用

1. signal

复制代码

模块，通过在 5 秒后向处理器发送信号来引发异常。不过，这种方法相对底层且不够直观。

1. import signal
   
2. def handler(signum, frame):
   
3.     raise TimeoutError("函数超时")
   
4. def my_function():
   
5.     pass
   
6. signal.signal(signal.SIGALRM, handler)
   
7. signal.alarm(5)
   
8. try:
   
9.     my_function()
   
10. except TimeoutError:
    
11.     print("函数超时")
    
12. finally:
    
13.     signal.alarm(0)

复制代码

总结

在开发中，限制函数执行时间是提升程序稳定性和用户体验的重要手段。本文介绍了几种实现方法：

• func-timeout 库：通过
  1. func_timeout

  复制代码
  或装饰器
  1. func_set_timeout

  复制代码
  ，可为函数设置超时时间，超时则抛出异常。适用于网络请求或计算密集型任务。
  
• 自定义进程：利用
  1. multiprocessing

  复制代码
  模块创建子进程执行函数，通过
  1. join(timeout)

  复制代码
  控制超时，超时后终止进程。
  
• subprocess 模块：从 Python 3.5 起，
  1. subprocess.run()

  复制代码
  支持超时参数，超时会抛出
  1. TimeoutExpired

  复制代码
  异常，适合外部命令调用。
  
• 信号机制：在 UNIX 系统中，使用
  1. signal

  复制代码
  模块设置超时信号，超时后触发异常，但实现较底层。
  

这些方法各有优劣，开发者可根据实际需求选择合适的方案。
以上就是Python处理函数调用超时的四种方法的详细内容，更多关于Python函数调用超时的资料请关注脚本之家其它相关文章！

来源：https://www.jb51.net/python/339981atm.htm
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！