本文主要是介绍python 限时装饰器 广泛应用于接口，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
python超时装饰器，应用与限时访问接口，比如15min之内下订单，一定时间访问网页，打开视频等等，上代码，用的是多线程监控的方式；
#_*_ coding:utf-8 _*_
import ctypes
import inspect
import time
from threading import Threaddef timelimited(timeout):''':function:函数执行超时器:param timeout: 限制时间:return: none'''def decorator(function):def decorator2(*args,**kwargs):class TimeoutException(Exception):passdef stop_thread(thread):def _async_raise(tid,exctype):'''raise the exception,performs cleanup if needed'''tid=ctypes.c_long(tid)if not inspect.isclass(exctype):exctype=type(exctype)res=ctypes.pythonapi.PyThreadState_SetAsyncExc(tid,ctypes.py_object(exctype))if res==0:raise ValueError("invaild thread id")elif res!=1:'''if it returns a number greater than one,you're in troubleand you should call it again with exc=NULL to revert the effect'''ctypes.pythonapi.PyThreadState_SetAsyncExc(tid,None)raise SystemError("PyThreadState_SetAsyncExc failed")_async_raise(thread.ident,SystemExit)class TimeLimited(Thread):def __init__(self,_error=None,):Thread.__init__(self)self._error=_errordef run(self):try:self.result=function(*args,**kwargs)except Exception as e:self._error=str(e)def stop(self):if self.isAlive():stop_thread(self)t=TimeLimited()#设置为守护线程t.setDaemon(True)t.start()#最大等待时间t.join(timeout)if isinstance(t._error,TimeoutException):t.stop()print('query timeout')return Falseif t.isAlive():t.stop()print("t is alive,query timeout")return Falseif t._error is None:return t.resultreturn decorator2return decorator@timelimited(3)
def testTime(t):time.sleep(t)print ("already execute it is")if __name__ == '__main__':testTime(5)C:\Python26\python.exe C:/Users/pwx357471/PycharmProjects/untitled/test.py
testTime(5)时候
t is alive,query timeoutProcess finished with exit code 0
C:\Python26\python.exe C:/Users/pwx357471/PycharmProjects/untitled/test.py
testTime(2)时候
already execute it isProcess finished with exit code 0这篇关于python 限时装饰器 广泛应用于接口的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
