Wie kann man eine Funktion alle x Sekunden wiederholt ausführen?

Question 1

Ich möchte eine Funktion in Python alle 60 Sekunden für immer wiederholen (genau wie eine NStimer in Objective C oder setTimeout in JS). Dieser Code wird als Daemon ausgeführt und ist praktisch so, als ob das Python-Skript jede Minute mit einem Cron aufgerufen wird, ohne dass dies vom Benutzer eingerichtet werden muss.

In dieser Frage nach einem in Python implementierten Cron scheint die Lösung effektiv gerecht zu sein schlafen() für x Sekunden. Ich brauche solche erweiterten Funktionen nicht, also würde vielleicht so etwas funktionieren

while True:
    # Code executed here
    time.sleep(60)

Gibt es absehbare Probleme mit diesem Code?

Question 2

Wenn Ihr Programm noch keine Ereignisschleife hat, verwenden Sie die geplant -Modul, das einen Allzweck-Ereignisplaner implementiert.

import sched, time
s = sched.scheduler(time.time, time.sleep)
def do_something(sc): 
    print("Doing stuff...")
    # do your stuff
    sc.enter(60, 1, do_something, (sc,))

s.enter(60, 1, do_something, (s,))
s.run()

Wenn Sie bereits eine Ereignisschleifenbibliothek wie z asyncio, trio, tkinter, PyQt5, gobject, kivyund viele andere – planen Sie die Aufgabe stattdessen einfach mit den Methoden Ihrer vorhandenen Ereignisschleifenbibliothek.

Question 3

Verknüpfen Sie Ihre Zeitschleife wie folgt mit der Systemuhr:

import time
starttime = time.time()
while True:
    print("tick")
    time.sleep(60.0 - ((time.time() - starttime) % 60.0))

Question 4

Wenn Sie eine nicht blockierende Möglichkeit wünschen, Ihre Funktion regelmäßig auszuführen, würde ich anstelle einer blockierenden Endlosschleife einen Thread-Timer verwenden. Auf diese Weise kann Ihr Code weiter ausgeführt und andere Aufgaben ausgeführt werden, und Ihre Funktion wird trotzdem alle n Sekunden aufgerufen. Ich verwende diese Technik häufig zum Drucken von Fortschrittsinformationen bei langen, CPU-/Festplatten-/Netzwerk-intensiven Aufgaben.

Hier ist der Code, den ich in einer ähnlichen Frage gepostet habe, mit start()- und stop()-Steuerelement:

from threading import Timer

class RepeatedTimer(object):
    def __init__(self, interval, function, *args, **kwargs):
        self._timer     = None
        self.interval   = interval
        self.function   = function
        self.args       = args
        self.kwargs     = kwargs
        self.is_running = False
        self.start()

    def _run(self):
        self.is_running = False
        self.start()
        self.function(*self.args, **self.kwargs)

    def start(self):
        if not self.is_running:
            self._timer = Timer(self.interval, self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        self._timer.cancel()
        self.is_running = False

Verwendungszweck:

from time import sleep

def hello(name):
    print "Hello %s!" % name

print "starting..."
rt = RepeatedTimer(1, hello, "World") # it auto-starts, no need of rt.start()
try:
    sleep(5) # your long-running job goes here...
finally:
    rt.stop() # better in a try/finally block to make sure the program ends!

Merkmale:

Nur Standardbibliothek, keine externen Abhängigkeiten

start() und stop() können sicher mehrmals aufgerufen werden, auch wenn der Timer bereits gestartet/gestoppt wurde
Die aufzurufende Funktion kann Positionsargumente und benannte Argumente haben
Du kannst ändern interval jederzeit, es wird nach dem nächsten Lauf wirksam. Das gleiche für args, kwargs und sogar function!

Question 5

Vielleicht möchten Sie darüber nachdenken Verdrehte Dies ist eine Python-Netzwerkbibliothek, die die implementiert Reaktormuster.

from twisted.internet import task, reactor

timeout = 60.0 # Sixty seconds

def doWork():
    #do work here
    pass

l = task.LoopingCall(doWork)
l.start(timeout) # call every sixty seconds

reactor.run()

Während “while True: sleep(60)” wahrscheinlich funktionieren wird, implementiert Twisted wahrscheinlich bereits viele der Funktionen, die Sie irgendwann benötigen werden (Dämonisierung, Protokollierung oder Ausnahmebehandlung, wie von Bobince hervorgehoben) und wird wahrscheinlich eine robustere Lösung sein

Question 6

Hier ist eine Aktualisierung des Codes von MestreLion, die ein Driften im Laufe der Zeit vermeidet.

Die RepeatedTimer-Klasse ruft hier die angegebene Funktion alle “Intervall”-Sekunden auf, wie vom OP angefordert; Der Zeitplan hängt nicht davon ab, wie lange die Ausführung der Funktion dauert. Ich mag diese Lösung, da sie keine externen Bibliotheksabhängigkeiten hat; das ist nur reines Python.

import threading 
import time

class RepeatedTimer(object):
  def __init__(self, interval, function, *args, **kwargs):
    self._timer = None
    self.interval = interval
    self.function = function
    self.args = args
    self.kwargs = kwargs
    self.is_running = False
    self.next_call = time.time()
    self.start()

  def _run(self):
    self.is_running = False
    self.start()
    self.function(*self.args, **self.kwargs)

  def start(self):
    if not self.is_running:
      self.next_call += self.interval
      self._timer = threading.Timer(self.next_call - time.time(), self._run)
      self._timer.start()
      self.is_running = True

  def stop(self):
    self._timer.cancel()
    self.is_running = False

Beispielnutzung (kopiert aus der Antwort von MestreLion):

from time import sleep

def hello(name):
    print "Hello %s!" % name

print "starting..."
rt = RepeatedTimer(1, hello, "World") # it auto-starts, no need of rt.start()
try:
    sleep(5) # your long-running job goes here...
finally:
    rt.stop() # better in a try/finally block to make sure the program ends!

Question 7

import time, traceback

def every(delay, task):
  next_time = time.time() + delay
  while True:
    time.sleep(max(0, next_time - time.time()))
    try:
      task()
    except Exception:
      traceback.print_exc()
      # in production code you might want to have this instead of course:
      # logger.exception("Problem while executing repetitive task.")
    # skip tasks if we are behind schedule:
    next_time += (time.time() - next_time) // delay * delay + delay

def foo():
  print("foo", time.time())

every(5, foo)

Wenn Sie dies tun möchten, ohne Ihren verbleibenden Code zu blockieren, können Sie dies verwenden, um ihn in einem eigenen Thread ausführen zu lassen:

import threading
threading.Thread(target=lambda: every(5, foo)).start()

Diese Lösung kombiniert mehrere Funktionen, die selten in anderen Lösungen kombiniert zu finden sind:

Ausnahmebehandlung: Ausnahmen werden auf dieser Ebene so weit wie möglich ordnungsgemäß behandelt, dh sie werden zu Debugging-Zwecken protokolliert, ohne unser Programm abzubrechen.
Keine Verkettung: Die übliche kettenartige Implementierung (zur Planung des nächsten Ereignisses), die Sie in vielen Antworten finden, ist insofern spröde, als dass, wenn innerhalb des Planungsmechanismus etwas schief geht (threading.Timer oder was auch immer), dies wird die Kette beenden. Es finden dann keine weiteren Ausführungen statt, selbst wenn die Ursache des Problems bereits behoben ist. Eine einfache Schleife und wartet mit einer einfachen sleep() ist im Vergleich deutlich robuster.
Kein Abdriften: Meine Lösung verfolgt genau die Zeiten, zu denen sie ausgeführt werden soll. Es gibt keine Drift in Abhängigkeit von der Ausführungszeit (wie bei vielen anderen Lösungen).

Überspringen: Meine Lösung überspringt Aufgaben, wenn eine Ausführung zu lange gedauert hat (z. B. mache X alle fünf Sekunden, aber X hat 6 Sekunden gedauert). Dies ist das Standard-Cron-Verhalten (und das aus gutem Grund). Viele andere Lösungen führen die Aufgabe dann einfach ohne Verzögerung mehrfach hintereinander aus. Für die meisten Fälle (zB Aufräumarbeiten) ist dies nicht erwünscht. Wenn es ist gewünscht, einfach verwenden next_time += delay stattdessen.

Question 8

Der einfachere Weg ist meiner Meinung nach:

import time

def executeSomething():
    #code here
    time.sleep(60)

while True:
    executeSomething()

Auf diese Weise wird Ihr Code ausgeführt, dann wartet er 60 Sekunden, dann wird er erneut ausgeführt, wartet, führt aus, etc … Keine Notwendigkeit, die Dinge zu komplizieren: D