schedule任务不执行的根本原因是未调用run_pending()启动轮询;需配合sleep防止空转,避免多线程混用,用sqlite3存状态(启wal模式),通过字典管理任务启停,并在do函数外加try/except防中断。

Python任务管理系统不是非得用 Django 或 FastAPI 从零搭后台——多数内部工具、自动化调度场景,用 schedule + sqlite3 + 简单 CLI 就能跑通核心逻辑,且便于调试和嵌入现有脚本。
用 schedule 做定时任务时,为什么任务不执行?
常见现象是调用了 schedule.every().hour.do(func) 却没反应,根本原因是没启动事件循环。这个库本身不运行,只负责“记事”,必须手动轮询。
- 必须在主循环里持续调用
schedule.run_pending(),通常配合time.sleep(1)防止 CPU 空转 - 不要把
schedule和多线程/多进程混用:它默认非线程安全;若需并发执行,得用threading.Thread包裹每个do()函数,并自行处理状态同步 - 时间精度有限:最小间隔为 1 秒,
every(30).seconds实际可能延迟 100–500ms,别指望毫秒级触发
任务状态和历史存哪?选 sqlite3 而不是 JSON 文件
JSON 适合配置,不适合任务日志——并发写入会丢数据,没事务,查最近 10 条失败记录都得全文件加载解析。
- 建表语句建议包含:
id(INTEGER PRIMARY KEY AUTOINCREMENT)、task_name(TEXT)、status(TEXT CHECK(status IN ('pending','running','success','failed'))、started_at、finished_at、error_msg(TEXT) - 每次任务开始前插入一条
pending记录,用lastrowid拿到 ID,结束时用该 ID 更新状态和时间戳——避免靠名字或时间范围做模糊匹配 - SQLite 的 WAL 模式(
PRAGMA journal_mode = WAL)能显著降低多进程写冲突概率,尤其当 CLI 启动任务和后台守护进程共用同一 DB 时
CLI 命令怎么支持 “启停指定任务” 而不重启整个调度器?
schedule 本身不提供任务粒度的启停 API,得自己维护一个可变的任务注册表。
- 定义一个全局字典
active_jobs = {},key 是任务名,value 是schedule.Job实例 - 封装
add_job(name, func, *args, **kwargs):先创建 job,再存进字典;同时加个remove_job(name),调用schedule.cancel_job(active_jobs[name])并从字典删掉 - CLI 接收
start task_x时,检查task_x是否已在active_jobs中,不在就新建并注册;接收stop task_x就调remove_job—— 这样不用 kill 进程也能动态调整
真正麻烦的是错误传播:某个任务抛异常,默认会中断整个 run_pending() 循环。必须在每个 do() 函数外层套 try/except,把异常捕获后写进数据库,再继续下一轮轮询——这点很容易被忽略,结果就是某次网络超时导致所有后续任务全部停摆。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











