改进现有 setInterval 实现
我在尝试找出如何在Python中创建一个可以取消的setInterval,而不需要专门写一个新的类。现在我已经找到了方法,但我在想是否还有更好的办法。
下面的代码看起来运行得不错,但我还没有彻底测试过。
import threading
def setInterval(func, sec):
def inner():
while function.isAlive():
func()
time.sleep(sec)
function = type("setInterval", (), {}) # not really a function I guess
function.isAlive = lambda: function.vars["isAlive"]
function.vars = {"isAlive": True}
function.cancel = lambda: function.vars.update({"isAlive": False})
thread = threading.Timer(sec, inner)
thread.setDaemon(True)
thread.start()
return function
interval = setInterval(lambda: print("Hello, World"), 60) # will print Hello, World every 60 seconds
# 3 minutes later
interval.cancel() # it will stop printing Hello, World
有没有办法做到上面提到的,而不需要创建一个专门继承自threading.Thread
的类,或者使用type("setInterval", (), {})
?还是说我只能在创建专门的类和继续使用type
之间做选择?
1 个回答
30
要定期调用一个函数,每次调用之间间隔 interval
秒,并且可以取消未来的调用:
from threading import Event, Thread
def call_repeatedly(interval, func, *args):
stopped = Event()
def loop():
while not stopped.wait(interval): # the first call is in `interval` secs
func(*args)
Thread(target=loop).start()
return stopped.set
示例:
cancel_future_calls = call_repeatedly(60, print, "Hello, World")
# ...
cancel_future_calls()
注意:这个版本在每次调用后都会等待大约 interval
秒,无论 func(*args)
执行需要多长时间。如果想要像节拍器一样的精准时间间隔,可以用 timer()
来控制执行时间:可以把 stopped.wait(interval)
替换成 stopped.wait(interval - timer() % interval)
,其中 timer()
用来定义当前的时间(可以是相对时间),单位是秒,比如 time.time()
。更多信息可以查看 在Python中每隔x秒重复执行一个函数的最佳方法是什么?