考虑到函数执行时间，如何在（相对）准确的时间按照 fequencey（例如，每 3 秒）执行线程？

Question

我想每 3 秒运行一次函数，并且我确实搜索过类似的主题。但是，我没有找到任何真正能满足我要求的解决方案，关键问题是在这些解决方案中，他们没有考虑函数本身的执行时间。考虑以下代码。

import datetime as dt
import time
import threading

def counting():
    global  num
    time_now = dt.datetime.now()
    if num > 0:
        print(f'count: {num}, time now: {time_now}') 
        num -= 1
        t = threading.Timer(3.0, counting)
        t.start()

num = 5
counting()

每 3.0 秒打印一次。主要问题是，在实际案例中，我将调用一个函数，例如 func1()，而不是 print(f'count: {num}, time now: {time_now}') ，这将需要 1 秒到 2.5 秒之间的时间。因此，两次通话之间的实际间隔时间将超过 3 秒（约 4-5.5 秒）。我怎样才能在两次调用之间每 3 秒准确地写一次（当然，允许非常小的错误）？谢谢！

Answer 1

这样做的方法是使用monotonic clock 找出当前时间，然后从您下次希望调用计划函数的时间中减去该时间；那么无论您的func1() 执行多长时间，您都会确切地知道您需要睡眠多长时间。这是一个示例（我删除了线程，因为无论它是在主线程还是某些子线程中运行，逻辑都是相同的）：

import random
import time

def func1():
    seconds_to_sleep = random.randrange(1000, 2500) / 1000.0
    print("pretending to work for %f seconds" % seconds_to_sleep)
    time.sleep(seconds_to_sleep)

def scheduled_function(scheduled_time, now):
    if (now > scheduled_time):
       print("scheduled_function was called %f seconds late" % (now-scheduled_time))
    else:
       print("scheduled_function was called %f seconds early" % (scheduled_time-now))

next_call_time = time.monotonic()  # schedule the first call to happen right away
while True:
    now = time.monotonic()
    time_until_call_time = next_call_time-now
    if (time_until_call_time > 0.0):
       time.sleep(time_until_call_time)  # wait until our next scheduled call-time
    scheduled_function(next_call_time, time.monotonic())
    func1()  # sleep for some unpredictable amount of time to simulate a workload
    next_call_time = next_call_time + 3.0  # each call should happen ~3 seconds after the previous call

Answer 2

使用 Ada 编程语言的解决方案非常简单。 Ada 提供了一种“延迟到”语法，允许延迟到未来某个时间。

with Ada.Text_IO; use Ada.Text_IO;
with Ada.Calendar; use Ada.Calendar;
with Ada.Calendar.Formatting; use Ada.Calendar.Formatting;

procedure Main is
   task periodic;
   
   task body periodic is
      Now : Time;
      Future : Time;
      The_Delay : constant duration := 3.0;
   begin
      for I in 1..10 loop
         Now := Clock;
         Put_Line("Periodic message at time " & Image(Now));
         Future := Now + The_Delay;
         delay 1.0;
         delay until Future;
      end loop;
   end periodic;
   
begin
   null;
end Main;

最近执行此程序会产生以下输出（时间是 UTC 时区）：

Periodic message at time 2021-05-27 02:26:46
Periodic message at time 2021-05-27 02:26:49
Periodic message at time 2021-05-27 02:26:52
Periodic message at time 2021-05-27 02:26:55
Periodic message at time 2021-05-27 02:26:58
Periodic message at time 2021-05-27 02:27:01
Periodic message at time 2021-05-27 02:27:04
Periodic message at time 2021-05-27 02:27:07
Periodic message at time 2021-05-27 02:27:10
Periodic message at time 2021-05-27 02:27:13

变量 The_Delay 是一个常数值，表示 3.0 秒。现在时间是每次迭代开始的时间。未来时间是现在加上 3.0 秒。只要执行不超过 3.0 秒，结果时间就不会被任务执行所抵消。为了模拟长时间的任务执行，任务在每次迭代期间延迟（休眠）1.0 秒。 “延迟1.0；”声明是绝对延迟，而“延迟到未来；”声明是一个相对延迟。

Answer 3

我想我自己得到了答案。我们需要使用schedule 模块。请参阅https://schedule.readthedocs.io/en/stable/examples.html中的示例

以下是我的测试代码。

import datetime as dt
import time
import threading
import schedule

def job():
    print("I'm running on thread %s" % threading.current_thread())
    print(dt.datetime.now())
    time.sleep(2)

def run_threaded(job_func):
    job_thread = threading.Thread(target=job_func)
    job_thread.start()

schedule.every(6).seconds.do(run_threaded, job)
schedule.every(6).seconds.do(run_threaded, job)
schedule.every(6).seconds.do(run_threaded, job)

t_end = dt.datetime.now() + dt.timedelta(seconds = 20)

while dt.datetime.now() < t_end:
    schedule.run_pending()

您可以看到我让它每 6 秒运行一次（通过应用三个多线程并行计算），我确实让它在函数 job() 中休眠 2 秒以替换实际运行时间。而且从输出结果中，你会看到它每 6 秒运行一次，而不是 8 秒！