Python - 函数超时异常处理

本文最后更新于:2021年12月31日 下午

Python程序运行中,可能会遇到各种超时异常的情况,那么处理这部分异常就是处理此类异常的直接需求,本文记录相关内容。

超时异常

  • 程序由于种种原因运行了异常多的时间,甚至死循环
  • 处理此类问题的思路有新建线程和使用 signal 两种思路
  • signal 对 Windows 支持很有限,在Linux下运行良好
  • 常用的工具包有:timeout-decoratorfunc_timeoutstopit
  • 解决问题的框架都是为需要计时的函数添加装饰器,在装饰器中使用线程或信号量技术控制运行时间

signal

python 自带的 信号量 可以作为计时装置参与超时异常检测,支持 LinuxWindows 支持不佳

示例代码

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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
# coding:utf8
import time
import signal


# 自定义超时异常
class TimeoutError(Exception):
def __init__(self, msg):
super(TimeoutError, self).__init__()
self.msg = msg


def time_out(interval, callback):
def decorator(func):
def handler(signum, frame):
raise TimeoutError("run func timeout")

def wrapper(*args, **kwargs):
try:
signal.signal(signal.SIGALRM, handler)
signal.alarm(interval) # interval秒后向进程发送SIGALRM信号
result = func(*args, **kwargs)
signal.alarm(0) # 函数在规定时间执行完后关闭alarm闹钟
return result
except TimeoutError as e:
callback(e)
return wrapper
return decorator


def timeout_callback(e):
print(e.msg)


@time_out(2, timeout_callback)
def task1():
print("task1 start")
time.sleep(3)
print("task1 end")


@time_out(2, timeout_callback)
def task2():
print("task2 start")
time.sleep(1)
print("task2 end")


if __name__ == "__main__":
task1()
task2()
  • Linux下 输出

    1
    2
    3
    4
    task1 start
    run func timeout
    task2 start
    task2 end

    超时的函数被叫停并抛出异常,没有超时的函数正常执行

  • Windows 下

    1
    2
    发生异常: AttributeError       (note: full exception trace is shown but execution is paused at: <module>)
    module 'signal' has no attribute 'SIGALRM'

    无法正常使用

timeout-decorator

一个处理超时的装饰器,只需要在你想要的函数前面加上这个装饰器,就可以设置超时时间,如果超过了容忍的超时时间,那么程序将抛异常。

默认工作原理为 signal,因此Linux支持更好Windows支持不佳

安装

1
pip install timeout-decorator

使用方法

  • 引入包

    1
    import timeout_decorator
  • 将装饰器装饰在需要控制时间的函数上,参数单位为秒

    1
    2
    3
    @timeout_decorator.timeout(5)
    def mytest():
    pass

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import timeout_decorator
import time

@timeout_decorator.timeout(5)
def mytest():
print("start")
for i in range(1, 10):
time.sleep(1)
print("() seconds have passed", format(i))

def main():
try:
mytest()
except Exception as e:
print(e)

if __name__ == '__main__':
main()
print('finish!')

  • Linux 下输出
1
2
3
4
5
6
7
8
python timeout.py 
start
() seconds have passed 1
() seconds have passed 2
() seconds have passed 3
() seconds have passed 4
'Timed Out'
finish!
  • Windows 下输出

    1
    2
    module 'signal' has no attribute 'SIGALRM'
    finish!

    表明使用了信号量,并且在Windows 下支持不好

    timeout函数参数定义

    1
    (seconds=None, use_signals=True, timeout_exception=TimeoutError, exception_message=None) -> (function) -> (*args, **kwargs) -> Any

    也就是可以将 use_signals 设置为 false,设置后在Windows 下仍然无法运行,Linux仍然运行正常。

func_timeout (推荐)

基于线程技术的函数工作计时器,可以很好地兼容 Linux, Windows

可以装饰类函数,可以在被装饰函数中动态设置超时时间

安装

1
pip install func_timeout

使用方法

  • 引入包

    1
    from func_timeout import func_set_timeout, FunctionTimedOut
  • 将装饰器装饰在需要控制时间的函数上,参数单位为秒,可以装饰类成员函数

    1
    2
    3
    @func_set_timeout(5)
    def mytest():
    pass
  • 需要说明的是,该装饰器产生的异常种类不会被 except Exception as e 捕捉, 需要捕捉包内的 FunctionTimedOut 异常作为超时异常

  • 装饰器的参数在编译过程中确定,如果需要作为参数传入可以按照如下步骤进行:

    1. 在装饰器参数中设置 allowOverride=True
    2. 在被装饰的函数中加入关键词参数 **kwargs
    3. 增加输入参数forceTimeout,以覆盖装饰器中的超时参数

示例代码

基础示例
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from func_timeout import func_set_timeout, FunctionTimedOut
import time

@func_set_timeout(5)
def mytest():
print("Start")
for i in range(1, 10):
print("%d seconds have passed" % i)
time.sleep(1)

if __name__ == '__main__':
try:
mytest()
except FunctionTimedOut as e:
print('mytest2:::', e)
print('finish test')

在 Windows 和 Linux 下输出相同:

1
2
3
4
5
6
7
Start
1 seconds have passed
2 seconds have passed
3 seconds have passed
mytest2::: Function mytest (args=()) (kwargs={}) timed out after 3.000000 seconds.

finish test
进阶示例

装饰类方法,同时动态配置计时时间,捕捉超时异常

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
from func_timeout import func_set_timeout, FunctionTimedOut
import time

class TimeoutTest:
def __init__(self, timeout):
self.timeout = timeout

@func_set_timeout(0, allowOverride=True)
def _mytest(self, **kwargs):
print("Start")
for i in range(1, 10):
print("%d seconds have passed" % i)
time.sleep(1)

def mytest(self):
self._mytest(forceTimeout=self.timeout)

if __name__ == '__main__':

test_obj = TimeoutTest(3)

try:
test_obj.mytest()
except FunctionTimedOut as e:
print('mytest2:::', e)
print('finish test')

在 Windows 和 Linux 下输出相同:

1
2
3
4
5
6
7
Start
1 seconds have passed
2 seconds have passed
3 seconds have passed
mytest2::: Function _mytest (args=(<__main__.TimeoutTest object at 0x7f5d9559f880>,)) (kwargs={}) timed out after 3.000000 seconds.

finish test

stopit

安装

1
pip install stopit

使用方法

  • 引入包

    1
    import stopit
  • 将装饰器装饰在需要控制时间的函数上,参数单位为秒,可以装饰类成员函数

    1
    2
    3
    @stopit.threading_timeoutable()
    def mytest():
    pass
  • 在被装饰的函数中输入参数 timeout 来控制时长,异常可以用 Exception 捕获

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import stopit
import time
import traceback


@stopit.threading_timeoutable()
def infinite_loop():
# As its name says...
try:
print("Start")
for i in range(1, 10):
print("%d seconds have passed" % i)
time.sleep(1)
except Exception as e:
traceback.print_exc()

if __name__ == '__main__':
infinite_loop(timeout=3)
print('finish test')

在 Windows 和 Linux 下输出相同:

1
2
3
4
5
6
7
8
9
10
Start
1 seconds have passed
2 seconds have passed
3 seconds have passed
Traceback (most recent call last):
File "timeout.py", line 13, in infinite_loop
time.sleep(1)
stopit.utils.TimeoutException
finish test

参考资料