Python 循环从开始时间开始 n 分钟后运行

Question

我正在尝试创建一个 while 循环，它将在 2 个时间对象之间迭代，while datetime.datetime.now().time() <= datetime.datetime.now() +relativedelta(hour=1):但每隔 n 分钟或秒间隔一次。因此，如果开始时间为凌晨 1:00，则下一次迭代应在凌晨 1:05 开始，其中 n 为 5 分钟。因此，迭代应该在开始时间的 5 分钟后开始，而不是从迭代结束时开始，使用 时就是这种情况sleep。您能否告知如何实现这一点？

一个可能的解决方案是从这里开始： 编写每 5 分钟执行一次的 python 脚本

import schedule 
import time 

def func():
    print("this is python")

schedule.every(5).minutes.do(func)
while True:
    schedule.run_pending()
    time.sleep(1)

这样，开始时间必须是凌晨 1 点。其次，如果程序需要运行 5 分钟 + 1，该怎么办？在这种情况下，6 分钟的间隔将不起作用。

Answer 1

我相信这可以被认为是面向对象的“规范”解决方案，它创建一个Thread子类实例，该实例将在每个单元重复调用指定的函数datetime.timedelta，直到取消。启动和运行多长时间不是类本身关心的细节，而是由使用类的代码来确定。

由于大部分操作发生在单独的线程中，因此如果需要，主线程可以同时执行其他操作。

import datetime
from threading import Thread, Event
import time
from typing import Callable


class TimedCalls(Thread):
    """Call function again every `interval` time duration after it's first run."""
    def __init__(self, func: Callable, interval: datetime.timedelta) -> None:
        super().__init__()
        self.func = func
        self.interval = interval
        self.stopped = Event()

    def cancel(self):
        self.stopped.set()

    def run(self):
        next_call = time.time()
        while not self.stopped.is_set():
            self.func()  # Target activity.
            next_call = next_call + self.interval
            # Block until beginning of next interval (unless canceled).
            self.stopped.wait(next_call - time.time())


def my_function():
    print(f"this is python: {time.strftime('%H:%M:%S', time.localtime())}")

# Start test a few secs from now.
start_time = datetime.datetime.now() + datetime.timedelta(seconds=5)
run_time = datetime.timedelta(minutes=2)  # How long to iterate function.
end_time = start_time + run_time

assert start_time > datetime.datetime.now(), 'Start time must be in future'

timed_calls = TimedCalls(my_function, 10)  # Thread to call function every 10 secs.

print(f'waiting until {start_time.strftime("%H:%M:%S")} to begin...')
wait_time = start_time - datetime.datetime.now()
time.sleep(wait_time.total_seconds())

print('starting')
timed_calls.start()  # Start thread.
while datetime.datetime.now() < end_time:
    time.sleep(1)  # Twiddle thumbs while waiting.
print('done')
timed_calls.cancel()

示例运行：

import datetime
from threading import Thread, Event
import time
from typing import Callable


class TimedCalls(Thread):
    """Call function again every `interval` time duration after it's first run."""
    def __init__(self, func: Callable, interval: datetime.timedelta) -> None:
        super().__init__()
        self.func = func
        self.interval = interval
        self.stopped = Event()

    def cancel(self):
        self.stopped.set()

    def run(self):
        next_call = time.time()
        while not self.stopped.is_set():
            self.func()  # Target activity.
            next_call = next_call + self.interval
            # Block until beginning of next interval (unless canceled).
            self.stopped.wait(next_call - time.time())


def my_function():
    print(f"this is python: {time.strftime('%H:%M:%S', time.localtime())}")

# Start test a few secs from now.
start_time = datetime.datetime.now() + datetime.timedelta(seconds=5)
run_time = datetime.timedelta(minutes=2)  # How long to iterate function.
end_time = start_time + run_time

assert start_time > datetime.datetime.now(), 'Start time must be in future'

timed_calls = TimedCalls(my_function, 10)  # Thread to call function every 10 secs.

print(f'waiting until {start_time.strftime("%H:%M:%S")} to begin...')
wait_time = start_time - datetime.datetime.now()
time.sleep(wait_time.total_seconds())

print('starting')
timed_calls.start()  # Start thread.
while datetime.datetime.now() < end_time:
    time.sleep(1)  # Twiddle thumbs while waiting.
print('done')
timed_calls.cancel()