七贝PS网

 找回密码
立即注册
快捷导航
七贝PS网 首页 PS抠图教程 查看内容

fffffffffffffffff

发布者: 七贝PS网

前言

在实际开发过程中,我们可能会遇到一些场景,需要对函数的执行时间进行限制。例如,当一个函数执行时间过长时,可能会导致程序卡顿、资源占用过高,甚至影响整个系统的稳定性。因此,在某些情况下,我们希望限制函数调用的最大时间,以确保程序能够在合理的时间范围内完成任务,或者在超时的情况下采取其他措施。

为了实现这一目标,可以通过多种方式来控制函数的执行时间。例如,可以使用多线程或异步编程技术,在指定的时间范围内监控函数的执行情况。如果函数在规定时间内未能完成执行,则可以中断该函数的运行,并返回一个超时提示或执行备用逻辑。这种方式不仅能够提高程序的健壮性,还能有效避免因单个函数执行时间过长而导致的系统性能问题。

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

func-timeout

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

1. 安装 func-timeout

可以使用 pip 安装:

1

pip install func-timeout

2. 基本用法

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

1

2

3

4

5

6

7

8

9

10

11

12

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!”)

解释:

func_timeout(3, long_running_task):尝试在 3 秒内运行 long_running_task

FunctionTimedOut 异常表示函数超时未完成

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

1

2

3

4

5

6

7

8

9

10

11

12

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!”)

这种方式适用于需要多次调用的函数,避免每次调用都手动设置超时。

func-timeout 本质上还是依赖 多线程 或 多进程 实现超时控制,在某些情况下可能不适用于主线程(如 Jupyter Notebook)。它也不能用于 main 线程内的 while True 死循环,因为 Python 的 GIL 可能会影响信号处理。

自定义进程

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

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

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!’)

运行结果如下:

1

2

3

4

5

6

7

8

9

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,它内置了超时支持。以下是示例代码:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

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)

运行结果如下:

1

2

3

4

5

6

7

8

9

10

11

12

13

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 秒后向处理器发送信号来引发异常。不过,这种方法相对底层且不够直观。

1

2

3

4

5

6

7

8

9

10

11

12

13

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)

总结

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

func-timeout 库:通过 func_timeout 或装饰器 func_set_timeout,可为函数设置超时时间,超时则抛出异常。适用于网络请求或计算密集型任务。

自定义进程:利用 multiprocessing 模块创建子进程执行函数,通过 join(timeout) 控制超时,超时后终止进程。

subprocess 模块:从 Python 3.5 起,subprocess.run() 支持超时参数,超时会抛出 TimeoutExpired 异常,适合外部命令调用。

信号机制:在 UNIX 系统中,使用 signal 模块设置超时信号,超时后触发异常,但实现较底层。

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

以上就是Python处理函数调用超时的四种方法的详细内容,更多关于Python函数调用超时的资料请关注脚本之家其它相关文章!


路过

雷人

握手

鲜花

鸡蛋
上一篇:edtd下一篇:f嘎嘎嘎嘎嘎嘎嘎嘎嘎嘎

Archiver|手机版|小黑屋|七贝PS网 ( 粤ICP备20019910号 )

GMT+8, 2025-8-3 07:30 , Processed in 0.436882 second(s), 20 queries .

Powered by Discuz! X3.4

Copyright © Tencent Cloud.

返回顶部