APScheduler:强大的Python定时任务调度器

安装

使用pip安装APScheduler

pip install apscheduler

基本概念

APScheduler有四种组件:

  • Triggers:包含调度逻辑,每个作业有其专属触发器,决定下次运行时间。触发器无状态,仅依据初始配置工作。
  • Job Stores:存储预定作业。默认存储于内存,但支持多种数据库存储。作业数据在持久化存储时序列化,加载时反序列化。非默认存储不保存作业数据于内存,而是作为后端存储、加载、更新和搜索作业的桥梁。注意,作业存储不可在调度器间共享。
  • Executors:负责执行作业,通常通过线程或进程池提交可调用对象。作业完成时,执行器通知调度器,触发相应事件。
  • Schedulers:连接其他组件。应用中通常运行一个调度器。开发者通常不直接处理作业存储、执行器或触发器,而是通过调度器接口管理。

选择合适的组件

调度器

选择调度器主要依据编程环境和用途:

  • BlockingScheduler:适用于调度程序为进程唯一运行时。
  • BackgroundScheduler:适用于无特定框架且希望调度器后台运行的应用。
  • AsyncIOScheduler:适用于使用 asyncio 模块的应用。
  • GeventScheduler:适用于使用 gevent 的应用。
  • TornadoScheduler:适用于构建 Tornado 应用。
  • TwistedScheduler:适用于构建 Twisted 应用。
  • QtScheduler:适用于构建 Qt 应用。

作业存储

选择作业存储需考虑作业是否需要持久性。若作业在应用启动时重建,可使用默认值 MemoryJobStore。若需作业在调度器重启或应用崩溃后保持不变,推荐在编程环境支持的数据库中选择,如 SQLAlchemyJobStore(基于 PostgreSQL)。

执行器

若使用上述框架之一,执行器通常已选定。否则,默认的 ThreadPoolExecutor 适用于大多数场景。CPU 密集型操作应考虑 ProcessPoolExecutor,甚至可同时使用两者,将进程池执行器作为辅助。

触发器

触发器决定作业运行逻辑,有三种类型:

  • Date:一次性运行作业。

    from apscheduler.schedulers.background import BackgroundScheduler  
    from apscheduler.triggers.date import DateTrigger  
    from datetime import datetime, timedelta  
    import time  def my_job():  print("在指定时间执行一次任务...")  # 创建调度器实例  
    scheduler = BackgroundScheduler()  # 获取当前时间,并加上30秒作为任务执行时间  
    run_date = datetime.now() + timedelta(seconds=30)  # 添加一个在指定时间执行一次的任务  
    scheduler.add_job(my_job, DateTrigger(run_date=run_date))  # 启动调度器  
    scheduler.start()  try:  # 保持主线程运行,直到任务执行完毕  while True:  time.sleep(2)  # 检查任务是否已执行完毕,并关闭调度器  if not scheduler.get_jobs():  break  
    except (KeyboardInterrupt, SystemExit):  # 关闭调度器  scheduler.shutdown()
    
  • Interval:固定时间间隔运行作业。

    from apscheduler.schedulers.background import BackgroundScheduler  
    import time  def my_job():  print("每隔10秒执行一次任务...")  # 创建调度器实例  
    scheduler = BackgroundScheduler()  # 添加一个每隔10秒执行一次的任务  
    scheduler.add_job(my_job, 'interval', seconds=10)  # 启动调度器  
    scheduler.start()  try:  # 保持主线程运行  while True:  time.sleep(2)  
    except (KeyboardInterrupt, SystemExit):  # 关闭调度器  scheduler.shutdown()
    
  • cron:定期在特定时间运行作业。

    from apscheduler.schedulers.background import BackgroundScheduler  
    from apscheduler.triggers.cron import CronTrigger  
    import time  def my_job():  print("每天凌晨2点执行一次任务...")  # 创建调度器实例  
    scheduler = BackgroundScheduler()  # 添加一个每天凌晨2点执行一次的任务  
    scheduler.add_job(my_job, CronTrigger(hour=2, minute=0))  # 启动调度器  
    scheduler.start()  try:  # 保持主线程运行  while True:  time.sleep(2)  
    except (KeyboardInterrupt, SystemExit):  # 关闭调度器  scheduler.shutdown()
    

可组合多个触发器,在参与触发器均同意或任一触发时触发。

启动scheduler

调用 start() 启动调度器。对于 BlockingScheduler,需在初始化步骤后调用;其他调度器调用后立即返回,继续应用初始化。

添加jobs

添加作业有两种方法:

  • add_job():最常见方法。

  • scheduled_job():适用于不变作业。

add_job() 返回一个 Job 实例,用于修改或删除作业。在持久存储中安排作业时,需定义显式 ID 并使用 replace_existing=True,避免每次应用重启时创建新作业副本。

[!NOTE]

如果在应用程序初始化期间在持久作业存储中安排作业,则必须为作业定义一个显式ID并使用replace_existeting=True,否则每次应用程序重新启动时都会得到作业的新副本!

删除jobs

删除作业有两种方法:

  • remove_job(job_id, jobstore=None):使用作业 ID 和作业存储别名。

  • Job.remove():对 add_job() 返回的 Job 实例调用。

作业计划结束时自动删除。

示例:

job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()

或:

scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
scheduler.remove_job('my_job_id')

暂停和恢复jobs

通过 Job 实例或调度器本身暂停和恢复作业。暂停时,清除下次运行时间,恢复前不计算运行时间。

  • Job.pause() / BaseScheduler.pause_job():暂停作业。
  • Job.resume() / BaseScheduler.resume_job():恢复作业。

修改jobs

调用 Job.modify()scheduler.modify_job() 修改作业属性(除 ID 外)。要更改触发器,使用 Job.reschedule()scheduler.reschedule_job()

示例:

job.modify(max_instances=6, name='Alternate name')

或:

scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')

关闭scheduler

要关闭调度程序,请执行以下操作:

scheduler.shutdown()

默认情况下,调度程序会关闭其作业存储和执行器,并等待所有当前正在执行的作业完成。如果不想等待:

scheduler.shutdown(wait=False)

这仍将关闭作业存储和执行器,但不会等待任何正在运行的任务完成。

暂停/恢复job processing

暂停作业处理:

scheduler.pause()

恢复作业处理:

scheduler.resume()

也可以在暂停状态下启动调度器:

scheduler.start(paused=True)

当您需要在不需要的作业有机会运行之前对其进行修剪时,这很有用。

限制作业并发执行实例的数量

默认每个作业仅同时运行一个实例。使用 max_instances 关键字参数设置并发运行实例数。

scheduler.add_job(myfunc, 'interval', minutes=2, max_instances=10)

Scheduler事件

将事件监听器附加到调度器,监听特定事件。使用 add_listener() 提供掩码参数监听特定类型事件。

def my_listener(event):if event.exception:print('The job crashed :(')else:print('The job worked :)')scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

故障排除

若调度器未按预期工作,将 apscheduler 记录器日志级别提升至 DEBUG。

import logginglogging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/pingmian/58469.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

电子电气架构 --- 车载芯片现状

我是穿拖鞋的汉子,魔都中坚持长期主义的汽车电子工程师。 老规矩,分享一段喜欢的文字,避免自己成为高知识低文化的工程师: 所有人的看法和评价都是暂时的,只有自己的经历是伴随一生的,几乎所有的担忧和畏惧…

我docker拉取mysql镜像时用的是latest,我该怎么看我的镜像版本是多少?可以通过一下三种方法查看

我docker拉取mysql镜像时用的是latest,我该怎么看我的镜像版本是多少? 方法一:查看 Docker 镜像标签 你可以查看 Docker 镜像的标签信息,了解当前使用的 MySQL 镜像版本。 具体步骤如下: 1. 列出本地 Docker 镜像&am…

识别风险的提示清单

在PMBOK(《项目管理体系指南》)中,介绍了这样一个概念,在识别风险时,可以提供一个参考清单,让大家基于一个清单来思考,这个项目可能有哪些具体的风险。 PMBOK中的风险提示清单,包括…

【网络安全】从NA到P1,我是如何扩大思路的?

未经许可,不得转载。 文章目录 正文正文 在这篇文章中,我将向大家展示一个我发现的漏洞,该漏洞利用了一个硬编码的 Basic 认证头,获取了管理员权限。 首先,假设公司域名为“target.com”。 第一步是使用多种工具(如 Amass、subfinder、findomain 等)收集与该组织相关…

Unreal Engine 5 C++(C#)开发:使用蓝图库实现插件(二)编辑BPLibrary.h中的枚举及结构体

目录 引言 一、头文件编写 1.1Kismet/BlueprintFunctionLibrary.h 1.2BPLibrary.generated.h的作用 1.3IImageWrapper.h 1.4 IImageWrapperModule.h 1.5 Engine/Texture2D.h 1.6CoreMinimal.h 二、定义图片/路径类型的枚举 2.1图片枚举类EImageType 2.2路径枚举类EPath…

Vuestic 整理使用

简单示例 1. 条件渲染 2. 列表渲染 3. 组件插槽 4. 插值语法 5. 前后端路由的区别(还是转一下,可以减少代码量)SFC 构建 … … Okay,可以干活了,通顺 数据表的操作更加简化了 数据类别通过后端路由区别,但是还得由前端路由转一下 简单了许多呀,上脚手…

ssm017网上花店设计+vue(论文+源码)_kaic

设计题目:网上花店的设计与实现 摘 要 网络技术和计算机技术发展至今,已经拥有了深厚的理论基础,并在现实中进行了充分运用,尤其是基于计算机运行的软件更是受到各界的关注。加上现在人们已经步入信息时代,所以对于信…

网页上视频没有提供下载权限怎么办?

以腾讯会议录屏没有提供下载权限为例,该怎么办呢? 最好的办法就是找到管理员,开启下载权限。如果找不到呢,那就用这个办法下载。 1.打开Microsoft Edge浏览器的扩展 2.搜索“视频下载”,选择“视频下载Pro” 3.点击“…

「Mac畅玩鸿蒙与硬件5」鸿蒙开发环境配置篇5 - 熟悉 DevEco Studio 界面

本篇将详细介绍 DevEco Studio 的界面布局和主要功能模块,帮助开发者熟悉开发环境,提高开发效率。通过了解各个界面区域,开发者可以顺利找到所需工具,流畅开展鸿蒙应用开发。 关键词 DevEco Studio界面布局功能模块开发环境 一、…

理解 Vue 的单向数据流

引言 Vue.js 是一个渐进式的 JavaScript 框架,以其简洁和高效的特性受到广大开发者的喜爱。在 Vue 应用中,数据流的设计是其核心概念之一。Vue 采用的是单向数据流(One-Way Data Flow),这种设计模式使得数据流动更加清…

使用pyecharts绘制词云图

一、什么是词云图? 词云图是一种用来展现高频关键词的可视化表达,通过文字、色彩、图形的搭配,产生有冲击力地视觉效果,而且能够传达有价值的信息。 制作词云图的网站有很多,简单方便,适合小批量操作。 …

一、ARMv8寄存器之通用、状态、特殊寄存器

ARMV8核心寄存器数量是非常大的,为了更好的学习,可以划分为以下几大类: 通用寄存器。这类寄存器主要是用来暂存数据和参与运算。通过load\store指令操作。状态寄存器。AArch64体系结构使用PSTATE寄存器表示当前处理器状态。特殊寄存器。有专门…

云渲染突破酒店3D动画渲染速度与成本瓶颈!

3D动画已经成为众多行业,尤其是酒店业,用于营销和展示其独特卖点的重要工具。通过生动的3D动画,酒店能够突出其特色和优势,从而吸引更多潜在客户。然而,在3D动画制作过程中,渲染环节往往是一个耗时且技术要…

Mybatis-Plus中 updateById 无法将已有值的字段更新为 null

在 MyBatis-Plus 中,使用updateById,null字段并不会更新,其实是和更新的策略有关,当然,也有插入策略。 1、调整全局策略(会对所有的字段都忽略判断,如果一些字段不想要修改,但是传值…

HTML入门教程23:HTML脚本

一、嵌入JavaScript的方式 在HTML中嵌入JavaScript脚本有三种主要方式&#xff1a; 内联脚本&#xff1a;直接在HTML文档的<script>标签中编写JavaScript代码。外部脚本&#xff1a;将JavaScript代码保存在独立的.js文件中&#xff0c;并在HTML文档中通过<script>…

LabVIEW偏振调制激光高精度测距系统

在航空航天、汽车制造、桥梁建筑等先进制造领域&#xff0c;许多大型零件的装配精度要求越来越高&#xff0c;传统的测距方法在面对大尺寸、高精度测量时&#xff0c;难以满足工业应用的要求。绝对测距技术在大尺度测量上往往会因受环境影响大、测距精度低而无法满足需求。基于…

微信小程序如何实现地图轨迹回放?

要在Uni-app中实现微信小程序的地图轨迹回放功能&#xff0c;你可以按照以下步骤进行操作&#xff1a; 在Uni-app项目中引入地图组件&#xff1a;在页面中使用uni-app提供的map组件&#xff0c;可以使用uni.createMapContext方法获取地图上下文对象&#xff0c;以便后续操作地图…

WPF+MVVM案例实战(六)- 自定义分页控件实现

文章目录 1、项目准备2、功能实现1、分页控件 DataPager 实现2、分页控件数据模型与查询行为3、数据界面实现 3、运行效果4、源代码获取 1、项目准备 打开项目 Wpf_Examples&#xff0c;新建 PageBarWindow.xaml 界面、PageBarViewModel.cs ,在用户控件库 UserControlLib中创建…

WPF+MVVM案例实战(十一)- 环形进度条实现

文章目录 1、运行效果2、功能实现1、文件创建与代码实现2、角度转换器实现3、命名空间引用3、源代码下载1、运行效果 2、功能实现 1、文件创建与代码实现 打开 Wpf_Examples 项目,在Views 文件夹下创建 CircularProgressBar.xaml 窗体文件。 CircularProgressBar.xaml 代码实…

CSS常见适配布局方式

在网页设计中&#xff0c;布局是确保内容按预期显示的关键部分。CSS 提供了多种布局方式&#xff0c;每种方式都有其特定的用途和优势。以下是您提到的五种布局方式的详细解释&#xff1a; 1. 流式布局&#xff08;百分比布局&#xff09; 概述&#xff1a; 流式布局&#xf…