Python定时任务实现_schedule模块实践

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。