如何在 Python 中实现看门狗定时器？答案

【问题标题】：How to implement a watchdog timer in Python?如何在 Python 中实现看门狗定时器？
【发布时间】：2013-04-22 13:46:17
【问题描述】：

我想用两个用例在 Python 中实现一个简单的看门狗定时器：

看门狗确保函数执行时间不超过x秒
看门狗确保某些定期执行的函数确实至少每 y 秒执行一次

我该怎么做？

【问题讨论】：

这里是WatchdogTimer implementation that creates only one thread

标签： python watchdog

【解决方案1】：

只是发布我自己的解决方案：

from threading import Timer

class Watchdog(Exception):
    def __init__(self, timeout, userHandler=None):  # timeout in seconds
        self.timeout = timeout
        self.handler = userHandler if userHandler is not None else self.defaultHandler
        self.timer = Timer(self.timeout, self.handler)
        self.timer.start()

    def reset(self):
        self.timer.cancel()
        self.timer = Timer(self.timeout, self.handler)
        self.timer.start()

    def stop(self):
        self.timer.cancel()

    def defaultHandler(self):
        raise self

如果您想确保函数在不到x 秒内完成，请使用：

watchdog = Watchdog(x)
try:
  # do something that might take too long
except Watchdog:
  # handle watchdog error
watchdog.stop()

如果您定期执行某事并希望确保它至少每y 秒执行一次：

import sys

def myHandler():
  print "Whoa! Watchdog expired. Holy heavens!"
  sys.exit()

watchdog = Watchdog(y, myHandler)

def doSomethingRegularly():
  # make sure you do not return in here or call watchdog.reset() before returning
  watchdog.reset()

【讨论】：

听说过 PEP8 吗？ Timer() 这里是什么？
不，还没有听说过。我是初学者。但是，一旦我有更多时间，我会检查...
我不确定您的 reset 方法。在您的__init__ 中，您的Timer 开关基于是否提供userHandler（应该是is not None BTW），在这种情况下使用它。在reset - 它总是self.handler 然后将被使用...
我认为__init__函数和reset函数需要调用self.timer.start()。
这个Watchdog在一个单独的线程（Timer线程）中引发Exception（本身），所以使用示例中的try/except是没用的。 Watchdog 也不会中断长操作，这将是看门狗的重点。

【解决方案2】：

signal.alarm() 为您的程序设置超时，您可以在主循环中调用它，并将其设置为您准备容忍的两次中的较大者：

import signal
while True:
    signal.alarm(10)
    infloop()

【讨论】：

假设解释器运行在实现 POSIX 信号的平台上。

【解决方案3】：

这是我在没有类的应用程序中使用的 wdt。没有办法阻止它：

from threading import Event, Thread

def wdt(time, callback):
    # a reset flag
    reset_e = Event()
    # a function to reset the wdt
    def reset(): reset_e.set()
    # the function to run in a differen thread
    def checker():
        # check if reset flag is set.
        # wait for specified time to give chance to reset.
        while reset_e.wait(time):
            # it was set in time. clear and wait again
            reset_e.clear()
        # time run out.
        callback()
    # the event is not set by default. Set it
    reset()
    # create and start the wdt
    t = Thread(target=checker)
    t.start()
    # return the resetter
    return reset


# The callback to run if wdt is not reset
def bark():
    print('woof')       

# Test
import time
reset = wdt(1.0, bark)
time.sleep(0.9)
reset()
time.sleep(2.0)

【讨论】：