【引言】
我们经常需要多函数进行耗时测试,测试方法有许多,这里介绍装饰者的方法,提高耗时测试代码的可复用性,在其他方面的应用也是如此。
【设计原则】
类应该对扩展开放,对修改关闭。
【代码】
(1)定义装饰者具体方法
#encoding: UTF-8 ''' Created on 2016��12��7�� @filename: test.py @author: YYH ''' import time from functools import wraps class TimeRecorder: def __init__(self, name="function"): print(name +"()"+ " Start...") print(name +"()"+ " Running...") self.name = name self.startTime = time.time() def __del__(self): print("{0}() Ended,Cost Time:{1} s".format(self.name, time.time() - self.startTime)) #使用装饰者测试函数运行时间,这里看起来像是“钩子”的方法,实际并不是的,借助于Python的装饰者,这个方法将发挥巨大的作用。 def fn_timer(function): @wraps(function) #解决打印函数名的bug def function_timer(*args, **kwargs): tR=TimeRecorder(function.__name__) #增加变量,由使得该对象的生命器存在整个函数 result = function(*args, **kwargs) return result return function_timer
(2)定义装饰者(该方法就就具有fn_timer的“能力”)
(3)使用装饰者
(4)结果
【实例二】增加指示函数正在执行的打印语句
一开始,我的 想法是通过对线程的销毁来实现这个功能,但是python提供的线程并不提供stop方法,需要自己实现,由自己来管理。(想想python这样做也是有道理的,毕竟,如果线程有一些资源没有释放就贝强制退出了,是十分危险的事情)
所有我使用最简单的方法,继承Threading.Thread; 然后在需要定时调用的函数
StoppableThread类中的
callrun()
判断是否需要stop,需要时就停止循环。
callrun会调用带有阻塞的打印函数print_running
然后使用装饰者模式,为某个function 添加定时打印running的功能。
具体实现可以参考代码
#encoding: UTF-8 ''' Created on 2016��12��7�� @filename: test.py @author: YYH @version: 1.1: 增加了打印函数参数(args 和 kwargs) ''' import time from functools import wraps from Crypto.SelfTest.Signature.test_pkcs1_15 import isStr from numba.types import none class TimeRecorder: def __init__(self,argstr=''): print(argstr+ " Start...") print(argstr+ " Running...") self.argstr = argstr self.startTime = time.time() def __del__(self): print("{0} Ended,Cost Time:{1} s".format(self.argstr, time.time() - self.startTime)) #使用装饰者测试函数运行时间 def fn_timer(function): @wraps(function) #解决打印函数名的bug def function_timer(*args, **kwargs): argStr = str(locals())#得到参数字符串 tR=TimeRecorder(argstr=argStr) #增加变量,由使得该对象的生命器存在整个函数 result = function(*args, **kwargs) return result return function_timer #定时执行的函数 import threading # http://stackoverflow.com/questions/18018033/how-to-stop-a-looping-thread-in-python #创建允许销毁的线程类 使用继承 class StoppableThread(threading.Thread): """Thread class with a stop() method. The thread itself has to check regularly for the stopped() condition.""" def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, verbose=None): threading.Thread.__init__(self, group, self.callrun, name, args, kwargs, verbose) self.setDaemon(True) self.callback = target self._stop = threading.Event() def stop(self): self._stop.set() def stopped(self): return self._stopEvt.isSet() def callrun(self,*args,**kwargs): while self._stop.isSet() == False: self.callback(*args,**kwargs) #使用装饰者添加定时打印信息 def print_running(funcname="function"): time.sleep(1) print("<"+funcname+" running..."+">") def run_indicate(function): @wraps(function) #解决打印函数名的bug def function_timer(*args, **kwargs): st=StoppableThread(target=print_running,kwargs={'funcname':function.__name__}) #增加变量,由使得该对象的生命器存在整个函数 argDict = locals()#function得到参数字符串 argStr = "" argStr = argStr+str(argDict['function'])+'; ' argStr = argStr+'args:'+str(argDict['args'])+'; ' argStr = argStr+'kwargs:'+str(argDict['kwargs'])+'; ' tR=TimeRecorder(argstr=argStr) #增加变量,由使得该对象的生命器存在整个函数 st.start() result = function(*args, **kwargs) st.stop() st.join(1) #等待线程结束 return result return function_timer
【使用方法】与上面类似
【运行结果示例】