python3 限制函数调用时长的方法,并实现一种基于geven
在调用函数的时候,我们经常有用需要从业务层面限制函数或某个方法调用的需求。如果业务只是记录一下这个函数调用消耗的多少时间,那很简单,调用一下time模块用结束时间减去开始时间就可以了,代码如下,如下图:
import time
def func():
print("Do something")
start_time = time.time()
func()
cost_time = time.time() - start_time
print(f"消耗的时间为{cost_time}")
我们如果需要根据函数返回的结果,决定是否结束等待,并设置等待的最大时间。有时对于一些可以快速响应的同步请求,我们采取循环的方式来限制函数的运行时间,比如下面这样:
import time
import random
def func():
print("Do something")
time.sleep(1)
return random.randint(1,10)
start_time = time.time()
cost_time = 10 #运行时间
while time.time() - start_time < 10:
result = func()
if result == 1: #假设我们判断返回值为1时认为业务完成
break
else:
result = False # 在10s加函数运行时间的内未获取到结果,退出循环,结果设为False
获取到指定业务结果的最大的运行时间为10+1 ,1表示函数本身的运行时间。这种方式,如果函数本身的调用时间很长(IO阻塞或者依赖的外部接口阻塞),那这种限制方式的时长取决于阻塞的时长,如果一直阻塞,那这种方式也会一直阻塞。是否有其它的方式来解决在这种问题呢?答案是肯定的,而且有很多,各有优劣。
1.用信号量signal。有其他大神总结的代码,本人就不赘述了,介绍一个采样类似原理的轮子,timeout-decorator;
安装方式 pip install timeout-decorator
官方例子:
import time
import timeout_decorator
@timeout_decorator.timeout(5)
def mytest():
print("Start")
for i in range(1,10):
time.sleep(1)
print("{} seconds have passed".format(i))
if __name__ == '__main__':
mytest()
此方法很简单也很优秀,但因为signal模块的原因,目前只对Unix系统提供完美支持,无法在windos下面的应用
2.运用多个进程,线程,携程的方式。开篇所描写的困境就是在单个线程类的窘境,类似于人不可能通过,拽自己头发的方式把自己拧起来,但另一个抠脚大汉却可以轻松做到。因此此类方式基本的思路是,有至少两个执行单元(进程,线程,携程),一个单元用于执行用具体的业务,另一个单元计时,当超时后强制的结束执行业务的单元。因为python语言的特性,强制kill进程,线程,携程,都存在一定风险。因此需谨慎!!!
给个链接,这位大佬总计的不错,大家可以看一下:https://blog.csdn.net/Homewm/article/details/92127567
但个人更倾向于,用基于gevent的装饰器来实现类似的功能,使用更简单和优雅。下面是本人的实现的限制函数调用时长的装饰器,拿走不谢:
# coding:utf8
import time
import gevent
from gevent import monkey
monkey.patch_all()
import traceback
from functools import wraps
def time_limit(interval):
"""
:param interval:超时间隔
:return:运行超时抛出RuntimeError
"""
def wrap(func):
def time_out(interval):
gevent.sleep(interval)
raise RuntimeError(f"Callable object[{func.__name__}] timed out")
@wraps(func)
def deco(*args, **kwargs):
reulst_list = [None]
def hander(*args, **kwargs):
try:
result = func(*args, **kwargs)
except Exception as e:
print(traceback.format_exc())
return
reulst_list.append(result)
Timer = gevent.spawn(time_out, interval)
job = gevent.spawn(hander, *args, **kwargs)
Timer.start()
job.start()
# Timer或者job进程运行完成时退出
while not (Timer.ready() or job.ready()):
time.sleep(0.5)
gevent.killall([Timer, job])
return reulst_list[-1]
return deco
return wrap
自测在装饰async def 定义的函数时,在多线程的时候会与全局的loop事件冲突,建议不要与asyncio模块混用,谁让大家都是携程,一山不容二虎!!!