发布于2026-07-06 阅读(0)
扫一扫,手机访问
定时任务开发中,Python 的 schedule 库虽然用起来方便,但很多人在用它实现“每 60 秒执行一次”时会踩一个坑:任务跑着跑着,间隔就飘了。比如你设了 60 秒,结果实际两次触发间隔变成了 91 秒——因为 .every(n).seconds.do() 默认是串行阻塞的,下一个任务要等上一个跑完才开始计时。这在金融结算、数据快照、IoT 采样等场景下,是没法接受的。
要解决这个问题,核心思路很简单:**不用相对延迟,改用绝对时间戳做锚点**——每次任务执行前,先算好下一次应该几点几分几秒触发,然后直接在那个时刻安排下去。Python 标准库里的 sched.scheduler 就是干这个的。
下面是一个可以直接上手的例子,实现**每 60 秒整点触发**(改成每 15 分钟也很简单,把 frequency_seconds 设为 900 就行):
#!/usr/bin/env python3
import sched
import time
# 配置调度参数
frequency_seconds = 60 # 改为 900 即为每15分钟
priority = 1
# 创建调度器(使用 time.time 和 time.sleep 作为时间源与阻塞函数)
scheduler = sched.scheduler(time.time, time.sleep)
def job():
# ✅ 关键:立即预约下一次执行的绝对时间点(当前时间 + 固定间隔)
next_run_at = time.time() + frequency_seconds
scheduler.enterabs(next_run_at, priority, job)
# 执行业务逻辑(此处模拟耗时操作)
print("Process triggered at:", time.strftime("%Y-%m-%d %H:%M:%S"))
time.sleep(25) # 模拟长任务(>30s),不影响下次触发时间
print("Process ended at:", time.strftime("%Y-%m-%d %H:%M:%S"))
# 启动首次调度(延迟0秒,即立即执行)
scheduler.enter(0, priority, job)
# 运行调度循环(阻塞式,适合单任务主程序)
print("Scheduler started...")
scheduler.run()
运行效果是这样的(记住,任务耗时 25 秒,但下一次触发依然严格在 60 秒后,而不是从结束时间算起):
Process triggered at: 2024-03-25 06:57:23 Process ended at: 2024-03-25 06:57:48 Process triggered at: 2024-03-25 06:58:23 ← 严格间隔60秒(非从结束时间起算) Process ended at: 2024-03-25 06:58:48
关键机制说明:
scheduler.enterabs(when, priority, action) 接受的是**绝对时间戳**(Unix timestamp),这就保证了下次调用精准落在 t₀ + Δt 那一刻;scheduler.enter(0, ...) 用来启动第一个任务,后续的所有调度都由 job() 内部自己预约,形成一个自维持的闭环;sched 是线程安全的单线程调度器,天然避免并发冲突,特别适合 I/O 密集型的定时任务。注意事项与进阶建议:
sched 了,改用 threading.Timer 或者 concurrent.futures.ThreadPoolExecutor 来封装调度逻辑;job() 里加上 try...except,避免一次异常让整个调度挂掉;time.time() 换成 time.perf_counter(),并相应调整调度器的初始化参数;IntervalTrigger 配合 coalesce=False + max_instances 也能实现类似效果,不过要额外安装第三方库。
说到底,当“准时”这件事关系到业务正确性时,与其凑合用那个“简单但漂移”的 schedule 库,不如花五分钟换成标准库里的 sched + enterabs()——更可靠、更轻量、也更好控制。
售后无忧
立即购买>office旗舰店
售后无忧
立即购买>office旗舰店
售后无忧
立即购买>office旗舰店
售后无忧
立即购买>office旗舰店
正版软件
正版软件
正版软件
正版软件
正版软件
1
2
3
7
8