
Python 中用 schedule 模块实现定时任务,轻量、易读、上手快,适合中小型项目或脚本级调度需求。它不依赖外部服务(如系统 cron 或 Celery),纯 Python 实现,核心逻辑就是“轮询 + 函数调用”,适合开发阶段快速验证或低频任务。
安装与基础用法
通过 pip 安装即可:
pip install schedule
最简示例:每 5 秒执行一次函数
立即学习“Python免费学习笔记(深入)”;
import schedule
import time
def job():
print("任务正在运行...")
schedule.every(5).seconds.do(job)
while True:
schedule.run_pending()
time.sleep(1)
注意:run_pending() 必须在循环中主动调用,模块本身不启新线程或进程 —— 这是它的设计特点,也是常见踩坑点。
常用时间规则写法
支持自然语言风格的调度语法,清晰直观:
-
schedule.every().hour.do(func)—— 每小时执行 -
schedule.every(10).minutes.do(func)—— 每 10 分钟执行 -
schedule.every().day.at("10:30").do(func)—— 每天 10:30 执行 -
schedule.every().monday.do(func)—— 每周一执行 -
schedule.every().wednesday.at("13:15").do(func)—— 每周三 13:15 执行
时间字符串格式固定为 "HH:MM"(24 小时制),不支持秒或日期偏移。
传参与任务管理
向任务函数传递参数,直接在 do() 后追加:
schedule.every(30).seconds.do(send_email, "hello@demo.com", subject="提醒")
取消任务可用标签(tag)批量管理:
schedule.every(10).minutes.do(cleanup_logs).tag('maintenance')
schedule.every().hour.do(backup_db).tag('backup')
# 取消所有 maintenance 类任务
schedule.clear('maintenance')
也可用 schedule.jobs 查看当前全部待执行任务列表,方便调试。
生产环境注意事项
schedule 本质是单线程轮询,不适合高精度、高并发或长期无人值守场景:
- 时间误差:依赖
sleep和检查间隔,最小粒度约 1 秒,不适用于毫秒级任务 - 阻塞风险:若任务执行时间 > 间隔时间,会堆积或跳过(默认不并发执行同一任务)
- 无持久化:程序退出即任务丢失;崩溃后不会自动恢复
- 无分布式支持:多实例部署时会重复触发
上线前建议包装成守护进程(如用 systemd 或 supervisor 管理),并添加简单日志和异常捕获:
def safe_job():
try:
job()
except Exception as e:
logging.error(f"任务失败: {e}")
对可靠性要求高的场景,应考虑 APScheduler(支持多触发器、持久化、协程)、Celery(分布式)或系统级 cron。










