如何在python中创建一个在间隔函数调用上线程化的后台？

Posted 2023-02-23

【中文标题】如何在python中创建一个在间隔函数调用上线程化的后台？

【英文标题】：How to create a background threaded on interval function call in python?

【发布时间】：2015-06-28 10:40:01

【问题描述】：

我正在尝试实现在后台运行的心跳调用。如何创建一个每 30 秒的线程间隔调用，它调用以下函数：

self.mqConn.heartbeat_tick()

另外，我将如何停止此线程？

非常感谢。

【参考方案1】：

使用包含循环的线程

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

---

或者，您可以将计时器子类化，以便轻松取消：

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

【讨论】：

如果heartbeat_tick() 花费大量时间，则不会每 30 秒执行一次

可能很重要

是的，每次至少需要 30 秒。最多 30 秒比较难。

如果您的前台函数受 CPU 限制，GIL 会阻止后台线程执行吗？

@JeffWidman：受 CPU 限制和“不释放 GIL”并不总是一回事。是的，如果前台线程一次在 GIL 上保持一分钟，那么我想后台线程会被延迟。

感谢@Eric，我对 David Beazley 的讲座进行了更多阅读：GIL 并且更有意义。

【参考方案2】：

或者您可以在线程模块中使用 Timer 类：

from threading import Timer

def hello():
    print "hello, world"

t = Timer(30.0, hello)
t.start() # after 30 seconds, "hello, world" will be printed
t.cancel() # cancels execution, this only works before the 30 seconds is elapsed

这不会每 x 秒启动一次，而是会延迟线程在 x 秒内执行。但是你仍然可以把它放在一个循环中并使用 t.is_alive() 来查看它的状态。

【讨论】：

循环会去哪里？

这取决于用例？

@Eric，如果您对我的帖子不满意，请随时改进。

【参考方案3】：

对Eric 的回答的快速跟进：您不能在python 2 中继承Timer，因为它实际上是一个真正的类的轻量级函数包装器：_Timer。如果你这样做了，你会得到this post 中弹出的问题。

改用_Timer 修复它：

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

【讨论】：

如果_Timer 会受到版本的重大更改，而Timer 不会（因为前导前缀“隐私”），我不会感到惊讶。我在 3.7 中对 Timer 进行子类化没有问题，并且文档说它是 Thread 的子类。

@Far_Eggs 你可能是正确的 - 很好的评论。我应该明确指出这是针对 2.7 的。

【参考方案4】：

一种方法是使用circuits 应用程序框架，如下所示：

from circuits import Component, Event, Timer


class App(Component):

    def init(self, mqConn):
        self.mqConn = mqConn
        Timer(30, Event.create("heartbeat"), persist=True).register(self)

    def heartbeat(self):
        self.mqConn.heartbeat_tick()


App().run()

注意：我是电路的作者 :)

这只是一个基本的想法和结构——您需要对其进行调整以适合您的确切应用和要求！