如何用schedule轻量级定时任务调度库实现复杂任务调度?
摘要:schedule是一款专为简化定时任务调度而设计的Python库,它通过直观的语法降低了周期性任务的实现门槛。作为进程内调度器,它无需额外守护进程,轻量且无外部依赖,适合快速搭建自动化任务。不过,该库在功能完整性上有所取舍,目前暂不支持断点
schedule是一款专为简化定时任务调度而设计的Python库,它通过直观的语法降低了周期性任务的实现门槛。作为进程内调度器,它无需额外守护进程,轻量且无外部依赖,适合快速搭建自动化任务。不过,该库在功能完整性上有所取舍,目前暂不支持断点续传、亚秒级精度控制以及多任务并行执行等复杂场景。
schedule库的官方仓库地址见:schedule,schedule库的官方文档见:schedule-doc。
schedule库支持在Python 3.7及以上版本的环境中运行,schedule库的安装命令如下:
pip install schedule
目录1 使用入门1.1 基础使用1.1.1 相对调用1.1.2 绝对调用1.2 进阶使用1.2.1 调用程序管理1.2.2 调用时间管理1.2.3 调用运行方式管理2 参考
1 使用入门
1.1 基础使用
1.1.1 相对调用
schedule库通过创建调度器,设置时间单位,注册待调用函数,返回任务对象实现任务周期调用。例如:
# 每3秒执行一次任务
schedule.every(3).seconds.do(job)
这种调用方式是相对调用方式,任务执行间隔是相对于当前时间点计算的。例如17:59:01创建调度器,下一次执行时间为17:59:04。
示例代码如下:
import schedule
import time
from datetime import datetime
def get_now_time():
now = datetime.now()
now = now.strftime("%Y-%m-%d %H:%M:%S")
return now
def job():
"""定时执行的任务函数"""
print(f"{get_now_time()} I'm working...")
print(f"调度器创建时间:{get_now_time()}")
# 调度器返回job对象
schedule_job = schedule.every(3).seconds.do(job)
print(f"调度器下一次运行时间:{schedule_job.next_run}")
# 任务调度主循环
# 持续检查是否有待执行的任务
while True:
schedule.run_pending() # 检查并执行待执行的任务
time.sleep(1) # 休眠1秒避免CPU占用过高
# 为None就是没运行
print(f"调度器上一次运行时间:{schedule_job.last_run}")
print(f"调度器下一次运行时间:{schedule_job.next_run}")
其他相对时间间隔调用代码如下:
# 每3分钟执行一次任务
schedule.every(3).minutes.do(job)
# 每小时执行一次任务
schedule.every().hours.do(job)
# 每3天执行一次任务
schedule.every(3).days.do(job)
# 每3周执行一次任务
schedule.every(3).weeks.do(job)
# 每周一执行任务
schedule.every().monday.do(job)
1.1.2 绝对调用
schedule可以在相对调用的基础上结合at函数实现绝对时间的调度。例如:
schedule.every(3).hours.at("11:16").do(job)
上述代码可拆解为:
job1 = schedule.every(3).hours
job2 = job1.at("11:16").do(job)
此处的job1基于当前调用器创建时间(例如18:34:54),以小时为间隔粒度进行设定,即每间隔3小时执行一次任务,因此下一次任务执行时间为21:34:54。而at()函数的作用是在job1设定的时间粒度和间隔范围内,具体指定分钟和秒。以job2中的at("11:16")为例,这里的11:16代表分钟和秒。它会在18:34:54至21:34:54的时间区间内,定位离21:34:54最近的11分16秒时刻,即21:11:16。
若按每分钟调用一次任务,可通过at指定固定执行秒数。
