如何每隔x秒重复执行一个函数?

488

我想在Python中每60秒永久重复执行一个函数(就像Objective C中的NSTimer或JS中的setTimeout)。这段代码将作为守护进程运行,并且实际上类似于使用cron每分钟调用Python脚本,但不需要用户设置。

关于Python实现cron的问题中,解决方案似乎只是使用sleep()睡眠x秒。我不需要这样高级的功能,因此可能类似下面的内容会起作用。

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

这段代码有没有可预见的问题?


171
一个学究式的观点,但可能很关键。你上面的代码并不是每60秒执行一次,而是在执行之间留下了60秒的间隔。只有当你执行的代码根本不花费时间时,才会每60秒执行一次。 - Simon
7
time.sleep(60) 也可能会比预期的早或晚返回。 - jfs
9
我还在思考:这段代码是否存在任何可预见的问题? - dwitvliet
9
“可预见的问题”是你不能单纯地使用time.sleep(60)来期望每小时进行60次迭代。因此,如果您每次迭代只添加一个项目并保持固定长度的列表...该列表的平均值将无法表示一致的“时间段”;因此,例如“移动平均”这样的函数可能会引用过时的数据点,从而扭曲您的指示。 - litepresence
14
@Banana,你需要注意,如果你的脚本不是每60秒准确执行一次,可能会出现问题。例如,我曾尝试分割视频流并将其上传,但由于媒体队列在循环内部处理数据时进行了缓冲,导致流变得比预期长5-10秒左右。这取决于你的数据。如果该函数只是一个简单的看门狗程序,比如在硬盘已满时发出警告,则使用这种方法不应该会有任何问题。但如果你正在检查核电站警报,那么结果可能是整个城市都被炸毁。 - DGoiko
显示剩余4条评论
23个回答

0
    ''' tracking number of times it prints'''
import threading

global timeInterval
count=0
def printit():
  threading.Timer(timeInterval, printit).start()
  print( "Hello, World!")
  global count
  count=count+1
  print(count)
printit

if __name__ == "__main__":
    timeInterval= int(input('Enter Time in Seconds:'))
    printit()

根据用户输入,它将在每个时间间隔迭代该方法。 - raviGupta

0

我认为这取决于你想做什么,而你的问题没有指定很多细节。

对我来说,我想在我的一个已经多线程处理的进程中执行一个昂贵的操作。因此,我让主进程检查时间,并只有她执行昂贵的操作(检查深度学习模型)。为了做到这一点,我增加计数器以确保每隔5秒钟(或使用math.floor的模算术)保存5秒钟:

def print_every_5_seconds_have_passed_exit_eventually():
    """
    https://dev59.com/FnA75IYBdhLWcg3wRWoU
    https://dev59.com/5XRB5IYBdhLWcg3w6LN2
    :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代码中添加线程和调度程序并不是我想要增加的复杂性,如果可以避免,那就最好了。检查工作进程ID很容易确保只有一个进程在执行此操作。

请注意,我使用True打印语句来确保模运算技巧有效,因为检查精确时间显然行不通!但令人惊喜的是,floor函数起到了奇效。


0
我在为FL Studio控制器编写脚本时偶然发现了这个。我甚至不会用Python编程(我是C++程序员)。但是顶部的答案都不清楚,所以我会补充一下。
使用游戏开发者的编程模式来控制时间。通常用于将游戏更新循环的某些部分锁定在特定的时间间隔内。例如,使物理引擎的更新与渲染无关。
基本思想是:
1. 计算当前时间与上一帧(或调用您的函数)的时间差。 2. 然后累积帧之间的时间差。 3. 如果累积的时间差超过目标时间间隔,则执行代码。 4. 您还可以选择性地响应并了解您的函数是否延迟。
CurrentTime = time.monotonic()
AccumDiff = 0

def Update():
    global CurrentTime
    global AccumDiff

    LastTime = CurrentTime            # Store previous frame time
    CurrentTime = time.monotonic()    # Get the current frame time
    TimeDiff = CurrentTime - LastTime # Calculate the difference
    AccumDiff += TimeDiff             # Accumulate the difference in time

    # Do something every second
    if AccumDiff >= 1.0: # A second passed!
        Overtime = AccumDiff - 1.0 # (optional) Respond to overtime
        print("This print is delayed by:", Overtime ,"Seconds!")  
        
        # Reset our accumulated time
        AccumDiff = 0
        # Do something every X seconds
        print("Printing............")
    else: # A second has not passed yet.
        # Continue accumulating time (pass or do something in response)
        print("Not gonna print yet.")
    

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接