pythonheartbeat

How to create a background threaded on interval function call in python?


I am trying to implement a heartbeat call that works in the background. How do I create a threaded on interval call of say every 30 seconds, which calls the following function:

self.mqConn.heartbeat_tick()

Also how would I stop this thread?

Many thanks.


Solution

  • Use a thread containing a loop

    from threading import Thread
    import time
    
    def background_task():
        while not background_task.cancelled:
            self.mqConn.heartbeat_tick()
            time.sleep(30)
    background_task.cancelled = False
    
    t = Thread(target=background_task)
    t.start()
    
    background_task.cancelled = True
    

    Alternatively, you could subclass timer, to make cancelling easy:

    from threading import Timer
    
    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(30.0, self.mqConn.heartbeat_tick)
    t.start() # every 30 seconds, call heartbeat_tick
    
    # later
    t.cancel() # cancels execution