APScheduler

简介: 【10月更文挑战第09天】

APScheduler(Advanced Python Scheduler)是一个轻量级的 Python 定时任务调度框架,它支持多种调度任务的方式,包括固定时间间隔、固定时间点(日期)以及类似 Linux 下的 Crontab 命令。以下是如何使用 APScheduler 的基本步骤和代码示例:

安装 APScheduler

首先,使用 pip 安装 APScheduler:

pip install APScheduler

创建调度器

创建一个调度器对象,可以选择不同类型的调度器,例如 BackgroundSchedulerBlockingScheduler

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:

目录
相关文章
|
6月前
|
调度 Apache
airflow scheduler -D 是什么作用
【6月更文挑战第30天】airflow scheduler -D 是什么作用
112 1
|
4月前
|
分布式计算 Hadoop 关系型数据库
dolphinscheduler
先根据伪集群来部署、部署中参考非伪集群 1、mysql数据库 mysql -h主机地址 -u用户名 -p 2、查看等 less:G、上下按键、ctrl+b、ctrl+f、q find / -iname '*mysql*' 更改目录所有者 chown -R dolphinscheduler:dolphinscheduler apache-dolphinscheduler-*-bin /usr/apache-dolphinscheduler-3.1.0-bin/tools/libs 里面也需要mysql驱动
63 4
|
7月前
|
Apache 调度 数据库
Apache DolphinScheduler VS WhaleScheduler
Apache DolphinScheduler VS WhaleScheduler
722 59
|
7月前
|
调度
APScheduler任务相关操作
APScheduler任务相关操作
50 0
|
7月前
|
存储 NoSQL Java
APScheduler简介
APScheduler简介
74 0
|
7月前
|
Java 调度 数据库管理
APScheduler自定义配置
APScheduler自定义配置
72 0
|
7月前
|
资源调度 分布式计算 算法
Gang Scheduling
Gang Scheduling(Coscheduling)、FIFO Scheduling、Capacity Scheduling、Fair sharing、Binpack/Spread等是云计算和分布式系统中的任务调度算法,用于在资源有限的情况下,公平、高效地分配任务和资源。下面是这些调度算法的基本介绍和如何在实际应用中使用它们的一些建议:
302 2
|
Java 调度
ScheduledExecutorService使用介绍
JUC包(java.util.concurrent)中提供了对定时任务的支持,即ScheduledExecutorService接口。 本文对ScheduledExecutorService的介绍,将基于Timer类使用介绍进行,因此请先阅读Timer类使用介绍文章。
1273 1
|
Kubernetes 算法 调度
基于kube-scheduler-simulator编写自己的调度程序
基于kube-scheduler-simulator编写自己的调度程序
155 0