Python 每 5 分钟执行一次

Question

我需要检查 API 上的数据。 API 每 5 分钟刷新一次新数据（10:00、10:05、10:10 等...）

我不想使用 time.sleep(300)，因为我希望我的脚本在 10:05:03，然后是 10:05:03 等执行某些操作，而不是脚本开始前 5 分钟（也许是10h12 开始 我该如何构建它？

谢谢大家。

Answer 1

更新：

只是想消除递归错误的可能性，所以我重写了代码：

from threading import Thread
from time import sleep
import datetime

def check_api():
    # ... your code here ...
    pass

def schedule_api():
    while datetime.datetime.now().minute % 5 != 0:
        sleep(1)
    check_api()
    while True:
        sleep(300)
        check_api()

thread = Thread(target=schedule_api)
thread.start()

此外，如果您希望线程在主程序退出时退出，您可以在线程上将 daemon 设置为 True，例如：

thread.daemon = True

但这并不强制终止此线程，因此您也可以尝试以下方法：

# ...
RUNNING = True
# ...
thread = Thread(target=schedule_api)
thread.start()
#...
def main():
    # ... all main code ...
    pass

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        RUNNING = False

---

您可以使用以下代码：

import threading

def check_api():
    pass

timer_thread = threading.Timer(300, check_api)
timer_thread.start()
# call timer_thread.cancel() when you need it to stop

这将每 5 分钟调用一次您的 check_api 函数，并且不会阻止您的主代码的执行。

正如@scotyy3785 所提到的，上面的代码只会运行一次，但我知道你想要什么并且已经为它编写了代码：

from threading import Thread
from time import sleep
import datetime


def check_api():
    # ... your code here ...
    pass


def caller(callback_func, first=True):
    if first:
        while not datetime.datetime.now().minute % 5 == 0:
            sleep(1)
    callback_func()
    sleep(300)
    caller(callback_func, False)


thread = Thread(target=caller, args=(check_api,))
thread.start()

# you'll have to handle the still running thread on exit

上面的代码会在00、05、10、15...等分钟调用check_api

Answer 2

定期循环检查时间，并在某些分钟标记处做某事：

import time

# returns the next 5 minute mark
# e.g. at minute 2 return 5
def get_next_time():
    minute = time.localtime().tm_min
    result = 5 - (minute % 5) + minute
    if result == 60:
        result = 0
    return result

next_run = get_next_time()

while True:
    now = time.localtime()
    # at minute 0, 5, 10... + 3 seconds:
    if next_run == now.tm_min and now.tm_sec >= 3:
        print("checking api")
        next_run = get_next_time()
    time.sleep(1)