python - 在线程中实时定期执行函数，每 N 秒

Question

我有一个线程类，其循环需要每秒执行 4 次。我知道我可以做类似的事情

do_stuff()
time.sleep(0.25)

但问题是这并没有考虑到它需要的时间do_stuff()。实际上，这需要是一个实时线程。有没有办法做到这一点？理想情况下，线程在不执行代码时仍会进入睡眠状态。

Answer 1

简单的解决方案

import threading

def work (): 
  threading.Timer(0.25, work).start ()
  print "stackoverflow"

work ()

---

以上将确保work以每秒四次的间隔运行，其背后的理论是它将“排队”对自身的调用，该调用将在未来0.25秒后运行，而无需等待发生.

正因为如此，它可以（几乎）完全不间断地工作，并且我们非常接近每秒执行 4 次该函数。

---

threading.Timer可以通过以下指向 python 文档的链接阅读更多信息：

• docs.python.org - 16.2.7。定时器对象

---

推荐]更高级/动态的解决方案

即使前面的函数按预期工作，您也可以创建一个辅助函数来帮助处理未来的定时事件。

对于这个例子来说，下面的内容就足够了，希望代码会自己说话——它并不像看起来那么先进。

当您可以实现自己的包装器以满足您的确切需求时，将此视为灵感。

import threading

def do_every (interval, worker_func, iterations = 0):
  if iterations != 1:
    threading.Timer (
      interval,
      do_every, [interval, worker_func, 0 if iterations == 0 else iterations-1]
    ).start ()

  worker_func ()

def print_hw ():
  print "hello world"

def print_so ():
  print "stackoverflow"


# call print_so every second, 5 times total
do_every (1, print_so, 5)

# call print_hw two times per second, forever
do_every (0.5, print_hw)

---

Answer 2

我对一个线程做了一些不同的方法，在一个while循环中循环。对我来说优点是：

• 只有一个线程，这里提到的其他解决方案为每个间隔启动和停止线程

• .stop()对间隔的更多控制，您可以使用方法 停止 IntervalTimer

代码：

from threading import Thread, Event

# StoppableThread is from user Dolphin, from http://stackoverflow.com/questions/5849484/how-to-exit-a-multithreaded-program
class StoppableThread(Thread):  

    def __init__(self):
        Thread.__init__(self)
        self.stop_event = Event()        

    def stop(self):
        if self.isAlive() == True:
            # set event to signal thread to terminate
            self.stop_event.set()
            # block calling thread until thread really has terminated
            self.join()

class IntervalTimer(StoppableThread):

    def __init__(self, interval, worker_func):
        super().__init__()
        self._interval = interval
        self._worker_func = worker_func

    def run(self):
        while not self.stop_event.is_set():
            self._worker_func()
            sleep(self._interval)

用法：

def hw():
    print("Hello World")

interval = IntervalTimer(1,hw)
interval.start()

sleep(10)

interval.stop()