投稿
  1. 极客之音首页
  2. 后端开发
  3. Python

Python-APScheduler介绍及使用

小半 • • Python • 阅读 52

世上唯一不能复制的是时间，唯一不能重演的是人生，唯一不劳而获的是年龄。该怎么走，过什么样的生活，全凭自己的选择和努力。人生很贵，请别浪费！与智者为伍，与良善者同行。Python-APScheduler介绍及使用，希望对大家有帮助，欢迎收藏，转发！站点地址：www.bmabk.com，来源：原文

在许多应用程序中，需要定期执行某些任务，例如发送电子邮件、生成报告或清理临时文件等。为了管理和调度这些任务，Python中有许多优秀的调度器库，其中APScheduler是一个流行且功能强大的选择。本文将介绍APScheduler的基本概念，并展示如何使用它来实现定时任务调度。

一、APScheduler简介

APScheduler是一个轻量级的Python任务调度库，提供了多种调度器（Scheduler）、执行器（executor）、触发器（Trigger）和作业存储（Job Store）等组件，可以满足各种定时任务调度需求。它的特点包括：

灵活的调度方式：支持间隔调度、定时调度、CRON表达式调度等多种调度方式。
多种触发器：提供了简单的时间触发器、日期触发器、CRON触发器等多种触发器类型。
可扩展的作业存储：支持内存、数据库等多种作业存储方式，方便管理和持久化存储作业信息。

  • 触发器 (trigger)
    包含调度逻辑，每一个作业有它自己的触发器，用于决定接下来哪一个作业会运行。除了他们自己初始配置意外，触发器完全是无状态的

    • APScheduler 有三种内建的 trigger:
    • date: 特定的时间点触发
    • interval: 固定时间间隔触发
    • cron: 在特定时间周期性地触发
  • 作业存储 (job store)
    存储被调度的作业，默认的作业存储是简单地把作业保存在内存中，其他的作业存储是将作业保存在数据库中。一个作业的数据讲在保存在持久化作业存储时被序列化，并在加载时被反序列化。调度器不能分享同一个作业存储。
    APScheduler 默认使用 MemoryJobStore，可以修改使用 DB 存储方案
  • 执行器 (executor)
    处理作业的运行，他们通常通过在作业中提交制定的可调用对象到一个线程或者进城池来进行。当作业完成时，执行器将会通知调度器。
    最常用的 executor 有两种：

    • ProcessPoolExecutor
    • ThreadPoolExecutor
  • 调度器 (scheduler) 
    from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
sched = BlockingScheduler()
sched.add_job(job, 'interval', seconds=5)
sched.start()

# 上述代码创建了一个 BlockingScheduler，并使用默认内存存储和默认执行器。
# 如果想要显式设置 job store(使用mongo存储)和 executor 可以这样写：
# (默认选项分别是 MemoryJobStore 和 ThreadPoolExecutor，其中线程池的最大线程数为10)。配置完成后使用 start() 方法来启动。
from datetime     import datetime
from pymongo import MongoClient
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
# MongoDB 参数
client = MongoClient('127.0.0.1', 27017)
# 输出时间
def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# 存储方式
jobstores = {
    'mongo': MongoDBJobStore(collection='job', database='test', client=client),
    'default': MemoryJobStore()
}
executors = {
    'default': ThreadPoolExecutor(10),
    'processpool': ProcessPoolExecutor(3)
}
job_defaults = {
    'coalesce': False,     # 默认情况下关闭新的作业
    'max_instances': 10     # 设置调度程序将同时运行的特定作业的最大实例数10
}
scheduler = BlockingScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)
scheduler.add_job(job, 'interval', seconds=5, jobstore='mongo')
scheduler.start()

二、安装APScheduler

在开始使用APScheduler之前，首先需要安装该库。

pip install apscheduler

三、使用示例

  • BlockingScheduler是APScheduler中的调度器，APScheduler 中有两种常用的调度器，

    • BlockingScheduler
      当调度器是应用中唯一要运行的任务时，使用 BlockingSchedule
    • BackgroundScheduler
      如果希望调度器在后台执行，使用BackgroundScheduler，例如配合web框架(diango、flask、fastapi)使用

  • 创建定时任务
    下面是一个简单的示例，演示了如何使用APScheduler创建一个每隔5秒执行一次的定时任务。

    """
interval参数说明
    weeks               (int)                   number of weeks to wait
    days                (int)                   number of days to wait
    hours               (int)                   number of hours to wait
    minutes             (int)                   number of minutes to wait
    seconds             (int)                   number of seconds to wait
    start_date          (datetime|str)          starting point for the interval calculation
    end_date            (datetime|str)          latest possible date/time to trigger on
    timezone            (datetime.tzinfo|str)   time zone to use for the date/time calculations
    --------------------------------------------------------------------------------
    weeks               (int)                   间隔几周
    days                (int)                   间隔几天
    hours               (int)                   间隔几小时
    minutes             (int)                   间隔几分钟
    seconds             (int)                   间隔多少秒
    start_date          (datetime 或 str)        开始日期
    end_date            (datetime 或 str)        结束日期
    timezone            (datetime.tzinfo 或str)  用于日期/时间计算的时区
"""
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

# 定义任务函数
def job_function():
    print("Hello World")

# 创建调度器
sched = BlockingScheduler()

# 安排每两小时调用一次
sched.add_job(job_function, 'interval', hours=2)
# 与之前相同，但于2010年10月10日9:30开始，于2014年6月15日11:00停止
sched.add_job(job_function, 'interval', hours=2, start_date='2010-10-10 09:30:00', end_date='2014-06-15 11:00:00')

# 启动调度器
sched.start()

  • 使用CRON表达式
    除了间隔调度外，APScheduler还支持使用CRON表达式来指定定时调度。

    """
cron参数说明
    year            (int|str) – 4-digit year
    month           (int|str) – month (1-12)
    day             (int|str) – day of the (1-31)
    week            (int|str) – ISO week (1-53)
    day_of_week     (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
    hour            (int|str) – hour (0-23)
    minute          (int|str) – minute (0-59)
    second          (int|str) – second (0-59)
    start_date      (datetime|str) – earliest possible date/time to trigger on (inclusive)
    end_date        (datetime|str) – latest possible date/time to trigger on (inclusive)
    timezone        (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone)
"""
from apscheduler.schedulers.blocking import BlockingScheduler

def job_function():
    print("Hello World")

# 创建调度器
sched = BlockingScheduler()

# 6月、7月、8月、11月和12月的00:00、01:00、02:00和03:00
sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
# 周一至周五上午5:30运行至2014-05-30 00:00:00
sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')
# 每天凌晨1点执行任务
scheduler.add_job(job_function, 'cron', hour=1, minute=0)

# 启动调度器
sched.start()

  • date任务
    最基本的一种调度，作业只会执行一次。

    """
date
    它的参数如下：
    run_date (datetime|str) – the date/time to run the job at
    timezone (datetime.tzinfo|str) – time zone for run_date if it doesn’t have one already
"""
from datetime import date
from apscheduler.schedulers.blocking import BlockingScheduler

def my_job(text):
    print(text)
sched = BlockingScheduler()
# 该作业将于2009年11月6日执行
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text'])
sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5), args=['text'])
sched.add_job(my_job, 'date', run_date='2009-11-06 16:30:05', args=['text'])
# “date”触发器和作为run_date的datetime.now（）是隐式的
sched.add_job(my_job, args=['text'])
sched.start()

  • 添加任务监听器
    可以添加任务监听器来监听任务的执行状态，例如任务开始、任务执行出错等。

    def my_listener(event):
    if event.exception:
        print('任务出错了！')

scheduler.add_listener(my_listener, 'job_error')

四、高级用法

  • 添加持久化存储
    可以将任务调度信息存储到持久化存储中，以便在应用重启后恢复调度状态。

    from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
scheduler = BlockingScheduler(jobstores=jobstores)

结语

通过本文的介绍，您已经了解了APScheduler的基本概念及如何使用它来实现定时任务调度。APScheduler提供了丰富的调度方式和灵活的配置选项，能够满足各种复杂的任务调度需求。在实际应用中，可以根据具体情况选择合适的调度方式和配置选项，并根据需要添加任务监听器和持久化存储，以提高任务调度的可靠性和可维护性。

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/249006.html

赞 (0)
小半的头像小半
0 0

相关推荐

极客之音——专业性很强的中文编程技术网站，欢迎收藏到浏览器，订阅我们！