【引言】
我们经常需要多函数进行耗时测试,测试方法有许多,这里介绍装饰者的方法,提高耗时测试代码的可复用性,在其他方面的应用也是如此。
【设计原则】
类应该对扩展开放,对修改关闭。
【代码】
(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://*.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
【使用方法】与上面类似
【运行结果示例】