python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python处理函数调用超时

Python处理函数调用超时方法的详细教学

作者：Python资讯站

限制函数调用的最大时间是一种非常实用的技术手段,能够帮助开发者更好地控制程序的行为,本文将介绍四种Python处理函数调用超时的方法,大家可以根据需要进行选择

限制函数调用的最大时间是一种非常实用的技术手段，能够帮助开发者更好地控制程序的行为，提升用户体验，同时确保系统的稳定性和可靠性。

func-timeout

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

1. 安装 func-timeout

可以使用 pip 安装：

pip install func-timeout

2. 基本用法

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

from func_timeout import func_timeout, FunctionTimedOut
import time

def long_running_task():
    time.sleep(5)  # 模拟长时间运行的任务
    return "Task completed"
try:
    result = func_timeout(3, long_running_task)  # 设置3秒超时
    print(result)
except FunctionTimedOut:
    print("Function execution timed out!")

解释：

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

from func_timeout import func_set_timeout
import time

@func_set_timeout(2)  # 限制该函数的运行时间为2秒
def long_task():
    time.sleep(5)  # 任务实际需要5秒
    return "Finished"

try:
    print(long_task())
except FunctionTimedOut:
    print("Function execution timed out!")

自定义进程

除了使用上面的库，也可以自己使用一个进程来计时和检测超时，另一个进程来调用 Python 函数。

以下是具体实现代码：

import time
from itertools import count
from multiprocessing import Process

def inc_forever():
    print('Starting function inc_forever()...')
    while True:
        time.sleep(1)
        print(next(counter))

def return_zero():
    print('Starting function return_zero()...')
    return 0

if __name__ == '__main__':
    # counter 是一个无限迭代器
    counter = count(0)

    p1 = Process(target=inc_forever, name='Process_inc_forever')
    p2 = Process(target=return_zero, name='Process_return_zero')
    p1.start()
    p2.start()
    p1.join(timeout=5)
    p2.join(timeout=5)
    p1.terminate()
    p2.terminate()

if p1.exitcode is None:
    print(f'Oops, {p1} timeouts!')

if p2.exitcode == 0:
    print(f'{p2} is luck and finishes in 5 seconds!')

运行结果如下：

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

从退出码可以看出，inc_forever() 函数超时了（退出码为 None），而 return_zero() 函数在 5 秒内成功完成。

subprocess

参数设置超时

从 Python 3.5 开始，subprocess 模块提供了一个便捷且推荐使用的 run() API，它内置了超时支持。

以下是示例代码：

import subprocess

r = subprocess.run(['echo', 'hello timeout'], timeout=5)
print(
    f'''type(r)={type(r)},
    r.args={r.args},
    r.returncode={r.returncode},
    r.stdout={r.stdout},
    r.stderr={r.stderr}''')
try:
    r = subprocess.run(['ping', 'www.google.com'], timeout=5)
except subprocess.TimeoutExpired as e:
    print(e)

运行结果如下：

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

当超时时，会抛出一个 TimeoutExpired 异常。

信号（Signals）

对于 UNIX 系统，还可以使用 signal 模块，通过在 5 秒后向处理器发送信号来引发异常。不过，这种方法相对底层且不够直观。

import signal
def handler(signum, frame):
    raise TimeoutError("函数超时")
def my_function():
    pass
signal.signal(signal.SIGALRM, handler)
signal.alarm(5)
try:
    my_function()
except TimeoutError:
    print("函数超时")
finally:
    signal.alarm(0)

知识补充

Python函数超时后不抛出异常的方法

1.func_timeout库的使用及异常处理

func_timeout 库简介：使用python第三方 func_timeout 模块中提供的 func_set_timeout 装饰器可以非常简单的设置python程序的超时时间。

import time
import func_timeout
from func_timeout import func_set_timeout

@func_set_timeout(3)
def task():
    while True:
        print('hello world')
        time.sleep(1)

if __name__ == '__main__':
    try:
        task()
    except func_timeout.exceptions.FunctionTimedOut:
        print("执行已超时3秒")

上述代码使用@func_set_timeout(3)，设置了函数超时时间为3秒，同时使用except func_timeout.exceptions.FunctionTimedOut: print("执行已超时3秒")，进行异常处理，输出结果如下图：

2.func_timeout阻塞主线程问题

什么是线程阻塞：在某一时刻某一个线程在运行一段代码的时候，这时候另一个线程也需要运行，但是在运行过程中的那个线程执行完成之前，另一个线程无法获取到CPU执行权，这时就会造成线程阻塞。

这里发现另外一个问题，链接中的如下代码，其实还是会阻塞主线程。

import time
from func_timeout import func_set_timeout
​
@func_set_timeout(3)
def task():
    while True:
        print('hello world')
        time.sleep(1)
​
if __name__ == '__main__':
    task()

怎么验证呢，我们在task()后面加一句print("a")，改成如下代码：

import time
from func_timeout import func_set_timeout

@func_set_timeout(3)
def task():
    while True:
        print('hello world')
        time.sleep(1)

if __name__ == '__main__':
    task()
    print("a")

输出结果如下：

从程序的输出结果可以发现，控制台并没有打印字母a，所以按链接的代码，还是会导致整个程序阻塞，等待task()执行完毕，直到超时报错。

3.启动多线程解决阻塞问题

Python的多线程：Python提供两个模块进行多线程的操作，分别是thread和threading，前者是比较低级的模块，用于更底层的操作，一般应用级别的开发不常用。

为了解决阻塞主线程的问题，这里尝试使用threading库，启动多线程来执行task()，代码如下：

import time
from func_timeout import func_set_timeout
from threading import Thread

@func_set_timeout(3)
def task():
    while True:
        print('hello world')
        time.sleep(1)

if __name__ == '__main__':
    t1 = Thread(target=task, args=())
    t1.start()
    print("a")

程序运行结果，如下图：

从程序输出的结果可以看出，现在程序并没有阻塞主线程，如期在控制台上打印了字母a。同时，在等待3秒后，task()函数超时抛出了超时异常。

4.try语句封装函数，实现全程捕获异常

异常处理是编程语言或计算机硬件里的一种机制，用于处理软件或信息系统中出现的异常状况（即超出程序正常执行流程的某些特殊条件）。

到这里，其实还是有一个问题没有解决。那就是刚才程序输出的异常还没被捕获和处理，为了得到完整的解决方案，代码中还需要加上异常处理的部分，同时还需要另外写一个函数try_task()，实现全程捕获异常。

代码如下：

import time
import func_timeout
from func_timeout import func_set_timeout
from threading import Thread


def try_task():
    try:
        task()
    except func_timeout.exceptions.FunctionTimedOut:
        print("执行已超时3秒")


@func_set_timeout(3)
def task():
    while True:
        print('hello world')
        time.sleep(1)
      

if __name__ == '__main__':
    t1 = Thread(target=try_task, args=())
    t1.start()
    print("a")

最终程序输出结果如下图，实现了不阻塞主线程，3秒后超时抛出异常并被正常捕获。

总结

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

这些方法各有优劣，开发者可根据实际需求选择合适的方案。

到此这篇关于Python处理函数调用超时方法的详细教学的文章就介绍到这了,更多相关Python处理函数调用超时内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文