Python 定时任务框架apscheduler

引言

在开发中，经常会遇到需要在指定的时间点或者时间间隔执行某些任务的场景，比如定时备份数据、定时发送邮件、定时清理临时文件等。为了解决这类问题，可以使用定时任务框架来自动化完成这些任务。本文将介绍一款常用的Python定时任务框架：apscheduler。

什么是apscheduler？

apscheduler是一个强大的Python定时任务框架，可以实现各种基于时间的任务调度。它基于Quartz框架开发，支持多种调度方式，比如固定时间点触发、时间间隔触发、每天固定时间触发等。apscheduler还支持分布式部署，可以应对高并发情况下的任务调度需求。

安装

可以使用pip来进行安装，执行以下命令：

pip install apscheduler

基本用法

首先，我们需要导入apscheduler的相关模块：

from apscheduler.schedulers.blocking import BlockingScheduler

然后，我们需要创建一个调度器对象，并指定调度器的类型为BlockingScheduler。BlockingScheduler是一种阻塞式调度器，适用于简单的任务调度场景。我们还可以选择其他类型的调度器，比如BackgroundScheduler，这是一种非阻塞式调度器，适用于复杂的任务调度场景。

接下来，我们需要定义一个任务函数，用于执行我们需要调度的任务。任务函数的定义格式如下：

def job_function():
    # 任务的具体逻辑

在任务函数中，可以编写任何我们需要执行的逻辑代码。

然后，我们需要创建一个触发器对象，用于指定任务的触发条件。触发器对象可以设置任务的触发方式，比如固定时间点触发、时间间隔触发等。以下是几个常用的触发器类：
– DateTrigger：在指定的日期和时间触发任务；
– IntervalTrigger：在指定的时间间隔触发任务；
– CronTrigger：使用类似于Linux的cron表达式来触发任务。

我们只需要选择合适的触发器类，并根据需求进行配置。

最后，我们需要将任务函数和触发器对象添加到调度器中，通过调用add_job方法来完成添加。

待所有配置都完成后，我们可以通过调用start方法来启动调度器，它会根据我们的配置自动触发任务执行。

下面是一段示例代码，实现了每隔5秒钟输出一次当前时间的任务：

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def job_function():
    print(datetime.now())

scheduler = BlockingScheduler()
scheduler.add_job(job_function, 'interval', seconds=5)

scheduler.start()

运行以上代码，可以看到每隔5秒钟，控制台会输出一次当前时间。

触发器详解

在上一节中，我们简单介绍了几个常用的触发器类。本节将详细介绍这些触发器的使用方法，并给出一些示例代码。

DateTrigger

DateTrigger是一个简单的日期触发器，可以用于指定任务在指定的日期和时间触发。它的常用参数有以下几个：
– run_date：触发任务的日期和时间，可以是日期对象或者字符串。字符串的格式可以是类似于"2022-01-01 10:00:00"的日期时间格式；
– timezone：触发任务的时区，可以是时区字符串或者时区对象。

以下是一个使用DateTrigger的示例代码，实现了在指定的日期时间触发任务：

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.date import DateTrigger
from datetime import datetime

def job_function():
    print(datetime.now())

scheduler = BlockingScheduler()
trigger = DateTrigger(run_date='2022-01-01 10:00:00')
scheduler.add_job(job_function, trigger)

scheduler.start()

运行以上代码后，可以在指定的日期时间触发任务。

IntervalTrigger

IntervalTrigger是一个简单的时间间隔触发器，可以用于指定任务在指定的时间间隔触发。它的常用参数有以下几个：
– weeks：指定的周数；
– days：指定的天数；
– hours：指定的小时数；
– minutes：指定的分钟数；
– seconds：指定的秒数；
– start_date：触发任务的起始日期和时间。

以下是一个使用IntervalTrigger的示例代码，实现了每隔5秒钟输出一次当前时间的任务：

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.interval import IntervalTrigger
from datetime import datetime

def job_function():
    print(datetime.now())

scheduler = BlockingScheduler()
trigger = IntervalTrigger(seconds=5)
scheduler.add_job(job_function, trigger)

scheduler.start()

运行以上代码后，可以每隔5秒钟输出一次当前时间。

CronTrigger

CronTrigger是一个基于cron表达式的触发器，可以用于实现更为复杂的任务调度。cron表达式由6个字段组成，分别表示秒、分、时、天、月、周。cron表达式可以使用以下几种形式：
– *：匹配任意值；
– */n：匹配n的倍数；
– 1,2,3：匹配指定值列表；
– 1-5：匹配指定范围内的值。

以下是一个使用CronTrigger的示例代码，实现了每天10点执行一次任务：

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime

def job_function():
    print(datetime.now())

scheduler = BlockingScheduler()
trigger = CronTrigger(day_of_week='*', hour=10)
scheduler.add_job(job_function, trigger)

scheduler.start()

运行以上代码后，可以在每天10点执行一次任务。

分布式部署

在实际开发中，可能需要部署多台服务器来处理任务调度。apscheduler提供了一种分布式部署的解决方案，支持多台服务器之间的任务调度。它通过使用共享数据存储，比如数据库或者消息队列来实现任务的跨服务器调度。

在分布式部署中，只需要在每台服务器上启动apscheduler调度器，并配置相同的作业存储地址即可。以下是一段示例代码，展示了如何使用MySQL数据库作为任务存储的示例：

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.triggers.interval import IntervalTrigger
from datetime import datetime

def job_function():
    print(datetime.now())

# 创建MySQL数据库连接字符串，格式为：'数据库类型+数据库驱动名称://用户名:密码@主机名:端口号/数据库名'
db_url = 'mysql+mysqlconnector://username:password@localhost:3306/mydatabase'
# 创建一个SQLAlchemyJobStore对象，用于存储任务信息到MySQL数据库
jobstore = SQLAlchemyJobStore(url=db_url)

scheduler = BlockingScheduler()
scheduler.add_jobstore(jobstore)  # 将SQLAlchemyJobStore对象添加到调度器中

trigger = IntervalTrigger(seconds=5)
scheduler.add_job(job_function, trigger)

scheduler.start()

以上代码中，我们引入了SQLAlchemyJobStore模块，它用于将任务信息存储到MySQL数据库中。我们需要创建一个MySQL数据库连接字符串，用于指定要连接的数据库信息。通过创建一个SQLAlchemyJobStore对象，并将其添加到调度器中，就可以实现将任务信息存储到MySQL数据库中。

在分布式部署中，多台服务器可以共享同一个MySQL数据库，通过相同的连接字符串连接到同一个数据库，从而实现任务信息的共享和跨服务器的任务调度。