我正在尝试为特定功能实现超时。我检查了SE中的许多问题,但找不到任何适合我的问题的解决方案,因为:
- 我在 Windows 中运行 python
- 超时应用于我无法控制的Python函数,即它是在已经设计的模块中定义的。
- python函数不是子进程
我有一个已经设计好的自定义模块(比如说MyModule
)是为特定任务而开发的,并且其中定义了函数。其中一个函数(比如MyFunc
)由于外部因素有永远运行的趋势,我只是不希望 python 脚本挂起。
我计划添加一个超时功能,如下伪代码所示:
import MyModule
set_timeout(T)
MyResult=MyModule.MyFunc()
#Come to this part of script after execution of MyFunc() or after T seconds (the latter on priority)
if Timeout occurred:
print 'MyFunc did not execute completely'
else:
print 'MyFunc completed'
但我不确定哪个模块可以用来在 python 上实现这一点。请注意,我是一个新手,我编写的所有脚本都是直接基于 SE Answers 或 Python Documentation。
我认为解决这个问题的一个好方法是创建一个装饰器并使用Thread.join(timeout=seconds) https://docs.python.org/3/library/threading.html#threading.Thread.join方法。请记住,没有很好的方法来终止线程,因此只要您的程序正在运行,它就会或多或少地继续在后台运行。
首先,创建一个像这样的装饰器:
from threading import Thread
import functools
def timeout(timeout):
def deco(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
res = [Exception('function [%s] timeout [%s seconds] exceeded!' % (func.__name__, timeout))]
def newFunc():
try:
res[0] = func(*args, **kwargs)
except Exception as e:
res[0] = e
t = Thread(target=newFunc)
t.daemon = True
try:
t.start()
t.join(timeout)
except Exception as je:
print ('error starting thread')
raise je
ret = res[0]
if isinstance(ret, BaseException):
raise ret
return ret
return wrapper
return deco
然后,做这样的事情:
func = timeout(timeout=16)(MyModule.MyFunc)
try:
func()
except:
pass #handle errors here
您可以在任何需要的地方使用此装饰器,例如:
@timeout(60)
def f():
...
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)