深入了解APScheduler触发器：Python定时任务调度的灵活选择

不止于python

发布于 2023-10-16 20:36:26

1.2K0

发布于 2023-10-16 20:36:26

文章被收录于专栏：不止于python不止于python

APScheduler是一个功能强大的Python任务调度库，它支持多种定时任务的调度方式，包括固定时间间隔、特定时间执行等。下面详细介绍APScheduler的使用和一个用例。

安装

pip install apscheduler

基本使用

基本构成

调度器（Scheduler）: 调度器是APScheduler的核心组件，负责管理和执行任务。APScheduler提供了不同类型的调度器，包括阻塞调度器（BlockingScheduler）和非阻塞调度器（BackgroundScheduler）。
触发器（Trigger）: 触发器定义了何时执行任务。APScheduler支持各种触发器类型，如固定时间间隔触发器（IntervalTrigger）、日期触发器（DateTrigger）等。
任务（Job）: 任务是要执行的操作或函数。在APScheduler中，你可以创建任务并将它们与触发器关联起来，以指定何时执行任务。

使用示例

以下是一个使用APScheduler的示例，演示如何创建一个简单的定时任务，每隔5秒输出一次当前时间：



# -*- coding: utf-8 -*-
# @Author: Mehaei
# @Date:   2023-10-09 20:26:13
# @Last Modified by:   Mehaei
# @Last Modified time: 2023-10-09 22:27:28
import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.interval import IntervalTrigger




# 定义一个任务函数
def my_task():
    current_time = datetime.datetime.now()
    print(f"任务执行时间：{current_time}")


# 创建一个阻塞调度器
scheduler = BlockingScheduler()


# 创建一个触发器，每隔5秒触发一次
trigger = IntervalTrigger(seconds=5)


# 添加任务和触发器到调度器
scheduler.add_job(my_task, trigger)


try:
    # 启动调度器
    scheduler.start()
except (KeyboardInterrupt, SystemExit):
    # 用户按Ctrl+C或者程序退出时停止调度器
    scheduler.shutdown()

在实际应用中，可以根据需要配置不同类型的触发器和任务，以执行各种定时任务。APScheduler还支持将任务持久化到数据库，以便在应用重启后保留任务信息。

这只是一个简单的示例，APScheduler提供了更多高级功能，如定时任务的持久化、任务的传参、任务的异常处理等。下面简单写下触发器和调度器的类型和使用。

触发器

1. IntervalTrigger（固定时间间隔触发器）

IntervalTrigger以固定的时间间隔执行任务。

from apscheduler.triggers.interval import IntervalTrigger
trigger = IntervalTrigger(seconds=10)  # 每隔10秒触发一次

2. CronTrigger（Cron表达式触发器）

CronTrigger使用Cron表达式来定义任务的执行时间。Cron表达式是一种时间表达式，可以非常精确地定义任务执行时间。

关于crontab可以看这篇文章

使用Crontab：在Linux中自动化任务调度的完全指南

from apscheduler.triggers.cron import CronTrigger
trigger = CronTrigger(second='0,30')  # 每分钟的0秒和30秒触发

3. DateTrigger（指定日期触发器）

DateTrigger在特定日期和时间执行任务。

from apscheduler.triggers.date import DateTrigger
from datetime import datetime
trigger = DateTrigger(datetime(2023, 9, 1, 10, 30))  # 在2023年9月1日10:30执行任务

4. DailyTrigger（每天触发器）

DailyTrigger每天的特定时间执行任务。

from apscheduler.triggers.daily import DailyTrigger
trigger = DailyTrigger(hour=12, minute=0)  # 每天中午12:00执行任务

5. WeeklyTrigger（每周触发器）

WeeklyTrigger每周的特定日期和时间执行任务。

from apscheduler.triggers.weekly import WeeklyTrigger
trigger = WeeklyTrigger(day_of_week='mon', hour=8, minute=30)  # 每周一早上8:30执行任务

6. MonthlyTrigger（每月触发器）

MonthlyTrigger每月的特定日期和时间执行任务。

from apscheduler.triggers.monthly import MonthlyTrigger
trigger = MonthlyTrigger(day=15, hour=14, minute=45)  # 每月15号下午2:45执行任务

这些触发器提供了广泛的灵活性，可以满足各种定时任务的需求。可以根据具体的任务执行规则选择适当的触发器类型，并将触发器与任务关联起来，以定义任务何时执行。

调度器

1. 阻塞调度器（BlockingScheduler）

阻塞调度器是一种同步调度器，它会阻塞主程序的执行，直到所有的定时任务执行完成。这意味着当一个任务正在执行时，其它任务无法同时执行。这通常用于简单的脚本和小型应用程序中，其中任务的执行不会占用太多时间。

使用示例：

from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

# 添加任务
scheduler.add_job(job_function, trigger='interval', seconds=10)

# 启动调度器
scheduler.start()

2. 非阻塞调度器（BackgroundScheduler）

非阻塞调度器是一种异步调度器，它不会阻塞主程序的执行。它使用多线程或多进程来并行执行定时任务，因此可以同时处理多个任务。这对于需要执行耗时任务或在大型应用程序中使用APScheduler的情况非常有用，因为它不会阻塞应用的其它部分。

使用示例：

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

# 添加任务
scheduler.add_job(job_function, trigger='interval', seconds=10)

# 启动调度器
scheduler.start()