
用 Python 实现一个简单任务调度器,核心是让函数在指定时间或按固定间隔自动执行,不需要引入复杂框架。关键在于轻量、可控、易理解。
基于 time.sleep 的轮询调度
适合低频、非精确要求的场景(比如每 5 秒检查一次文件、每分钟拉取一次状态)。原理是主循环中计算下次执行时间,用 time.sleep() 等待,避免忙等待。
- 用 datetime.now() 和 timedelta 计算下次触发时刻
- 每次循环前计算 sleep 秒数(确保不跳过,可处理执行耗时)
- 支持多个任务:每个任务记录自己的下一次执行时间
示例逻辑:
task = {"func": my_job, "interval": 30, "next_run": datetime.now() + timedelta(seconds=30)}使用 threading.Timer 实现单次/周期触发
比纯 sleep 更精准(尤其对单次延迟),且不阻塞主线程。适合需要并发执行多个定时任务的轻量需求。
立即学习“Python免费学习笔记(深入)”;
- threading.Timer(delay, func) 启动一次延时调用
- 在 func 内部重新创建 Timer,形成周期循环(注意防止重复启动)
- 用字典管理所有 Timer 实例,便于 stop() 取消
注意:Timer 不保证绝对准时(受 GIL 和线程调度影响),但比轮询更简洁可靠。
用 schedule 库快速上手(推荐入门)
第三方库 schedule(pip install schedule)语法极简,语义清晰,适合脚本和小型服务。
- 支持 .every(10).seconds.do(job)、.monday.at("10:30").do(report) 等自然写法
- 主线程调用 schedule.run_pending() 触发检查(配合 while + sleep 使用)
- 可随时 cancel_job() 或清空全部任务
它底层仍是轮询,但封装了时间解析、任务队列和并发安全细节,省去手动维护状态的麻烦。
注意事项与边界处理
简单调度器容易忽略实际运行中的问题:
- 任务执行异常需捕获,否则中断整个调度循环(用 try/except 包裹 func 调用)
- 长任务会挤压后续执行时机——可加超时控制或改用线程/进程池异步执行
- 程序退出前应清理 Timer 或停止 run_pending 循环,避免后台残留
- 跨天、夏令时等场景建议用 UTC 时间或成熟库(如 APScheduler)替代手写逻辑
不复杂但容易忽略。









