0
点赞
收藏
分享

微信扫一扫

python3 程序定时器执行(可循环),最佳有效方案

一世独秀 2022-03-12 阅读 65


前言

昨天遇到一个需求,定时让程序运行,查询某内容,找到了python3定时器的用法,

有一些精致用法在网上看到很经典!

特此记录!

python2和python3分别 如何写定时器

python2和3 如何写一个定时器,循环定时做某一操作呢?

下面是py2

from threading import Timer
def hello():
print "hello, world"

t = Timer(10.0, hello)
t.start()

10秒后输出:

hello, world

重点研究 ​t = Timer(10.0, hello)​ 这句代码,python 提供了一个Timer 对象,它会在指定的时间后执行某一操作;

它的完整形式:

class threading.Timer(interval, function, args=[], kwargs={})

interval​ 是时间间隔,​function​ 是可调用的对象,​args​ 和 ​kwargs​ 会作为 function 的参数。

注意:这里只会执行一次 function,而不会一直定时执行,且 Timer 在执行操作的时候会创建一个新的线程。

Timer​ 在 ​python2​ 和 ​python3​ 有点区别:

# python2.7
def Timer(*args, **kwargs):
return _Timer(*args, **kwargs)
# python3.7
class Timer(Thread):
pass

在 python3 ,​Timer​ 是 ​Thread​ 的子类;在 python2,​_Timer​ 是 ​Thread​ 的子类,而 ​Timer​ 只是 ​_Timer​ 类的工厂方法。

上面的代码只会打印一次 hello, world 后退出,那么如何循环间隔打印呢?

粗陋的循环定时器

一种方法是在 ​function​ 里继续注册一个 ​Timer​,这样就可以在下一个 ​interval​ 继续执行 ​function​;

from threading import Timer
def hello():
print "hello, world"
Timer(10.0, hello) .start()

t = Timer(10.0, hello)
t.start()

每隔 10 秒输出一个 ​hello, world。

达到效果了,但是这里面好像有点问题。回到 ​Timer​ 本身,它是一个 ​thread​,每次循环间隔操作,系统都要创建一个线程,然后再回收,这对系统来说开销很大。如果时间间隔 ​interval​ 很短,系统会一下子创建很多线程,这些线程很难快速回收,​导致系统内存和cpu资源被消耗掉​。 所以不提倡在 function 里继续注册一个 ​Timer​。

更 ​pythonic​ 循环定时器

python2:

from threading import _Timer
def hello():
print "hello, world"
class RepeatingTimer(_Timer):
def run(self):
while not self.finished.is_set():
self.function(*self.args, **self.kwargs)
self.finished.wait(self.interval)
t = RepeatingTimer(10.0, hello)
t.start()

重点研究 ​RepeatingTimer​ 类,它继承了 ​threading._Timer​,但是重写了父类的 ​run​ 方法。这是 ​Python2​ 的写法,​python3​ 中 ​RepeatingTimer​ 应该继承 ​threading.Timer​。

python3:

from threading import Timer
def hello():
print "hello, world"
class RepeatingTimer(Timer):
def run(self):
while not self.finished.is_set():
self.function(*self.args, **self.kwargs)
self.finished.wait(self.interval)
t = RepeatingTimer(10.0, hello)
t.start()

为什么要重写 Thread 的 run 方法?

_Timer 是一个 Thread 子类,我们先看看 Thread 类的 run 用法。

from threading import Thread
def hello():
print "hello, world"
# 继承 Thread
class MyThread(Thread):
# 把要执行的代码写到run函数里面 线程在创建后会直接运行run函数
def run(self):
hello()
t = MyThread()
t.start()

Thread​ 对象的完整定义:

class threading.Thread(group=None, target=None, name=None, args=(), kwargs={})

其中 ​run​ 方法代码:

class Thread(_Verbose):
def run(self):
try:
if self.__target:
self.__target(*self.__args, **self.__kwargs)
finally:
# Avoid a refcycle if the thread is running a function with
# an argument that has a member that points to the thread.
del self.__target, self.__args, self.__kwargs

标准的 ​run​ 方法用于执行用户传入构造函数的 ​target​ 方法。 子类可以重写 run 方法,把要执行的代码写到 run 里面,线程在创建后,用户调用 start() 方法会运行 run() 方法。

所以 RepeatingTimer 重写 _Timer 的 run() 方法,可以改变线程的执行体,当我们调用 RepeatingTimer 的 start() 方法时会执行我们重写的 run() 方法。

再看看 RepeatingTimer 类中的 ​while not self.finished.is_set()​ 语句,​self.finished.is_set()​ 直到 ​True​ 才会退出循环,定时器才结束。​finished​ 是 threading.Event 对象。一个 Event 对象管理着一个 flag 标志,它能被 set() 方法设置为 True,也能被 clear() 方法设置为 False,调用 wait([timeout]) 线程会一直 sleep 到 flag 为 True 或超时时间到达。

我们知道定时器有一个 cancel() 方法可以提前取消操作。它其实是调用 Event.clear() 方法提前让 wait 方法结束等待,并且判断在 flag 为 true 的情况下不执行定时器操作。具体的代码:

class _Timer(Thread):
"""Call a function after a specified number of seconds:
t = Timer(30.0, f, args=[], kwargs={})
t.start()
t.cancel() # stop the timer's action if it's still waiting
"""

def __init__(self, interval, function, args=[], kwargs={}):
Thread.__init__(self)
self.interval = interval
self.function = function
self.args = args
self.kwargs = kwargs
self.finished = Event()

def cancel(self):
"""Stop the timer if it hasn't finished yet"""
self.finished.set()

def run(self):
self.finished.wait(self.interval)
if not self.finished.is_set():
self.function(*self.args, **self.kwargs)
self.finished.set()

所以 ​RepeatingTimer​ 的 run 方法会一直执行 while 循环体,在循环体了会执行用户传入的 function 对象,并等待指定的时间。当用户想退出定时器时,只需要调用 ​cancel​ 方法,将 flag 置为 ​True​ 便不会继续执行循环体了。这样便完成了一个还不错的循环定时器。

总结​:对于​python3​,可以照着下面的模板来,把​func​定义改一下即可 :

from threading import Timer
def func():
pass
class RepeatingTimer(Timer):
def run(self):
while not self.finished.is_set():
self.function(*self.args, **self.kwargs)
self.finished.wait(self.interval)
t = RepeatingTimer(10.0,func)
t.start()



举报

相关推荐

0 条评论