Python - 函数超时异常处理

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

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

超时异常

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

signal

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

示例代码

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

参考资料