APScheduler(Advanced Python Scheduler)是一个轻量级的 Python 定时任务调度框架,它支持多种调度任务的方式,包括固定时间间隔、固定时间点(日期)以及类似 Linux 下的 Crontab 命令。以下是如何使用 APScheduler 的基本步骤和代码示例:
安装 APScheduler
首先,使用 pip 安装 APScheduler:
pip install APScheduler
创建调度器
创建一个调度器对象,可以选择不同类型的调度器,例如 BackgroundScheduler
或 BlockingScheduler
。
from apscheduler.schedulers.background import BackgroundScheduler
# 创建后台调度器
scheduler = BackgroundScheduler()
定义任务函数
定义一个任务函数,这个函数是你想要定时执行的代码。
def my_job():
print("执行定时任务")
添加任务
使用调度器的 add_job
方法添加任务,并设置触发器(trigger)。
# 添加任务到调度器,使用IntervalScheduler,每隔5秒执行一次
scheduler.add_job(my_job, 'interval', seconds=5)
启动调度器
调用调度器的 start
方法启动调度任务。
# 启动调度器
scheduler.start()
代码示例
以下是一个完整的代码示例,演示如何使用 APScheduler 创建定时任务并将其调度执行:
from apscheduler.schedulers.background import BackgroundScheduler
# 创建调度器
scheduler = BackgroundScheduler()
# 定义一个要执行的任务
def my_job():
print("执行定时任务")
# 添加任务到调度器,使用IntervalScheduler,每隔5秒执行一次
scheduler.add_job(my_job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
# 阻塞当前进程,直到按下Ctrl+C
try:
while True:
pass
except (KeyboardInterrupt, SystemExit):
# 关闭调度器
scheduler.shutdown()
异常处理
APScheduler 提供了异常处理机制,可以在任务执行中捕获异常。
from apscheduler.schedulers.background import BackgroundScheduler
# 创建调度器
scheduler = BackgroundScheduler()
# 定义一个可能抛出异常的任务
def my_job():
try:
# 执行可能引发异常的代码
result = 1 / 0
except Exception as e:
print(f"任务执行出现异常: {str(e)}")
# 添加任务到调度器,使用IntervalScheduler,每隔5秒执行一次
scheduler.add_job(my_job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
# 阻塞当前进程,直到按下Ctrl+C
try:
while True:
pass
except (KeyboardInterrupt, SystemExit):
# 关闭调度器
scheduler.shutdown()
调度器持久化
APScheduler 支持将任务持久化到数据库,以便在程序重启后能够恢复任务。
```python
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
创建数据库存储
jobstores = {
'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
创建后台调度器,并指定 jobstores 参数
scheduler = BackgroundScheduler(jobstores=jobstores)
定义任务函数
def my_job():
print("执行定时任务")
添加定时任务,每隔5秒执行一次
scheduler.add_job(my_job, 'interval', seconds=5)
启动调度器
scheduler.start()
主线程等待一段时间后结束
try:
while True: