logo标签列表

线程定时器 - 每隔'n'秒重复执行函数

pythonmultithreadingtimerpython-multithreading
126
我希望每隔0.5秒就可以触发一个函数，并且能够启动、停止和重置计时器。我不太了解Python线程是如何工作的，而且在使用Python定时器时遇到了困难。
然而，当我执行“threading.timer.start()”两次时，我一直收到“RuntimeError: threads can only be started once”的错误提示。是否有解决方法？我尝试在每次开始之前应用“threading.timer.cancel()”。
伪代码：
t=threading.timer(0.5,function)
while True:
    t.cancel()
    t.start()
16个回答
1

为了提供一个正确的答案，按照OP的要求使用Timer，我会改进https://dev59.com/zGcs5IYBdhLWcg3w-Yr4#41450617

(Note: I have translated the text into simplified Chinese)
from threading import Timer


class InfiniteTimer():
  """A Timer class that does not stop, unless you want it to."""

  def __init__(self, seconds, target, args=[], kwargs=dict()):
    self._should_continue = False
    self.is_running = False
    self.seconds = seconds
    self.target = target
    self.args = args
    self.kwargs = kwargs
    self.thread = None

  def _handle_target(self):
    self.is_running = True
    self.target(*self.args, **self.kwargs)
    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(i):
  print('ipsem lorem', i)


# Example Usage
t = InfiniteTimer(0.5, tick, kwargs=dict(i="i"))
t.start()
1

这是一个使用函数而非类的替代实现，受到 @Andrew Wilkins 的启发。

因为等待比休眠更准确（它考虑了函数运行时间）：

import threading

PING_ON = threading.Event()

def ping():
  while not PING_ON.wait(1):
    print("my thread %s" % str(threading.current_thread().ident))

t = threading.Thread(target=ping)
t.start()

sleep(5)
PING_ON.set()
1

我喜欢right2clicky的回答，特别是它不需要每次定时器滴答声时拆除线程并创建一个新线程。此外，它是一个简单的覆盖，可以创建一个具有定时器回调的类，该回调会定期调用。这是我的常规用例：

class MyClass(RepeatTimer):
    def __init__(self, period):
        super().__init__(period, self.on_timer)

    def on_timer(self):
        print("Tick")


if __name__ == "__main__":
    mc = MyClass(1)
    mc.start()
    time.sleep(5)
    mc.cancel()
1
4
尽管这个回答理论上可以回答问题，但最好在这里包含回答的关键部分并提供链接作为参考。 - Kalle Richter
1

我用单例模式提出了另一个解决方案。请告诉我是否存在任何内存泄漏。

import time,threading

class Singleton:
  __instance = None
  sleepTime = 1
  executeThread = False

  def __init__(self):
     if Singleton.__instance != None:
        raise Exception("This class is a singleton!")
     else:
        Singleton.__instance = self

  @staticmethod
  def getInstance():
     if Singleton.__instance == None:
        Singleton()
     return Singleton.__instance


  def startThread(self):
     self.executeThread = True
     self.threadNew = threading.Thread(target=self.foo_target)
     self.threadNew.start()
     print('doing other things...')


  def stopThread(self):
     print("Killing Thread ")
     self.executeThread = False
     self.threadNew.join()
     print(self.threadNew)


  def foo(self):
     print("Hello in " + str(self.sleepTime) + " seconds")


  def foo_target(self):
     while self.executeThread:
        self.foo()
        print(self.threadNew)
        time.sleep(self.sleepTime)

        if not self.executeThread:
           break


sClass = Singleton()
sClass.startThread()
time.sleep(5)
sClass.getInstance().stopThread()

sClass.getInstance().sleepTime = 2
sClass.startThread()
0

这是连续运行计时器的示例代码。在计时器耗尽后，只需创建一个新的计时器并调用相同的函数即可。虽然不是最好的方法，但也可以这样做。

import threading
import time


class ContinousTimer():
    def __init__(self):
        self.timer = None

    def run(self, msg='abc'):
        print(msg)

        self.timer = threading.Timer(interval=2, function=self.run, args=(msg, ))
        self.timer.start()


if __name__ == "__main__":
    t = ContinousTimer()
    try:
        t.run(msg="Hello")
        while True:
            time.sleep(0.1)
    except KeyboardInterrupt:
        # Cancel Timer
        t.timer.cancel()
网页内容由stack overflow 提供, 点击上面的
可以查看英文原文，
原文链接