不能靠backgroundtasks做定时调度,它仅支持单次、请求绑定的异步执行;必须用apscheduler配合lifespan事件管理生命周期,并使用sqlalchemyjobstore或redisjobstore持久化任务,否则重启丢任务、多进程重复执行、线程阻塞等问题必然发生。

不能靠 BackgroundTasks 做定时调度,它只负责“响应后顺手干点事”,不是“每5分钟跑一次”。真要调度脚本,必须用 APScheduler,且必须配合 lifespan 事件和外部存储(如 SQLite 或 Redis),否则重启就丢任务、多进程下重复执行、内存泄漏都是大概率事件。
为什么 BackgroundTasks.add_task() 里塞 while True 是错的
常见错误是把循环逻辑硬塞进 BackgroundTasks:
-
BackgroundTasks.add_task()只执行一次,内部while True: time.sleep(300)会卡死一个 Uvicorn worker 线程,无法回收 - Uvicorn 的 event loop 不允许阻塞式 sleep,会拖慢整个服务响应
- 没有持久化机制,进程一挂,所有任务状态全丢
- 如果用
--workers 4启动,每个进程都运行一遍这个循环,同一任务并发执行 4 次
APScheduler 必须用 BackgroundScheduler + lifespan 管理生命周期
AsyncIOScheduler 和 FastAPI/Uvicorn 的 event loop 冲突,强行用会导致调度器启动失败或任务不触发。正确做法是:
- 全局初始化一个
BackgroundScheduler实例,不要放在路由函数里反复 new - 在
@app.on_event("startup")中调用.start(),确保只启动一次 - 在
@app.on_event("shutdown")中调用.shutdown(wait=False),避免关机卡住 - 使用
SQLAlchemyJobStore或RedisJobStore,别用默认内存存储 —— 否则uvicorn main:app --reload一按保存,任务就清空
示例关键片段:
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
<p>jobstores = {"default": SQLAlchemyJobStore(url="sqlite:///jobs.sqlite")}
scheduler = BackgroundScheduler(jobstores=jobstores)</p><p>@app.on_event("startup")
async def start_scheduler():
scheduler.start()</p><p>@app.on_event("shutdown")
async def stop_scheduler():
scheduler.shutdown(wait=False)
</p>
如何通过 API 动态增删脚本任务(比如 run_backup.py)
核心是把脚本封装成可调用函数,并用 .add_job() 注册到调度器。注意 job_id 必须唯一,否则 .remove_job() 会失败。
- 任务函数建议统一接收
**kwargs,便于后续扩展参数(如env="prod"、target="/data") - 用 cron 表达式比 interval 更灵活,比如
"0 2 * * *"表示每天凌晨 2 点执行 - 添加任务时捕获
ConflictingIdError,避免重复注册同名 job - 删除前先用
.get_job(job_id)确认存在,否则抛JobLookupError
例如注册一个每日备份任务:
def run_backup_script(**kwargs):
import subprocess
subprocess.run(["python", "scripts/run_backup.py"], check=True)
<h1>添加任务</h1><p>scheduler.add_job(
func=run_backup_script,
trigger="cron",
hour="2",
id="daily_backup",
replace_existing=True
)
</p>
真正难的不是写这几行代码,而是让调度器在多 worker、自动 reload、容器重启、数据库连接中断等真实运维场景下依然稳定 —— 这些地方没做对,上线后第一周就会出问题。jobstore 选型、shutdown 的 wait 参数、日志级别设为 INFO 以上观察调度器状态,比写业务逻辑更值得花时间盯紧。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!











