如何用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指定固定执行秒数。
阅读全文