我想在Python中每60秒重复执行一个函数(就像Objective C中的NSTimer或JS中的setTimeout)。这段代码将作为守护进程运行，有效地类似于使用cron每分钟调用python脚本，但不需要用户设置。
在这个关于用Python实现的cron的问题中，解决方案似乎只有效地使()休眠x秒。我不需要这么高级的功能，所以也许这样的东西可以工作
while True:
# Code executed here
time.sleep(60)
这段代码是否存在任何可预见的问题?
当前回答
我认为这取决于你想做什么，你的问题没有详细说明。
对我来说，我想在一个已经多线程的进程中做一个昂贵的操作。所以我让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语句来确保模块化的算术技巧有效，因为检查确切的时间显然是行不通的!但令我惊喜的是，地板竟然起了作用。
其他回答
你可能会考虑Twisted，它是一个实现了Reactor Pattern的Python网络库。
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()
虽然“While True: sleep(60)”可能会工作，Twisted可能已经实现了许多你最终需要的功能(如bobince指出的守护进程化、日志记录或异常处理)，并且可能是一个更健壮的解决方案
像这样将你的时间循环锁定到系统时钟上:
import time
starttime = time.time()
while True:
print("tick")
time.sleep(60.0 - ((time.time() - starttime) % 60.0))
我认为这取决于你想做什么，你的问题没有详细说明。
对我来说，我想在一个已经多线程的进程中做一个昂贵的操作。所以我让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语句来确保模块化的算术技巧有效，因为检查确切的时间显然是行不通的!但令我惊喜的是，地板竟然起了作用。
例如:显示当前本地时间
import datetime
import glib
import logger
def get_local_time():
current_time = datetime.datetime.now().strftime("%H:%M")
logger.info("get_local_time(): %s",current_time)
return str(current_time)
def display_local_time():
logger.info("Current time is: %s", get_local_time())
return True
# call every minute
glib.timeout_add(60*1000, display_local_time)
我用这个方法使每小时产生60个事件，其中大多数事件在整分钟后的相同秒数内发生:
import math
import time
import random
TICK = 60 # one minute tick size
TICK_TIMING = 59 # execute on 59th second of the tick
TICK_MINIMUM = 30 # minimum catch up tick size when lagging
def set_timing():
now = time.time()
elapsed = now - info['begin']
minutes = math.floor(elapsed/TICK)
tick_elapsed = now - info['completion_time']
if (info['tick']+1) > minutes:
wait = max(0,(TICK_TIMING-(time.time() % TICK)))
print ('standard wait: %.2f' % wait)
time.sleep(wait)
elif tick_elapsed < TICK_MINIMUM:
wait = TICK_MINIMUM-tick_elapsed
print ('minimum wait: %.2f' % wait)
time.sleep(wait)
else:
print ('skip set_timing(); no wait')
drift = ((time.time() - info['begin']) - info['tick']*TICK -
TICK_TIMING + info['begin']%TICK)
print ('drift: %.6f' % drift)
info['tick'] = 0
info['begin'] = time.time()
info['completion_time'] = info['begin'] - TICK
while 1:
set_timing()
print('hello world')
#random real world event
time.sleep(random.random()*TICK_MINIMUM)
info['tick'] += 1
info['completion_time'] = time.time()
根据实际情况，你可能会得到长度的刻度:
60,60,62,58,60,60,120,30,30,60,60,60,60,60...etc.
但在60分钟结束时，你会有60个滴答;而且它们中的大多数都将出现在您喜欢的正确偏移时间。
在我的系统中，我得到了< 1/20秒的典型漂移，直到需要纠正。
该方法的优点是具有较好的时钟漂移分辨率;这可能会导致问题，如果你做的事情，比如每tick追加一个项目，你希望每小时追加60个项目。未能考虑漂移可能导致次要指标，如移动平均线，将数据考虑得过于深入过去，从而导致错误的输出。
