Python线程以精确的时间间隔执行

Posted 2023-02-25

【中文标题】Python线程以精确的时间间隔执行

【英文标题】：Python thread executing at precise time interval

【发布时间】：2017-11-27 10:50:52

【问题描述】：

我想创建一个以指定时间间隔执行的函数，假设每 30 秒执行一次，时间是这样的：12:37:00.000000、12:37:30.000000、12:38:00.000000

现在最简单的方法是创建一个线程来运行一个函数，该函数有一个 while 循环，最后有一个 sleep(30)，因此在最后等待 30 秒。我的应用程序将 24/7 运行，并且及时，假设从开始可能 2-3 小时，它将执行该功能的实际时间将是 X:X:00.5 或 X:X:33.5（因为有很多上面的代码sleep(30))，因此每 30 秒会稍微延迟我的工作，而在几个小时内，延迟会相当大。

我正在寻找一种优雅的设置方法，就像每 30 秒触发一次的警报一样，但总是恰好 30 秒，不多也不少，因为精度是必须的。

【问题讨论】：

您需要精确到什么程度？ 1/5 秒够吗？

就是这样，1/5 秒根本不算多，但如果我每 30 秒失去 1/5 秒，那么在一两天内它加起来就是也许超过一秒钟。我需要高精度，这意味着即使运行了一年，它仍然必须在 X:X:00.000000

处执行

【参考方案1】：

您可以推出自己的方法：

from datetime import datetime, timedelta
import time

def every_30s():
    print "fired at", datetime.now()


now = datetime.now()

# Align to the next 30 second event from the current time
if now.second >= 30:
    next_fire = now.replace(second=30, microsecond=0) + timedelta(seconds=30)
else:
    next_fire = now.replace(second=0, microsecond=0) + timedelta(seconds=30)

sleep = (next_fire - now).seconds - 2

while True:
    # Sleep for most of the time
    time.sleep(sleep)

    # Wait until the precise time is reached
    while datetime.now() < next_fire:
        pass

    every_30s()
    next_fire += timedelta(seconds=30)   # Advance 30 seconds
    sleep = 28

这首先与下一个 30 秒间隔对齐并触发事件。然后它会休眠 28 秒（以减少负载），然后快速检查下一个触发时间是否已过。然后下一次点火时间提前 30 秒。

这将确保时间永远不会漂移。如果机器负载过重，它可能会稍晚触发，但下一个事件将再次准时。

样本输出：

fired at 2017-11-27 11:41:00
fired at 2017-11-27 11:41:30
fired at 2017-11-27 11:42:00
fired at 2017-11-27 11:42:30
fired at 2017-11-27 11:43:00
fired at 2017-11-27 11:43:30
fired at 2017-11-27 11:44:00

【讨论】：

感谢您的回答！从一开始就非常精确，从不超过 1 毫秒！感人的！现在测试几天，会有一些结果。

因此在运行 13 小时后，最差的“延迟”为 0.001384 秒，这意味着超过一毫秒。再次感谢您的解决方案，它完美运行，正是我想要的！

【参考方案2】：

Python 可能不是完成这项工作的最佳工具，但如果您将sleep 等函数测量的时间解耦，并在每一步使用system.time 重新校准，您可能会得到准确的结果从长远来看，足以满足您的需求：

这使用time.sleep 来等待大部分时间间隔。在等待时间快结束时，它会更主动地检查经过的时间与系统时间的对比。

它会跟踪已过去的间隔数，并使用经过的总时间来最小化/纠正在多个间隔上堆叠间隔错误时发生的错误累积。

可能会发生一个事件偶尔会延迟触发（由于外部系统条件、负载等...），但会重新校准以下事件以按时触发。

import time

interval = 30
steps = 0

print("start period      periods        Error", flush=True)
print(time.time(), flush=True)

start = time.time()
while True:
    steps += 1

    time.sleep(interval - 0.1)

    while time.time() < (start + interval * steps):
        pass

    #execute your stuff
    print(time.time(), interval * steps, time.time() - (start + (interval * steps)), flush=True)  # prints the actual interval

输出：

start period      periods        Error
1511782735.662815
1511782765.664593 30 1.9073486328125e-06
1511782795.6645918 60 7.152557373046875e-07
1511782825.664594 90 3.814697265625e-06
1511782855.664593 120 1.9073486328125e-06
1511782885.664594 150 2.86102294921875e-06
1511782915.664593 180 1.9073486328125e-06
1511782945.664593 210 1.9073486328125e-06
1511782975.664594 240 2.86102294921875e-06
1511783005.664593 270 1.9073486328125e-06

误差代表从一开始的漂移，不是每个区间累积的。