当前位置：物联沃-IOTWORD物联网 > 技术教程 > Python定时任务管理利器APScheduler详解
代码收藏家代码收藏家 技术教程

Python定时任务管理利器APScheduler详解

APScheduler 是一个用于 Python 的定时任务调度库。它提供了多种方式来调度和执行任务，包括日期、时间间隔、计划时间以及任意的 Python 代码。

APScheduler 的主要特点包括：

  1. 简单易用：APScheduler 提供了简单的 API，可以轻松地创建和管理定时任务。

  2. 灵活的任务调度方式：APScheduler 支持多种任务调度方式，包括日期和时间间隔调度，计划时间调度以及任意的 Python 代码调度。

  3. 多种存储后端：APScheduler 支持多种存储后端，包括内存、数据库和 Redis 等，可以根据实际需要选择合适的后端作为任务调度的存储。

  4. 强大的可扩展性：APScheduler 支持自定义的触发器和作业存储，可以根据实际需求进行定制。

 一、安装
pip install apscheduler

下面是一个使用 APScheduler 创建定时任务的示例代码：

from apscheduler.schedulers.blocking import BlockingScheduler

def job():
    print("定时任务执行")

scheduler = BlockingScheduler()
scheduler.add_job(job, 'interval', seconds=5)  # 每隔5秒执行一次定时任务
scheduler.start()
 二、调度器

APScheduler 提供几种不同的调度器，每种调度器都有自己的特点和用途。BaseScheduler、BlockingScheduler 和 BackgroundScheduler 是 APScheduler 中的调度器选项。让我们来探索它们的异同：

BaseScheduler

BaseScheduler 是所有其他调度器类型        的基类。它提供了调度逻辑的核心实现，但不实现进入调度循环的方法。通常你不会直接使用 BaseScheduler，而是使用它的子类，这些子类提供了具体的调度方法。

在标准的 APScheduler 安装中，BaseScheduler 类中未实现的进入调度循环的方法包括 start、shutdown 等。继承 BaseScheduler 的子类需要实现这些方法，使其适应不同的运行环境和条件。

BlockingScheduler

BlockingScheduler 是 BaseScheduler 的一个子类，它在启动后会阻塞当前线程，直到显式地被关闭。在此期间，它将持续监听并运行已经安排的任务。

这种调度器适用于不需要运行其他Python代码的简单脚本或应用程序，因为这种情况下可以让 BlockingScheduler 占据主线程，它会管理和执行所有的计划任务。

以下是 BlockingScheduler 的一个示例：

from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

# 定义作业并添加到调度器
scheduler.add_job(my_job, 'interval', minutes=2)

# 启动调度器，阻塞当前主线程
scheduler.start()

BackgroundScheduler

BackgroundScheduler 同样是 BaseScheduler 的子类，它启动的时候不会阻塞当前线程。相反，它在后台运行，意味着你可以在同一个进程中运行其他代码。

BackgroundScheduler 是Web应用程序或复杂脚本的理想选择，它们需要同时处理其他任务（如响应Web请求），同时在后台运行定时任务。

以下是 BackgroundScheduler 的一个示例：

 

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

# 定义作业并添加到调度器
scheduler.add_job(my_job, 'interval', minutes=2)

# 启动调度器，不会阻塞主线程
scheduler.start()

# 主线程可以继续执行其他操作
do_other_stuff()
  • BaseScheduler 是基础的调度器类，提供核心调度功能，但不负责执行调度循环。
  • BlockingScheduler 是适合单任务应用程序的调度器，它将占据并阻塞主线程直到被关闭。
  • BackgroundScheduler 在后台运行，适合同时需要处理其他非定时任务功能的应用程序。

    • 选择哪种调度器取决于你的应用场景和对线程使用的需求。如果你的应用是多线程的或者是一个Web应用，那么 BackgroundScheduler 可能是更好的选择。如果你的应用仅需要运行调度任务，并且没有其他并行任务的需求，BlockingScheduler 或许更适合。

    三、触发器

    APScheduler 支持几种不同的触发器（trigger）类型，你可以使用它们来设定定时任务的执行计划。这里是四种主要触发器类型的例子：

    1. date trigger （一次性任务）

    Date触发器用于在特定日期和时间执行单次任务。当你想要在确定的时间点运行一次作业时，这是一个合适的选择。

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

scheduler = BlockingScheduler()

def my_job():
    print("执行单次任务: ", datetime.now())

# 10秒后执行一次任务
run_date = datetime.now() + timedelta(seconds=10)
scheduler.add_job(my_job, 'date', run_date=run_date)

scheduler.start()
  • run_date (datetime | str) – 指定任务应当被触发的精确时间。
  • timezone (datetime.tzinfo | str) – 为run_date指定时区。 
    # 示例：
scheduler.add_job(my_job, 'date', run_date=datetime(2021, 4, 30, 5, 30), timezone='UTC')

    • 2. interval trigger （固定时间间隔）

    Interval触发器用于以固定时间间隔执行任务。这适用于需要重复执行的周期性任务，比如每隔一段时间清理日志文件或检查更新。

    from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

def my_interval_job():
    print("周期性任务执行: ", datetime.now())

# 每隔3秒执行一次任务
scheduler.add_job(my_interval_job, 'interval', seconds=3)

scheduler.start()
  • weeks (int) – 设置每隔几周触发任务。
  • days (int) – 设置每隔几天触发任务。
  • hours (int) – 设置每隔几小时触发任务。
  • minutes (int) – 设置每隔几分钟触发任务。
  • seconds (int) – 设置每隔几秒触发任务。
  • start_date (datetime | str) – 设置开始执行任务的时间。
  • end_date (datetime | str) – 设置结束执行任务的时间。
  • timezone (datetime.tzinfo | str) – 指定时区。
    • # 示例：
scheduler.add_job(my_job, 'interval', hours=2, start_date='2021-04-30 05:00:00', timezone='UTC')

     

    3. cron trigger （Cron 风格）

    Cron触发器允许你使用cron语法来计划任务，为需要在特定时间执行的任务提供了最大的灵活性。

    from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

def my_cron_job():
    print("Cron任务执行: ", datetime.now())

# 每天的11:30执行任务
scheduler.add_job(my_cron_job, 'cron', hour=11, minute=30)

scheduler.start()
  • year (int | str) – 年。
  • month (int | str) – 月。
  • day (int | str) – 天。
  • week (int | str) – 周数。
  • day_of_week (int | str) – 周内的第几天，可以是0-6或者mon,tue,wed,thu,fri,sat,sun的组合。
  • hour (int | str) – 小时，0-23之间的数。
  • minute (int | str) – 分钟，0-59之间的数。
  • second (int | str) – 秒， 0-59之间的数。
  • start_date (datetime | str) – 设置开始执行任务的时间。
  • end_date (datetime | str) – 设置结束执行任务的时间。
  • timezone (datetime.tzinfo | str) – 指定时区。
    • # 示例：
scheduler.add_job(my_job, 'cron', day_of_week='mon-fri', hour=5, minute=30, timezone='UTC')

    4. combine trigger （组合触发器）

    Combine触发器实际上是一个并集操作，用于组合多个触发器。它可以在任务执行符合任意一个内部触发器条件时被触发。

    from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers import AndTrigger, OrTrigger, IntervalTrigger, CronTrigger

scheduler = BlockingScheduler()

def my_combine_job():
    print("组合触发器任务执行: ", datetime.now())

trigger1 = IntervalTrigger(seconds=10)
trigger2 = CronTrigger(day_of_week='mon-fri', hour=17, minute=0)
combine_trigger = OrTrigger([trigger1, trigger2])

scheduler.add_job(my_combine_job, combine_trigger)

scheduler.start()

    # 示例：
trigger1 = IntervalTrigger(seconds=10)
trigger2 = CronTrigger(hour=10, minute=30)
combine_trigger = AndTrigger([trigger1, trigger2])  # 需要同时满足两个条件才触发
combine_trigger = OrTrigger([trigger1, trigger2])   # 满足任一条件即触发

    作者：我姓曹，谢谢

    物联沃分享整理
    物联沃-IOTWORD物联网 » Python定时任务管理利器APScheduler详解

    代码收藏家 普通

    分享到：

    相关推荐

    发表回复