Python threading.timer – 每“n”秒重复一次函数

我在python计时器方面遇到困难,非常感谢一些build议或帮助:D

我不太了解线程是如何工作的,但我只想每0.5秒发一个函数,并能够启动和停止并重置计时器。

但是,我不断得到RuntimeError: threads can only be started once执行RuntimeError: threads can only be started once当我执行threading.timer.start()两次。 有没有解决这个问题? 我尝试在每次启动之前应用threading.timer.cancel()

伪代码:

 t=threading.timer(0.5,function) while True: t.cancel() t.start() 

最好的方法是启动一次定时器线程。 在你的计时器线程中,你可以编写以下代码

 class MyThread(Thread): def __init__(self, event): Thread.__init__(self) self.stopped = event def run(self): while not self.stopped.wait(0.5): print("my thread") # call a function 

在启动定时器的代码中,可以set停止的事件来停止定时器。

 stopFlag = Event() thread = MyThread(stopFlag) thread.start() # this will stop the timer stopFlag.set() 

使用计时器线程 –

 from threading import Timer,Thread,Event class perpetualTimer(): def __init__(self,t,hFunction): self.t=t self.hFunction = hFunction self.thread = Timer(self.t,self.handle_function) def handle_function(self): self.hFunction() self.thread = Timer(self.t,self.handle_function) self.thread.start() def start(self): self.thread.start() def cancel(self): self.thread.cancel() def printer(): print 'ipsem lorem' t = perpetualTimer(5,printer) t.start() 

这可以通过t.cancel()来停止

从python中的setInterval的等价 :

 import threading def setInterval(interval): def decorator(function): def wrapper(*args, **kwargs): stopped = threading.Event() def loop(): # executed in another thread while not stopped.wait(interval): # until stopped function(*args, **kwargs) t = threading.Thread(target=loop) t.daemon = True # stop if the program exits t.start() return stopped return wrapper return decorator 

用法:

 @setInterval(.5) def function(): "..." stop = function() # start timer, the first call is in .5 seconds stop.set() # stop the loop stop = function() # start new timer # ... stop.set() 

或者,这是相同的function,但作为一个独立的function,而不是装饰 :

 cancel_future_calls = call_repeatedly(60, print, "Hello, World") # ... cancel_future_calls() 

这是如何做到这一点,而不使用线程 。

为了提供一个正确的答案使用计时器作为OP请求,我会改进swapnil jariwala的答案 :

 from threading import Timer import time class InfiniteTimer(): """A Timer class that does not stop, unless you want it to.""" def __init__(self, seconds, target): self._should_continue = False self.is_running = False self.seconds = seconds self.target = target self.thread = None def _handle_target(self): self.is_running = True self.target() self.is_running = False self._start_timer() def _start_timer(self): if self._should_continue: # Code could have been running when cancel was called. self.thread = Timer(self.seconds, self._handle_target) self.thread.start() def start(self): if not self._should_continue and not self.is_running: self._should_continue = True self._start_timer() else: print("Timer already started or running, please wait if you're restarting.") def cancel(self): if self.thread is not None: self._should_continue = False # Just in case thread is running and cancel fails. self.thread.cancel() else: print("Timer never started or failed to initialize.") def tick(): print('ipsem lorem') # Example Usage t = InfiniteTimer(0.5, tick) t.start() 

导入时间是可选的,没有示例用法。

我已经实现了一个作为计时器的类。

我留下的链接在这里,以防万一需要它: https : //github.com/ivanhalencp/python/tree/master/xTimer

我已经改变了swapnil-jariwala代码中的一些代码来制作一个小控制台时钟。

 from threading import Timer, Thread, Event from datetime import datetime class perpetualTimer(): def __init__(self, t, hFunction): self.t = t self.hFunction = hFunction self.thread = Timer(self.t, self.handle_function) def handle_function(self): self.hFunction() self.thread = Timer(self.t, self.handle_function) self.thread.start() def start(self): self.thread.start() def cancel(self): self.thread.cancel() def printer(): tempo = datetime.today() print("{}:{}:{}".format(tempo.hour, tempo.minute, tempo.second)) t = perpetualTimer(1, printer) t.start() 

OUTPUT

 >>> 11:39:11 11:39:12 11:39:13 11:39:14 11:39:15 11:39:16 ... 

计时器与tkintergraphics界面

这段代码把时钟定时器放在tkinter的一个小窗口中

 from threading import Timer, Thread, Event from datetime import datetime import tkinter as tk app = tk.Tk() lab = tk.Label(app, text="Timer will start in a sec") lab.pack() class perpetualTimer(): def __init__(self, t, hFunction): self.t = t self.hFunction = hFunction self.thread = Timer(self.t, self.handle_function) def handle_function(self): self.hFunction() self.thread = Timer(self.t, self.handle_function) self.thread.start() def start(self): self.thread.start() def cancel(self): self.thread.cancel() def printer(): tempo = datetime.today() clock = "{}:{}:{}".format(tempo.hour, tempo.minute, tempo.second) try: lab['text'] = clock except RuntimeError: exit() t = perpetualTimer(1, printer) t.start() app.mainloop() 
 from threading import Timer def TaskManager(): #do stuff t = Timer( 1, TaskManager ); t.start() TaskManager(); 

这里是小样本,这将有助于了解它如何运行。 函数taskManager()在最后创build延迟函数调用它自己。

尝试改变“达赖”variables,你将能够看到不同

 from threading import Timer, _sleep # ------------------------------------------ DATA = []; dalay = 0.25; # sec counter = 0; allow_run = True; FIFO = True; def taskManager(): global counter, DATA, delay, allow_run; counter += 1; if len(DATA) > 0: if FIFO: print("["+str(counter)+"] new data: ["+str(DATA.pop(0))+"]"); else: print("["+str(counter)+"] new data: ["+str(DATA.pop())+"]"); else: print("["+str(counter)+"] no data"); if allow_run: #delayed method/function call to it self t = Timer( dalay, taskManager ); t.start(); else: print(" END task-manager: disabled"); # ------------------------------------------ def main(): DATA.append("data from main(): 0"); _sleep(2); DATA.append("data from main(): 1"); _sleep(2); # ------------------------------------------ print(" START task-manager:"); taskManager(); _sleep(2); DATA.append("first data"); _sleep(2); DATA.append("second data"); print(" START main():"); main(); print(" END main():"); _sleep(2); DATA.append("last data"); allow_run = False; 

我必须为一个项目做这个。 我最终做的是为这个函数启动一个单独的线程

 t = threading.Thread(target =heartbeat, args=(worker,)) t.start() 

****心跳是我的function,工作人员是我的一个论据****

在我的心跳function里面:

 def heartbeat(worker): while True: time.sleep(5) #all of my code 

所以当我启动线程时,函数会重复等待5秒钟,运行我所有的代码,并无限期地执行。 如果你想杀死进程就杀死线程。