Python schedule模块详细介绍

1. 创始时间与作者

• 创始时间：schedule 首次发布于 2013年（首个版本 v1.0.0 于 2014年发布），是一个轻量级的 Python 任务调度库。

• 核心开发者：

• Dan Bader：德国软件工程师，Python 培训师和作者，schedule 的主要创始人和维护者。

• 开源社区贡献：超过 100 位贡献者参与优化和扩展。

• 项目定位：简单易用的进程内任务调度器，用于定期执行 Python 函数，提供人性化的 API 语法。

2. 官方资源

• GitHub 地址：https://github.com/dbader/schedule

• 官方文档：https://schedule.readthedocs.io/

• PyPI 地址：https://pypi.org/project/schedule/

3. 核心功能

4. 应用场景

1. 定时数据备份

import scheduleimport timeimport shutildef backup_database():"""模拟数据库备份任务"""shutil.copy2('data.db', 'backups/data_backup.db')print("数据库备份完成")# 每天凌晨2点执行备份schedule.every().day.at("02:00").do(backup_database)while True:schedule.run_pending()time.sleep(1)

2. 定期爬取网站内容

import scheduleimport requestsimport timedef fetch_stock_prices():"""模拟获取股票价格"""# response = requests.get('https://api.example.com/stocks')print("股票价格已更新")# 每10分钟执行一次schedule.every(10).minutes.do(fetch_stock_prices)while True:schedule.run_pending()time.sleep(1)

3. 定时发送报告邮件

import scheduleimport smtplibfrom email.mime.text import MIMETextdef send_weekly_report():"""发送每周报告"""msg = MIMEText("这是本周的汇总报告...")msg['Subject'] = '周报'msg['From'] = 'sender@example.com'msg['To'] = 'receiver@example.com'# 发送邮件代码print("周报邮件已发送")# 每周一上午9:30发送schedule.every().monday.at("09:30").do(send_weekly_report)while True:schedule.run_pending()time.sleep(60)  # 每分钟检查一次

4. 系统监控与告警

import scheduleimport psutilimport timedef check_cpu_usage():"""检查CPU使用率，超过阈值则告警"""usage = psutil.cpu_percent(interval=1)if usage > 80:print(f"警告：CPU使用率过高 {usage}%")def check_memory():"""检查内存使用"""memory = psutil.virtual_memory()if memory.percent > 85:print("警告：内存不足")# 每30秒检查一次schedule.every(30).seconds.do(check_cpu_usage)schedule.every(1).minute.do(check_memory)while True:schedule.run_pending()time.sleep(1)

5. 底层逻辑与技术原理

核心架构

关键技术

1. 基于时间的循环检查：

• 核心是一个无限循环（通常在主线程中），不断调用 run_pending() 检查是否有任务到期。

• 使用 time.sleep() 控制检查频率，避免 CPU 空转。

2. Job 对象模型：

• 每个任务被封装为一个 Job 对象，包含要执行的函数、参数、调度间隔、下次运行时间等。

• 支持链式调用：schedule.every(10).minutes.do(job)。

3. 时间计算：

• 使用 datetime 模块计算下次执行时间。

• 支持多种时间单位：秒、分钟、小时、天、周，以及具体时间点（at("HH:MM")）。

4. 简单队列管理：

• 所有 Job 对象存储在一个列表中，每次调用 run_pending() 时遍历列表，判断时间并执行到期任务。

• 执行后自动更新 next_run 并重新排序。

5. 单线程执行：

• 默认所有任务在同一个线程中顺序执行，若一个任务耗时过长，会阻塞后续任务的准时执行。

• 可通过自行创建线程或使用 threading 模块实现并发。

6. 安装与配置

基础安装

pip install schedule

环境要求

验证安装

import scheduledef job():print("I'm working...")schedule.every(2).seconds.do(job)print("等待2秒...")schedule.run_all()  # 立即运行所有任务print("安装成功！")

7. 高级功能使用

1. 带参数的任务

import scheduledef greet(name):print(f"Hello, {name}!")# 传递参数schedule.every(5).seconds.do(greet, name="Alice")while True:schedule.run_pending()time.sleep(1)

2. 使用装饰器 @repeat

from schedule import repeat, every, run_pendingimport time@repeat(every(10).minutes)def job():print("I am a scheduled job")while True:run_pending()time.sleep(1)

3. 取消任务

import scheduledef job():print("我会运行三次")# 取消自身return schedule.CancelJob# 返回 CancelJob 会取消任务（只运行一次）schedule.every(5).seconds.do(job)# 或者通过标签取消def job2():print("每3秒运行，5秒后取消")schedule.every(3).seconds.do(job2).tag('fast-task')# 5秒后取消标签为 'fast-task' 的所有任务schedule.clear('fast-task')

4. 获取待执行任务

import scheduledef job():print("任务")schedule.every(5).seconds.do(job)schedule.every(10).minutes.do(job)# 查看所有任务all_jobs = schedule.get_jobs()print(f"当前有 {len(all_jobs)} 个任务")# 查看下次执行时间for job in all_jobs:print(f"下次运行: {job.next_run}")

5. 并行执行（多线程）

import scheduleimport threadingimport timedef run_continuously(interval=1):"""在后台线程中持续运行调度器"""cease_continuous_run = threading.Event()class ScheduleThread(threading.Thread):@classmethoddef run(cls):while not cease_continuous_run.is_set():schedule.run_pending()time.sleep(interval)continuous_thread = ScheduleThread()continuous_thread.start()return cease_continuous_run# 启动后台线程stop_run_continuously = run_continuously()def job1():print("任务1开始")time.sleep(5)  # 模拟耗时操作print("任务1结束")def job2():print("任务2")schedule.every(2).seconds.do(job1)schedule.every(2).seconds.do(job2)# 主线程可以做其他事time.sleep(10)stop_run_continuously.set()  # 停止后台线程

8. 与同类工具对比

9. 企业级应用案例

1. 运维自动化：

• 定期清理临时文件、备份数据库、轮转日志。

• 监控系统指标并在达到阈值时发送告警。

2. 数据管道：

• 每 10 分钟从 API 拉取数据并存入数据库。

• 每天凌晨汇总前一天的数据生成报表。

3. 测试环境：

• 定时执行自动化测试并发送报告。

• 模拟周期性负载测试。

4. 物联网设备：

• 在边缘设备上定期采集传感器数据并上报。

总结

schedule 是 Python 中最友好的轻量级任务调度库，核心价值在于：

1. 极简 API：人性化的链式语法，几行代码即可实现定时任务。

2. 零依赖：纯 Python 实现，无需外部服务或配置文件。

3. 灵活易用：支持多种时间单位和精确时间点，可传递参数。

4. 适合嵌入：可轻松集成到任何 Python 应用程序中。

技术亮点：

• 使用 datetime 进行精确时间计算。

• Job 对象模型支持丰富的调度规则。

• 可通过标签分组管理任务。

适用场景：

• 小型项目：需要定期执行简单任务。

• 脚本自动化：替代系统 crontab，便于跨平台。

• 应用内调度：在应用程序内部执行周期性操作。

• 教学与原型开发：快速验证定时任务逻辑。

何时可能不适用：

• 需要 毫秒级精度 或 分布式调度。

• 任务需要 持久化 和 故障恢复。

• 需要 复杂的调度规则（如每月的最后一个周五）。

安装使用：

pip install schedule

学习资源：

• 官方文档：https://schedule.readthedocs.io/

• GitHub 仓库：https://github.com/dbader/schedule

截至 2024 年，schedule 在 GitHub 收获 12k+ Star，月下载量超 数百万次，是 Python 生态中轻量级任务调度的首选工具。项目遵循 MIT 开源协议，可免费用于商业和非商业项目。