如何精确每分钟调用Python函数请求REST API？求最优实现方案

问题描述 #

我希望在每分钟整分时刻向API发送查询请求，因为该API每分钟更新数据，我想立即获取更新后的数据。计时精度要求极高，最终需实现持续运行。以下是我初步的实现思路：

import time, sched
time = 1549667056000 # starting timestamp
def get_data():
    # function to get some data from the API via requests
    #gets the Data
while true:
    s.scheduler(time)
    s.run(get_data()) # gets the data on the specified time(stamp)
    time = time + 60000 # adds 1 minute to the timestamp

请问这种实现方式是否可行？有没有更优的方法来精确每分钟从REST API获取数据？

回答 #

先说结论：你这个初步实现有几个明显的问题，实际跑起来大概率达不到你要的高精度定时效果，咱们一步步拆解问题，再给你几个更靠谱的方案。

先聊聊你现有代码的问题

• sched模块用法完全错误：sched.scheduler需要先初始化调度器实例，而且run()方法是启动调度器等待任务触发，不是直接传入函数调用。你写的s.run(get_data())会先执行get_data()，再把返回值传给run()，完全偏离了定时的初衷。
• 时间单位混淆：Python标准库的时间戳是秒级（比如time.time()返回带毫秒的浮点数），但你用的1549667056000是毫秒级时间戳，直接用的话调度器会等待几十年，完全不符合预期。
• 累积误差问题：哪怕用法正确，固定累加60秒的逻辑会因为get_data()的执行耗时（比如网络延迟）导致后续任务越来越偏离整分时刻，误差会不断累积。
• 无异常处理：API调用大概率会遇到网络问题、服务报错，如果不加捕获，程序直接崩溃，根本没法持续运行。

推荐的优化方案

根据你的精度需求，我给你分三个层级的方案：

方案1：基础精确定时（适合绝大多数场景，精度足够）

核心思路是每次重新计算下一个整分时刻，避免累积误差，同时加入异常保证程序健壮性。

import time
import requests

def get_data():
    try:
        # 替换成你的API调用逻辑
        response = requests.get("https://your-api-url.com/data")
        response.raise_for_status()  # 主动捕获HTTP错误
        data = response.json()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 获取数据成功：{data}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] API调用失败：{str(e)}")

def run_scheduler():
    while True:
        now = time.time()
        # 计算距离下一个整分的等待时间
        seconds_to_next_min = 60 - (now % 60)
        # 处理刚好卡在整分的情况，避免立即执行后重复触发
        if seconds_to_next_min < 0.1:
            seconds_to_next_min += 60
        # 等待到整分时刻
        time.sleep(seconds_to_next_min)
        # 执行任务
        get_data()

if __name__ == "__main__":
    run_scheduler()

这个方案的优势：

• 每次重新计算目标时间，彻底避免任务执行耗时带来的累积误差
• 异常捕获保证程序不会因为一次调用失败就终止
• 实现简单，不需要依赖第三方库

方案2：用schedule库简化代码（适合需要复杂定时规则的场景）

schedule是专门做定时任务的第三方库，语法更直观，可读性更强。
先安装依赖：pip install schedule

import schedule
import time
import requests

def get_data():
    try:
        response = requests.get("https://your-api-url.com/data")
        response.raise_for_status()
        data = response.json()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 获取数据成功：{data}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] API调用失败：{str(e)}")

# 设定每分钟整分执行任务
schedule.every().minute.at(":00").do(get_data)

def run_scheduler():
    while True:
        schedule.run_pending()
        # 每秒检查一次任务，避免CPU空转，不影响定时精度
        time.sleep(1)

if __name__ == "__main__":
    run_scheduler()

注意：这个方案的精度依赖于run_pending()的调用频率，每秒检查一次完全足够覆盖整分触发的需求，误差可以忽略。

方案3：高精度定时（适合毫秒级精度要求的场景）

如果你的场景对精度要求极高（比如误差不能超过几毫秒），可以用time.perf_counter()（系统级高精度计时器）替代普通时间判断，避免time.sleep()在高负载下的延迟问题。

import time
import requests

def get_data():
    try:
        response = requests.get("https://your-api-url.com/data")
        response.raise_for_status()
        data = response.json()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]}] 获取数据成功：{data}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]}] API调用失败：{str(e)}")

def run_high_precision_scheduler():
    while True:
        now = time.time()
        # 计算下一个整分的时间戳（秒级）
        next_minute = (now // 60 + 1) * 60
        # 循环等待，用高精度计时器判断时间，每次等待1毫秒减少CPU占用
        while time.perf_counter() < next_minute:
            time.sleep(0.001)
        get_data()

if __name__ == "__main__":
    run_high_precision_scheduler()

这个方案的精度能控制在几毫秒内，适合对时间敏感度极高的场景。

最后补充几个注意事项

• 永远不要用固定累加时间的方式定时，累积误差会让你的任务越来越偏离预期时间点
• 提前确认API的限流规则，避免每分钟一次的调用触发限流机制
• 可以把日志输出到文件，方便后续排查问题
• 如果需要后台持续运行，Linux下可以用nohup命令，Windows下可以做成系统服务

内容的提问来源于stack exchange，提问作者Devs11