AI 助理
文档备案控制台
开发者社区 开发与运维 文章 正文

Python----使用apscheduler模块设置定时任务

2022-05-25 1619
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： Python----使用apscheduler模块设置定时任务

一、安装

pip install apscheduler

二、ApScheduler 简介

1 APScheduler的组件

  • triggers：触发器

triggers包含任务执行的调度逻辑，决定任务按照什么逻辑进行定时执行

  • job stores；任务存储器

存储了调度任务

  • executors：执行器

用例执行任务的，包含线程池以及进程池等的创建和调用等等

  • schedulers：调度器

属于控制面，将其他几个方面组织起来的作用、

2 调度器的种类

调度器有以下几种常见类型，其中最常用的BackgroundScheduler，即非阻塞式，因为在一般情况下，定时任务都会在放到web服务中，如果使用阻塞式，则无法启动web服务，而使用非阻塞式，则将定时任务设定后，就不管了，继续执行后面的web服务，只要web服务在运行，定时任务就是一直有效的

  • BlockingScheduler: 阻塞式
  • BackgroundScheduler: 非阻塞式（后台运行）
  • AsyncIOScheduler: 当使用asyncio模块时使用
  • GeventScheduler: 当使用gevent模块时使用
  • TornadoScheduler: 构建Tornado应用时使用
  • TwistedScheduler: 构建Twisted应用时使用
  • QtScheduler: 构建Qt应用时使用

3 内置的触发器类型

  • date: 在某个时间点执行一次时使用
  • interval: 固定的时间间隔循环执行时使用
  • cron: 在一天中特定的时间点执行时使用
  • calendarinterval: 当想在在一天中特定时间点或以日历为基础的时间间隔内执行时使用

三、使用举例

这里jiu就以非阻塞式BackgroundScheduler调度器为例展开

1 使用date类型的触发器

如下，使用了三种设置日期和时间的方法

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 通过date 设置指定日期执行
    sched.add_job(do_func,trigger="date",run_date=date(2022,5,25),args=("张三丰",100))

    # 通过datetime，设置指定日期额指定时刻执行
    sched.add_job(do_func, trigger="date", run_date=datetime(2022, 5, 25,14,0,10), args=("张三丰", 100))

    # 直接使用文本的方式指定日期和时刻表
    sched.add_job(do_func, trigger="date", run_date="2022-05-25 14:0:20", args=("张三丰", 100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

执行结果如下,可以发现，第一个通过date指定日期的默认是0点执行，显然时间已经过了，不会执行，第二和第三个则在规定的时间点执行了，这里还需要注意的是，通过打印可以看出，main函数执行完成后，已经开始执行main函数下面的while循环打印语句了，而在执行循环的过程中，定时任务仍然生效，这就是非阻塞式调度器的原理，如果是阻塞式，则在此代码中，会一直卡在main函数中，main下面额while循环语句是不会执行的，因此在实际使用中，非阻塞式应用的是非常多的

2022-05-25 14:00:02
2022-05-25 14:00:02
C:\Users\hitre\.virtualenvs\zentaolinkgitlab-QCS5yxD9\lib\site-packages\apscheduler\util.py:436: PytzUsageWarning: The localize method is no longer necessary, as this time zone supports the fold attribute (PEP 495). For more details on migrating to a PEP 495-compliant implementation, see https://pytz-deprecation-shim.readthedocs.io/en/latest/migration.html
  return tzinfo.localize(dt)
Run time of job "do_func (trigger: date[2022-05-25 00:00:00 CST], next run at: 2022-05-25 00:00:00 CST)" was missed by 14:00:02.088260
2022-05-25 14:00:03
2022-05-25 14:00:04
2022-05-25 14:00:05
2022-05-25 14:00:06
2022-05-25 14:00:07
2022-05-25 14:00:08
2022-05-25 14:00:09
2022-05-25 14:00:10 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:00:10
2022-05-25 14:00:11
2022-05-25 14:00:12
2022-05-25 14:00:13
2022-05-25 14:00:14
2022-05-25 14:00:15
2022-05-25 14:00:16
2022-05-25 14:00:17
2022-05-25 14:00:18
2022-05-25 14:00:19
2022-05-25 14:00:20 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:00:20
2022-05-25 14:00:21
2022-05-25 14:00:22

2 使用interval类型的触发器

如下代码演示了时间间隔循环执行的使用例子

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 每3秒执行一次
    sched.add_job(do_func,trigger="interval",args=("张三丰",100),seconds=3)
    # 每3分钟执行一次
    sched.add_job(do_func, trigger="interval", args=("张三丰", 100), minutes=3)
    # 每3小时执行一次
    sched.add_job(do_func, trigger="interval", args=("张三丰", 100), hours=3)
    # 每3天执行一次
    sched.add_job(do_func, trigger="interval", args=("张三丰", 100), days=3)
    # 每3周执行一次
    sched.add_job(do_func, trigger="interval", args=("张三丰", 100), weeks=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

上面的代码中因为时间跨度比较大，这里只演示妹3秒执行一次的代码
代码如下：

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 每3秒执行一次
    sched.add_job(do_func,trigger="interval",args=("张三丰",100),seconds=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

执行结果如下：

2022-05-25 14:14:04
2022-05-25 14:14:04
2022-05-25 14:14:05
2022-05-25 14:14:06
2022-05-25 14:14:07 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:14:07
2022-05-25 14:14:08
2022-05-25 14:14:09
2022-05-25 14:14:10 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:14:10
2022-05-25 14:14:11
2022-05-25 14:14:12
2022-05-25 14:14:13 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:14:13

3 使用cron类型的触发器

cron的触发器有点类似linux上crontab定时器的使用，代码如下：

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 任务会在6月、7月、8月、11月和12月的第三个周五，00:00、01:00、02:00和03:00触发
    sched.add_job(do_func,trigger="cron",month='6-8,11-12', day='3rd fri', hour='0-3',args=("张三丰",100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

这里需要注意的是，可以省略不需要的字段。当省略时间参数时，在显式指定参数之前的参数会被设定为，之后的参数会被设定为最小值，week 和day_of_week的最小值为

day=1, minute=20
等同于
year='*', month='*', day=1, week='*', day_of_week='*', hour='*', minute=20, second=0

此外，也可以直接使用crontab表达式，如下：

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 任务会在6月、7月、8月、11月和12月的第三个周五，00:00、01:00、02:00和03:00触发
    sched.add_job(do_func,trigger=CronTrigger.from_crontab('48 10 1-15 sep-nov *'),args=("张三丰",100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

四、定时器使用装饰器的方法

以间隔时间循环执行的代码为例，如下为未使用装饰器的方式

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # 每3秒执行一次
    sched.add_job(do_func,trigger="interval",args=("张三丰",100),seconds=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

修改为使用装饰器的方式如下：

from apscheduler.schedulers.background import BackgroundScheduler
import time

sched=BackgroundScheduler()

@sched.scheduled_job(trigger="interval",args=("张三丰",100),seconds=3)
def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time()))+" in do func : 姓名："+name+" 年龄："+str(age))

if __name__=="__main__":
    sched.start()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

执行结果如下：

2022-05-25 14:34:10
2022-05-25 14:34:11
2022-05-25 14:34:12
2022-05-25 14:34:13 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:34:13
2022-05-25 14:34:14
2022-05-25 14:34:15
2022-05-25 14:34:16 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:34:16
2022-05-25 14:34:17
2022-05-25 14:34:18
2022-05-25 14:34:19 in do func : 姓名：张三丰 年龄：100
2022-05-25 14:34:19
2022-05-25 14:34:20
文章标签：
Java
Python
测试技术
Linux
调度
存储
关键词：
Python模块
Python定时任务
Python APScheduler
Python设置
redrose2100
目录
相关文章
站大爷
|
8月前
|
SQL 关系型数据库 数据库
Python SQLAlchemy模块：从入门到实战的数据库操作指南
免费提供Python+PyCharm编程环境，结合SQLAlchemy ORM框架详解数据库开发。涵盖连接配置、模型定义、CRUD操作、事务控制及Alembic迁移工具，以电商订单系统为例，深入讲解高并发场景下的性能优化与最佳实践，助你高效构建数据驱动应用。
站大爷
923 7
站大爷
|
8月前
|
监控 安全 程序员
Python日志模块配置：从print到logging的优雅升级指南
从 `print` 到 `logging` 是 Python 开发的必经之路。`print` 调试简单却难维护，日志混乱、无法分级、缺乏上下文；而 `logging` 支持级别控制、多输出、结构化记录，助力项目可维护性升级。本文详解痛点、优势、迁移方案与最佳实践，助你构建专业日志系统，让程序“有记忆”。
站大爷
675 0
站大爷
|
8月前
|
JSON 算法 API
Python中的json模块：从基础到进阶的实用指南
本文深入解析Python内置json模块的使用，涵盖序列化与反序列化核心函数、参数配置、中文处理、自定义对象转换及异常处理，并介绍性能优化与第三方库扩展，助你高效实现JSON数据交互。（238字）
站大爷
646 4
站大爷
|
8月前
|
存储 Java 调度
Python定时任务实战：APScheduler从入门到精通
APScheduler是Python强大的定时任务框架，通过触发器、执行器、任务存储和调度器四大组件，灵活实现各类周期性任务。支持内存、数据库、Redis等持久化存储，适用于Web集成、数据抓取、邮件发送等场景，解决传统sleep循环的诸多缺陷，助力构建稳定可靠的自动化系统。（238字）
站大爷
1372 1
站大爷
|
9月前
|
安全 大数据 程序员
Python operator模块的methodcaller：一行代码搞定对象方法调用的黑科技
`operator.methodcaller`是Python中处理对象方法调用的高效工具，替代冗长Lambda，提升代码可读性与性能。适用于数据过滤、排序、转换等场景，支持参数传递与链式调用，是函数式编程的隐藏利器。
站大爷
259 4
站大爷
|
8月前
|
Java 调度 数据库
Python threading模块：多线程编程的实战指南
本文深入讲解Python多线程编程，涵盖threading模块的核心用法：线程创建、生命周期、同步机制（锁、信号量、条件变量）、线程通信（队列）、守护线程与线程池应用。结合实战案例，如多线程下载器，帮助开发者提升程序并发性能，适用于I/O密集型任务处理。
站大爷
728 0
站大爷
|
8月前
|
XML JSON 数据处理
超越JSON：Python结构化数据处理模块全解析
本文深入解析Python中12个核心数据处理模块，涵盖csv、pandas、pickle、shelve、struct、configparser、xml、numpy、array、sqlite3和msgpack，覆盖表格处理、序列化、配置管理、科学计算等六大场景，结合真实案例与决策树，助你高效应对各类数据挑战。（238字）
站大爷
1049 0
站大爷
|
9月前
|
存储 数据库 开发者
Python SQLite模块：轻量级数据库的实战指南
本文深入讲解Python内置sqlite3模块的实战应用，涵盖数据库连接、CRUD操作、事务管理、性能优化及高级特性，结合完整案例，助你快速掌握SQLite在小型项目中的高效使用，是Python开发者必备的轻量级数据库指南。
站大爷
738 0
francisyu
|
10月前
|
存储 安全 数据处理
Python 内置模块 collections 详解
`collections` 是 Python 内置模块，提供多种高效数据类型，如 `namedtuple`、`deque`、`Counter` 等，帮助开发者优化数据处理流程，提升代码可读性与性能，适用于复杂数据结构管理与高效操作场景。
francisyu
584 0
游客fuyq5xtd5duvw
|
11月前
|
数据安全/隐私保护 Python
抖音私信脚本app，协议私信群发工具，抖音python私信模块
这个实现包含三个主要模块：抖音私信核心功能类、辅助工具类和主程序入口。核心功能包括登录
游客fuyq5xtd5duvw
959 0

热门文章

最新文章

  • 1
    Python复杂网络结构可视化——matplotlib+networkx
  • 2
    基于python大数据的用户行为数据分析系统
  • 3
    Python 基于 Django 的学生成绩管理系统，可视化界面+数据库（附源码，教程）
  • 4
    Python:快速去除PDF水印
  • 5
    python实战小项目——图片转换为字符画
  • 6
    Python调用百度API 实现地理位置经纬度坐标转换
  • 7
    python接口自动化7-参数关联
  • 8
    python字典
  • 9
    python模拟用户登录某某网
  • 10
    python使用遍历在列表（list）中添加字典（dict）
  • 1
    基于python大数据的台风灾害分析及预测系统
    657
  • 2
    基于Python大数据的热门游戏推荐系统
    594
  • 3
    基于python大数据的青少年网络使用情况分析及预测系统
    653
  • 4
    2026版基于python大数据的电影分析可视化系统
    604
  • 5
    基于Python大数据的的电商用户行为分析系统
    1392
  • 6
    基于python大数据技术的医疗数据分析与研究
    400
  • 7
    基于python大数据深度学习的酒店评论文本情感分析系统
    763
  • 8
    Python SQLAlchemy模块：从入门到实战的数据库操作指南
    923
  • 9
    基于python大数据的的海洋气象数据可视化平台
    393
  • 10
    基于Python大数据的主流汽车价格分析可视化系统
    773

    • 相关课程

    更多
  • Python爬虫实战
  • Python开发基础入门
  • Python常用数据科学库
  • Python网络爬虫实战
  • Python完全自学手册图文教程
  • Python基础快速入门实战教程

    • 相关电子书

    更多
  • From Python Scikit-Learn to Sc
  • Data Pre-Processing in Python:
  • 双剑合璧-Python和大数据计算平台的结合

    • 推荐镜像

    更多
  • python-release
    • 下一篇
    阿里云正式发布 Agentic 代码安全：AI驱动的双Agent协同引擎