python APScheduler

### Python APScheduler 使用指南 #### 什么是APScheduler？ APScheduler 是一个功能强大的 Python 定时任务调度库，支持多种类型的定时任务调度方式。它能够帮助开发者轻松实现周期性任务、一次性任务以及复杂的 Cron 表达式任务调度[^1]。 --- #### 基本概念 - **调度器 (Scheduler)**：负责管理任务队列并触发任务执行。 - **触发器 (Trigger)**：定义任务何时运行，常见的有 `date`（指定日期）、`interval`（间隔时间）和 `cron`（Cron表达式）。 - **存储机制 (Job Store)**：用于保存任务数据，默认内存存储，也可以连接到 Redis 或数据库等外部持久化机制[^5]。 --- #### 安装 APScheduler 可以通过 pip 工具安装 APScheduler： ```bash pip install apscheduler ``` --- #### 示例代码 ##### 1. 简单的一次性任务 以下是一个简单的例子，演示如何设置一个在特定时间运行的任务： ```python from datetime import datetime, timedelta from apscheduler.schedulers.blocking import BlockingScheduler def one_time_task(): print(f"One-time task executed at {datetime.now()}") # 创建调度器实例 scheduler = BlockingScheduler() # 添加一次性任务，在当前时间基础上延后10秒执行 run_date = datetime.now() + timedelta(seconds=10) scheduler.add_job(one_time_task, 'date', run_date=run_date) print("Starting scheduler...") scheduler.start() ``` 此代码将在程序启动后的第10秒打印消息[^3]。 --- ##### 2. 按固定间隔运行任务 下面的例子展示了每三秒钟运行一次的任务： ```python from apscheduler.schedulers.blocking import BlockingScheduler def interval_task(): print(f"Interval task executed at {datetime.now()}") # 创建调度器实例 scheduler = BlockingScheduler() # 设置每隔3秒运行一次任务 scheduler.add_job(interval_task, 'interval', seconds=3) print("Starting scheduler...") scheduler.start() ``` 通过调整 `seconds` 参数，可以控制任务的运行频率[^3]。 --- ##### 3. 使用 Cron 表达式的复杂任务 如果需要更灵活的时间安排，比如每天的工作日早上9点半运行某个任务，则可以使用 Cron 触发器： ```python from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger def cron_task(): print(f"Cron task executed at {datetime.now()}") # 创建后台调度器实例 scheduler = BackgroundScheduler() # 配置 Cron Trigger，仅周一至周五上午9:30运行 trigger = CronTrigger(day_of_week="mon-fri", hour=9, minute=30, timezone="Asia/Shanghai") scheduler.add_job(cron_task, trigger) print("Starting background scheduler...") scheduler.start() # 主线程保持活动状态 try: while True: pass except KeyboardInterrupt: scheduler.shutdown() ``` 在此示例中，我们还设置了时区为上海标准时间[^4]。 --- ##### 4. 错误监听与事件处理 为了增强系统的健壮性，可以注册事件监听器来捕获任务执行中的异常情况： ```python from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR def error_listener(event): if event.exception: print('Task failed:', event.traceback) else: print('Task succeeded') # 注册错误监听器 scheduler.add_listener(error_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR) ``` 这样可以在任务失败时记录日志或采取其他补救措施[^2]。 --- #### 进阶特性 - **任务持久化** 如果希望任务能够在服务器重启后继续存在，可以选择将 Job 存储切换到 SQLite 数据库或其他支持的存储引擎： ```python from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore jobstore = { 'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite') } scheduler = BackgroundScheduler(jobstores=jobstore) ``` - **动态修改任务** 可以随时更新现有任务的属性，例如更改其触发条件或暂停/恢复任务： ```python scheduler.reschedule_job('my_job_id', trigger='cron', hour=10) scheduler.pause_job('my_job_id') scheduler.resume_job('my_job_id') ``` --- #### 总结 APScheduler 提供了一个简单而高效的接口来进行各种类型的任务调度操作。无论是基本的定时任务还是复杂的业务逻辑需求，都可以借助该工具完成开发目标[^5]。 ---