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

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

作者: 逐风细雨 | 来源:发表于2020-01-10 10:17 被阅读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模块混用,谁让大家都是携程,一山不容二虎!!!

    相关文章

      网友评论

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

          本文链接:https://www.haomeiwen.com/subject/epcpactx.html