我想在Python中每60秒重复执行一个函数(就像Objective C中的NSTimer或JS中的setTimeout)。这段代码将作为守护进程运行，有效地类似于使用cron每分钟调用python脚本，但不需要用户设置。
在这个关于用Python实现的cron的问题中，解决方案似乎只有效地使()休眠x秒。我不需要这么高级的功能，所以也许这样的东西可以工作
while True:
# Code executed here
time.sleep(60)
这段代码是否存在任何可预见的问题?
当前回答
如果漂移不是一个问题
import threading, time
def print_every_n_seconds(n=2):
while True:
print(time.ctime())
time.sleep(n)
thread = threading.Thread(target=print_every_n_seconds, daemon=True)
thread.start()
异步输出。
#Tue Oct 16 17:29:40 2018
#Tue Oct 16 17:29:42 2018
#Tue Oct 16 17:29:44 2018
如果正在运行的任务需要相当多的时间，那么间隔就变成2秒+任务时间，所以如果您需要精确的调度，那么这并不适合您。
注意daemon=True标志意味着这个线程不会阻止应用程序关闭。例如，在运行测试等待此头停止后，pytest将无限期挂起的问题。
其他回答
我认为这取决于你想做什么，你的问题没有详细说明。
对我来说，我想在一个已经多线程的进程中做一个昂贵的操作。所以我让leader流程检查时间，只有她做昂贵的操作(检查点深度学习模型)。为了做到这一点，我增加了计数器，以确保5秒、10秒、15秒过去，每5秒保存一次(或使用math.floor的模块化算术):
def print_every_5_seconds_have_passed_exit_eventually():
"""
https://stackoverflow.com/questions/3393612/run-certain-code-every-n-seconds
https://stackoverflow.com/questions/474528/what-is-the-best-way-to-repeatedly-execute-a-function-every-x-seconds
:return:
"""
opts = argparse.Namespace(start=time.time())
next_time_to_print = 0
while True:
current_time_passed = time.time() - opts.start
if current_time_passed >= next_time_to_print:
next_time_to_print += 5
print(f'worked and {current_time_passed=}')
print(f'{current_time_passed % 5=}')
print(f'{math.floor(current_time_passed % 5) == 0}')
starting __main__ at __init__
worked and current_time_passed=0.0001709461212158203
current_time_passed % 5=0.0001709461212158203
True
worked and current_time_passed=5.0
current_time_passed % 5=0.0
True
worked and current_time_passed=10.0
current_time_passed % 5=0.0
True
worked and current_time_passed=15.0
current_time_passed % 5=0.0
True
对我来说，检查if语句是我所需要的。在我已经复杂的多处理器多gpu代码中拥有线程，调度器并不是我想要添加的复杂性，如果我可以避免它，似乎我可以。检查worker id很容易确保只有一个进程在做这件事。
注意，我使用True print语句来确保模块化的算术技巧有效，因为检查确切的时间显然是行不通的!但令我惊喜的是，地板竟然起了作用。
下面是MestreLion对代码的更新，它可以避免随着时间的推移而漂移。
这里的RepeatedTimer类按照OP的请求每隔“间隔”秒调用给定函数;调度并不取决于函数执行的时间。我喜欢这个解决方案，因为它没有外部库依赖关系;这是纯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
示例用法(摘自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!
我认为更简单的方法是:
import time
def executeSomething():
#code here
time.sleep(60)
while True:
executeSomething()
这样，你的代码被执行，然后等待60秒，然后再次执行，等待，执行，等等…… 没有必要把事情复杂化:D
我使用Tkinter after()方法，它不会“窃取游戏”(就像之前提出的sched模块)，即它允许其他东西并行运行:
import Tkinter
def do_something1():
global n1
n1 += 1
if n1 == 6: # (Optional condition)
print "* do_something1() is done *"; return
# Do your stuff here
# ...
print "do_something1() "+str(n1)
tk.after(1000, do_something1)
def do_something2():
global n2
n2 += 1
if n2 == 6: # (Optional condition)
print "* do_something2() is done *"; return
# Do your stuff here
# ...
print "do_something2() "+str(n2)
tk.after(500, do_something2)
tk = Tkinter.Tk();
n1 = 0; n2 = 0
do_something1()
do_something2()
tk.mainloop()
Do_something1()和do_something2()可以以任意的间隔速度并行运行。在这里，第2个将以两倍的速度执行。还要注意，我使用了一个简单的计数器作为终止任一函数的条件。你可以使用任何你喜欢的条件，或者不使用，如果你想让一个函数运行到程序终止(例如一个时钟)。
如果漂移不是一个问题
import threading, time
def print_every_n_seconds(n=2):
while True:
print(time.ctime())
time.sleep(n)
thread = threading.Thread(target=print_every_n_seconds, daemon=True)
thread.start()
异步输出。
#Tue Oct 16 17:29:40 2018
#Tue Oct 16 17:29:42 2018
#Tue Oct 16 17:29:44 2018
如果正在运行的任务需要相当多的时间，那么间隔就变成2秒+任务时间，所以如果您需要精确的调度，那么这并不适合您。
注意daemon=True标志意味着这个线程不会阻止应用程序关闭。例如，在运行测试等待此头停止后，pytest将无限期挂起的问题。
