Python - 函数超时异常处理

本文最后更新于：2024年5月7日下午

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

超时异常

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

signal

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

示例代码

# 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

示例代码

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 下输出

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，以覆盖装饰器中的超时参数

示例代码

基础示例

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 下输出相同：

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

进阶示例

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

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 下输出相同：

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 捕获

示例代码

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 下输出相同：

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