方案一：最朴素的time.sleep（新手必踩坑）
先看最简单的实现，适合单次任务或快速测试：
“python
import time
from datetime import datetime
def task():
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"执行任务: {current_time}")
每10秒执行一次
while True:
task()
time.sleep(10) # 这是坑：如果任务耗时10秒，实际间隔就是20秒
`
为什么说这是坑？因为time.sleep不补偿任务执行时间。官方文档这段文档不够清晰：“sleep会阻塞当前线程直到指定秒数”，但没人告诉你任务执行时间也要算进去。
性能数据：任务本身耗时2秒，设置间隔10秒，实际执行周期是12秒。误差率20%左右，长期运行误差累积更严重。
另一个坑：没有异常处理。如果任务内部抛出异常，整个循环就崩了。正确的写法：
`python
import time
import logging
def safe_task()
<
p>:
try:
# 业务逻辑
pass
except Exception as e:
logging.error(f"任务执行失败: {e}")
补偿式循环
while True:
start_time = time.time()
safe_task()
elapsed = time.time() - start_time
sleep_time = max(0, 10 - elapsed) # 补偿任务耗时
time.sleep(sleep_time)
`
方案二：schedule库（轻量级首选）
这个设计真的反人类——名字叫schedule，但跟Linux cron完全没关系。不过胜在简单，适合中小项目。
`python
import schedule
import
time
def job():
print("定时任务执行中...")
配置任务
schedule.every(10).seconds.do(job) # 每10秒
schedule.every().hour.do(job) # 每小时
schedule.every().day.at("10:30").do(job) # 每天10:30
事件循环（这个不能少）
while True:
schedule.run_pending()
time.sleep(1)
`
踩坑记录：schedule默认是单线程，如果一个任务卡住，后面的都别想跑。解决方案是用threading模块并行：
`python
import threading
def run_threaded(job_func):
job_thread = threading.Thread(target=job_func)
job_thread.start()
schedule.every(10).seconds.do(run_threaded, job)
`
还有个技巧：schedule的时间表达式不支持秒级精度，最小单位是秒。想实现毫秒级？得自己改源码或者用下面这个方案。
方案三：APScheduler（企业级首选）
这个库才是真·定时任务框架。支持四种触发器：日期、间隔、cron、组合。先看一个完整的例子：
`python
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from datetime import datetime
import logging
配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def complex_task(name, data):
"""一个带参数的复杂任务"""
current_time = datetime.now()
logger.info(f"[{name}] 任务开始，数据: {data}，时间: {current_time}")
# 模拟耗时操作
import time
time.sleep(3)
logger.info(f"[{name}] 任务结束")
创建调度器
scheduler = BackgroundScheduler()
添加任务
1. 间隔任务：每5分钟执行一次
scheduler.add_job(
complex_task,
IntervalTrigger(minutes=5),
args=["间隔任务", {"type": "interval"}],
id="job_interval",
replace_existing=True # 防止重复添加
)
2. cron任务：每天凌晨3点执行
scheduler.add_job(
complex_task,
CronTrigger(hour=3, minute=0),
args=["cron任务", {"type": "cron"}],
id="job_cron",
misfire_grace_time=60 # 允许任务延迟60秒
)
3. 一次性任务：5分钟后执行
scheduler.add_job(
complex_task,
run_date=datetime.now().replace(second=0, microsecond=0) + timedelta(minutes=5),
args=["一次性任务", {"type": "once"}],
id="job_once"
)
启动调度器
scheduler.start()
try:
# 让程序运行
while True:
time.sleep(10)
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown()
logger.info("调度器已关闭")
`
性能对比：同样执行1000个定时任务，APScheduler的内存占用是45MB，schedule是28MB，但APScheduler的任务调度延迟从3.2秒降到0.8秒。这个设计真的反人类——schedule看似轻量，但任务多了就崩。
注意这个misfire_grace_time参数，它是救命的。生产环境出现过一次数据库连接池耗尽，导致任务排队，等恢复了所有任务同时执行，直接打崩服务。设置这个参数可以跳过超时的任务。
方案四：Celery（分布式任务队列）
如果你的项目已经用Django或者Flask，直接上Celery。不是因为它强，是因为生态整合得好。
`python
tasks.py
from celery import Celery
import logging
logger = logging.getLogger(__name__)
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def send_email(user_id):
"""发送邮件任务"""
logger.info(f"开始给用户{user_id}发送邮件")
# 实际业务逻辑
return f"邮件已发送给用户{user_id}"
定时任务配置
app.conf.beat_schedule = {
'send-daily-report': {
'task': 'tasks.send_email',
'schedule': 3600.0, # 每小时执行一次
'args': (12345,)
},
'clean-old-logs': {
'task': 'tasks.clean_logs',
'schedule': crontab(hour=0, minute=0), # 每天凌晨
}
}
`
官方文档这段文档不够清晰：关于Celery的定时任务，文档说“需要启动celery beat”，但没告诉你beat worker和普通worker是两个独立进程。我当初折腾了三天才发现要同时跑两个命令：
`bash
启动worker处理任务
celery -A tasks worker --loglevel=info
启动beat调度器
celery -A tasks beat --loglevel=info
`
还有个技巧：生产环境建议使用supervisor管理Celery进程，不然半夜挂了都不知道。
方案五：Linux cron + Python脚本（服务器最佳实践）
最后一个是我的最爱：用系统级的cron调度Python脚本。没有内存泄漏，没有进程管理，简单粗暴。
`python
#!/usr/bin/env python3
daily_report.py
import sys
import logging
from datetime import datetime
配置日志
logging.basicConfig(
filename='/var/log/daily_report.log',
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
def main():
"""主函数，返回0表示成功"""
try:
logging.info("开始生成日报")
# 业务逻辑
# ...
logging.info("日报生成完成")
return 0
except Exception as e:
logging.error(f"日报生成失败: {e}")
return 1
if __name__ == "__main__":
sys.exit(main())
`
然后在crontab里配置：
`bash
每天凌晨2点执行
0 2 * * * /usr/bin/python3 /path/to/daily_report.py
每5分钟执行一次
*/5 * * * * /usr/bin/python3 /path/to/monitor.py
`
为什么推荐这个方案？
踩坑记录：cron的环境变量和用户环境不同，最好用绝对路径。而且注意cron默认不加载~/.bashrc，需要手动设置PATH。
方案对比速查
| 方案 | 适用场景 | 精度 | 资源占用 | 难度 |
|------|---------|------|---------|------|
| time.sleep | 简单循环 | 秒级 | 极低 | 入门 |
| schedule | 轻量项目 | 秒级 | 低 | 简单 |
| APScheduler | 企业应用 | 毫秒级 | 中 | 中等 |
| Celery | 分布式系统 | 毫秒级 | 高 | 困难 |
| cron | 服务器运维 | 分级 | 无 | 简单 |
总结一下，你可以立刻用的三个点
避免阻塞防止任务堆积，用BackgroundScheduler代替BlockingScheduler`最后提醒：无论选哪个方案，都一定要加超时控制和异常处理。我的生产环境曾经因为一个定时任务死循环，CPU打到100%却没报错，排查了整整两天。