APScheduler是一个强大的Python库,用于在指定时间执行定时任务。它基于触发器(trigger)和调度器(scheduler)的概念,允许你在程序中轻松地设置和管理各种定时任务。无论是周期性任务、一次性任务,还是固定时间执行的任务,APScheduler都提供了简单而灵活的解决方案。
你可以使用pip安装APScheduler:
pip install apscheduler
首先,你需要创建一个调度器实例:
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()
然后,定义一个任务函数,该函数将在指定时间被调用:
def my_job():
print("定时任务执行啦!")
使用触发器设置任务执行的时间,并将任务添加到调度器中:
from apscheduler.triggers.daily import DailyTrigger
trigger = DailyTrigger(hour=12, minute=0, second=0) # 每天中午12点执行
job = scheduler.add_job(my_job, trigger=trigger, id='my_job_id')
最后,启动调度器,让它开始执行任务:
scheduler.start()
通过修改任务的触发时间(使用 reschedule_job 方法)
# 修改任务:将任务的触发时间修改为每天下午3点
new_trigger = DailyTrigger(hour=15, minute=0, second=0)
scheduler.reschedule_job('my_job_id', trigger=new_trigger)
# 暂停任务
scheduler.pause_job('my_job_id')
print("任务已暂停")
# 恢复任务
scheduler.resume_job('my_job_id')
print("任务已恢复")
# 停止任务
scheduler.remove_job('my_job_id')
# 关闭调度器
scheduler.shutdown()
如果你的应用需要考虑不同时区的问题,可以通过配置调度器来设置时区:
from pytz import timezone
scheduler = BlockingScheduler(timezone=timezone('Asia/Shanghai'))
APScheduler支持将任务存储到数据库中,以便在应用重启后恢复。你可以使用不同的存储器,如SQLAlchemy、MongoDB等。例如,使用SQLite:
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
jobstores = {
'default': SQLAlchemyJobStore(url='sqlite:///jobs.db')
}
scheduler = BlockingScheduler(jobstores=jobstores)
为了捕获任务执行过程中的异常,你可以通过try-except块包裹任务函数的调用,并在except块中处理异常。
?
def my_job():
try:
# 任务代码
print("定时任务执行啦!")
except Exception as e:
print(f"任务执行失败:{e}")