

新闻资讯
技术学院APScheduler是Python中功能强大且灵活的定时任务调度库,支持内存级或持久化级任务管理,提供BlockingScheduler、BackgroundScheduler和AsyncIOScheduler三种常用调度器,以及date、interval、cron四种触发器,并建议生产环境使用SQLAlchemyJobStore持久化任务。
APScheduler(Advanced Python Scheduler)是 Python 中功能强大且灵活的定时任务调度库,适合在不依赖外部服务(如系统 cron 或 Celery)的情况下,实现内存级或持久化级的定时任务管理。它支持多种调度方式、多种作业存储后端和执行器,适用于从简单脚本到中型 Web 应用的各种场景。
安装后,最简示例只需导入、添加任务、启动:
pip install apscheduler
代码示例:
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
sched.add_job(lambda: print("Hello, world!"), 'interval', seconds=5)
sched.start()
这段代码每 5 秒打印一次,阻塞主线程运行。适合脚本类轻量任务。
APScheduler 提供 4 种调度器,日常使用主要关注以下三个:
:基于 asyncio,配合 async/await 任务使用,适合异步 Web 框架(如 FastAPI + uvicorn)不推荐在多进程环境(如 Gunicorn 多 worker)中直接使用内存型调度器,否则每个进程都会独立触发任务。
APScheduler 支持四种触发器(Trigger),最常用的是:
trigger='date', run_date='2025-04-10 14:30:00'
trigger='interval', minutes=30, seconds=15
trigger='cron', hour='9-17', day_of_week='mon-fri'(工作日早九晚五每小时一次)任务可配置的关键参数包括:id(唯一标识)、name(描述)、max_instances(防止并发堆积)、coalesce(错过执行时机是否合并为一次)、misfire_grace_time(容错窗口,默认 1 秒)。
默认所有任务存在内存里,重启即丢失。生产环境建议启用持久化:
SQLAlchemyJobStore 将任务存入 MySQL/PostgreSQL/SQLite,避免重复调度和状态丢失ThreadPoolExecutor 或 ProcessPoolExecutor 控制并发数,防止 I/O 或 CPU 密集型任务拖垮服务.shutdown() 释放资源try/except 包裹任务逻辑,并记录 error 级日志示例存储配置:
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
jobstores = {'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}