python3 限制函数调用时长的方法,并实现一种基于geven

2020-01-10  本文已影响0人  逐风细雨

在调用函数的时候,我们经常有用需要从业务层面限制函数或某个方法调用的需求。如果业务只是记录一下这个函数调用消耗的多少时间,那很简单,调用一下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模块混用,谁让大家都是携程,一山不容二虎!!!

上一篇 下一篇

猜你喜欢

热点阅读