兄弟们，刚开始我也以为定时任务不就是sleep循环加个时间判断吗？结果第一个项目上线就翻车了——任务没跑、内存溢出、时区错乱，连踩三个坑。今天咱就从头到尾把Python定时任务的玩法撸一遍，不废话，直接上干货。
（开篇配图：Python定时任务执行失败的报错截图，红色标注关键错误行，展示“Thread-1”和“MemoryError”字样，营造真实踩坑氛围）
先看最简单的场景：你只想每天跑个脚本发邮件、备份数据库，用最基础的方案就够了。
1. 基础版：time.sleep + 循环（新手别这么干）
“`python
import time
from datetime import datetime
def job():
print(f”{date
time.now()} 执行任务”)
try:
while True:
job()
time.sleep(60) # 每分钟执行一次
except KeyboardInterrupt:
print(“任务已手动停止”)
“`
为什么这么写？ 因为简单到没朋友，适合本地调试。但有个致命问题——time.sleep 是阻塞的，如果任务执行了50秒，那实际间隔就变成110秒了。更坑的是，程序崩了或者服务器重启，任务就彻底没了。我第一个线上项目就这么完蛋的，凌晨3点日志停了，老板第二天追问：“昨晚的数据呢？”
2. 升级版：schedule库（推荐小项目用）
这是专门为定时任务设计的轻量库，pip install schedule就能用。
“`python
import schedule
import time
def backup_db():
print(“备份数据库…”)
schedule.every().day.at(“02:00”).do(backup_db).tag(“critical”)
schedule.every(30).minutes.do(check_health)
while True:
schedule.run_pending()
time.sleep(1)
“`
为什么推荐？ 代码可读性极强，每行都像英语句子。而且支持多个任务同时调度。但注意：它仍然是单线程的，如果一个任务卡住了，后面的都会排队。有个技巧：用 schedule.every().day 的 tag 参数给任务打标签，比如 schedule.every().day.at("02:00").do(backup_db).tag("critical")，后续可以批量取消或暂停。
这个设计真的反人类：run_pending() 必须自己开线程，官方文档明确说它是单线程的，建议用户自行处理。我踩的第二个坑就是没注意，导致健康检查任务被备份任务阻塞了3小时。
3. 进阶版：APScheduler（企业级首选）
如果你要处理定时、间隔、cron表达式、持久化任务，APScheduler是亲爹级的库。
“`python
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime
def send_report():
print(f”{datetime.now()} 发送周报”)
if __name__ == ‘__main__’:
scheduler = BlockingScheduler()
# 每周一早上9点执行
scheduler.add_job(
send_report,
CronTrigger(day_of_week='monday', hour=9, minute=0),
id='weekly_report',
replace_existing=True # 防止重复添加
)
# 每5秒执行一次，最多3次
scheduler.add_job(
lambda: print("心跳"),
'interval',
seconds=5,
max_instances=1,
id='heartbeat'
)
scheduler.start()
“`
为什么这么写？ BlockingScheduler 适合主程序就干这一件事；CronTrigger 支持标准的cron表达式，比schedule灵活十倍。replace_existing=True 这个参数救过我——项目部署时不小心起了两个进程，结果任务被重复执行，数据库写入了两倍数据。
还有个技巧：如果任务需要传递参数，用 add_job(func, args=[...])，别用闭包，闭包在序列化时会报错。我因为这个被生产环境坑了两次。
4. 分布式版：Celery + Redis/RabbitMQ
当你的项目需要水平扩展、任务队列、重试机制，Celery是工业标准。
“`python
# tasks.py
from celery import Celery
from celery.schedules import crontab
app = Celery(‘tasks’, broker=’redis://localhost:6379/0′)
@app.task(bind=True, max_retries=3)
def process_data(self, data_id):
try:
# 模拟耗时任务
print(f”处理数据ID: {data_id}”)
except Exception as e:
raise self.retry(exc=e, countdown=60)
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
# 每30秒执行一次
sender.add_periodic_task(30.0, process_data.s(data_id=1), name=’every_30s’)
# 每天凌晨3点执行
sender.add_periodic_task(
crontab(hour=3, minute=0),
process_data.s(data_id=2),
name=’daily_cleanup’
)
“`
为什么这么写？ bind=True 让任务可以访问 self 实例，这样才能用 self.retry() 重试。crontab 函数和Linux cron完全一致。Celery还有个杀手锏：任务结果可以存储到Redis，失败时可以自动重试，还能指定重试间隔。
不过配置起来确实麻烦，需要启动worker、beat、broker三个组件。我第一次部署时，忘记启动celery beat，定时任务一个都没跑。官方文档这段文档不够清晰，建议直接看我的配置脚本：
“`bash
# start_celery.sh
nohup celery -A tasks worker –detach &
nohup celery -A tasks beat –detach &
“`