Python高效定时任务处理APScheduler库深入学习

更新时间：2023年08月14日 11:08:05 作者：子午

这篇文章主要为大家介绍了Python高效定时任务处理APScheduler库深入学习,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪

介绍

APScheduler是Python中一个强大的第三方库，用于在后台执行定时任务。它允许我们根据设定的时间间隔、日期规则或特定时间来执行任务，适用于定时执行脚本、定时发送邮件、定时处理数据等场景。APScheduler的功能使得在Python中实现定时任务变得非常简单和高效。本文将从入门到精通地介绍APScheduler库的使用方法，带你掌握在Python中实现定时任务的技巧。

安装和导入
创建定时任务
定时任务触发器
任务存储
并发执行
阻塞和非阻塞
错误处理
立即执行任务
调度器持久化
任务监听器
移除定时任务

1. 安装和导入

首先，我们需要安装APScheduler库。可以使用pip命令进行安装：

pip install apscheduler

安装完成后，我们可以在Python代码中导入APScheduler：

from apscheduler.schedulers.background import BackgroundScheduler

2. 创建定时任务

APScheduler提供了BackgroundScheduler和BlockingScheduler两种类型的调度器，用于创建定时任务。BackgroundScheduler在后台运行，不会阻塞主线程；而BlockingScheduler会阻塞主线程直到所有任务完成。

from apscheduler.schedulers.background import BackgroundScheduler
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每隔5秒执行一次
scheduler.add_job(job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
# 主线程等待一段时间后结束
time.sleep(20)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们首先创建了一个后台调度器scheduler，然后定义了一个名为job的任务函数，在其中打印当前时间。使用scheduler.add_job()添加了一个定时任务，设置为每隔5秒执行一次。然后，我们启动了调度器scheduler，让定时任务在后台执行。主线程等待20秒后结束，并调用scheduler.shutdown()关闭调度器。

3. 定时任务触发器

APScheduler提供了多种触发器类型，用于设置定时任务的触发条件。

interval触发器：按照设定的时间间隔来触发任务。

from apscheduler.schedulers.background import BackgroundScheduler
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每隔5秒执行一次
scheduler.add_job(job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
# 主线程等待一段时间后结束
time.sleep(20)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们使用'interval'触发器，设置任务每隔5秒执行一次。

cron触发器：使用类似于Linux中cron表达式的规则来触发任务，可以精确到秒。

from apscheduler.schedulers.background import BackgroundScheduler
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每天的13点30分触发任务
scheduler.add_job(job, 'cron', hour=13, minute=30)
# 启动调度器
scheduler.start()
# 主线程等待一段时间后结束
time.sleep(60)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们使用'cron'触发器，设置任务每天的13点30分触发。

date触发器：在指定的时间点触发任务。

from apscheduler.schedulers.background import BackgroundScheduler
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，设置任务在2023年7月31日10点30分触发
scheduler.add_job(job, 'date', run_date='2023-07-31 10:30:00')
# 启动调度器
scheduler.start()
# 主线程等待一段时间后结束
time.sleep(60)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们使用'date'触发器，设置任务在2023年7月31日10点30分触发。

4. 任务存储

APScheduler支持将任务存储在不同的后端存储中，如内存、数据库等。默认情况下，任务是存储在内存中的。

from apscheduler.schedulers.background import BackgroundScheduler
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每隔5秒执行一次
scheduler.add_job(job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
# 主线程等待一段时间后结束
time.sleep(20)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们使用默认的内存存储来存储任务。

如果需要将任务存储在数据库中，可以使用jobstores参数来设置。

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 创建数据库存储
jobstores = {
 'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每隔5秒执行一次
scheduler.add_job(job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
# 主线程等待一段时间后结束
time.sleep(20)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们使用了SQLAlchemyJobStore来将任务存储在SQLite数据库中。

5. 并发执行

默认情况下，APScheduler会将任务串行执行，也就是说一个任务结束后才会执行下一个任务。如果希望并发执行多个任务，可以使用max_instances参数来设置。

from apscheduler.schedulers.background import BackgroundScheduler
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 定义任务函数
def job(index):
 print(f"定时任务{index}执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每隔5秒执行一次，最多并发3个任务
scheduler.add_job(job, 'interval', seconds=5, args=[1], max_instances=3)
scheduler.add_job(job, 'interval', seconds=5, args=[2], max_instances=3)
scheduler.add_job(job, 'interval', seconds=5, args=[3], max_instances=3)
# 启动调度器
scheduler.start()
# 主线程等待一段时间后结束
time.sleep(20)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们使用了args参数传递参数给任务函数，并使用max_instances参数设置最多并发3个任务。

6. 阻塞和非阻塞

APScheduler提供了阻塞和非阻塞两种调度器类型。

阻塞调度器：在调度器启动后，会阻塞主线程直到所有任务完成。

from apscheduler.schedulers.blocking import BlockingScheduler
import time
# 创建阻塞调度器
scheduler = BlockingScheduler()
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每隔5秒执行一次
scheduler.add_job(job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
print("主线程结束")

非阻塞调度器：在调度器启动后，不会阻塞主线程。

from apscheduler.schedulers.background import BackgroundScheduler
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每隔5秒执行一次
scheduler.add_job(job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
# 主线程等待一段时间后结束
time.sleep(20)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们分别使用BlockingScheduler和BackgroundScheduler创建了阻塞和非阻塞调度器。

7. 错误处理

在任务执行过程中，可能会出现异常。APScheduler提供了异常处理机制，我们可以通过try...except...捕获任务函数中的异常，并进行相应的处理。

from apscheduler.schedulers.background import BackgroundScheduler
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 定义任务函数
def job():
 try:
     print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
     # 抛出一个异常
     raise ValueError("任务出现异常")
 except Exception as e:
     print("任务执行过程中发生异常：", str(e))
     # 添加定时任务，每隔5秒执行一次
scheduler.add_job(job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
# 主线程等待一段时间后结束
time.sleep(20)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们在任务函数中抛出了一个ValueError异常，并通过try...except...捕获并输出了异常信息。

8. 立即执行任务

有时候我们可能需要立即执行一个任务，而不是等到下次触发时间。APScheduler提供了run_job方法来立即执行任务。

from apscheduler.schedulers.background import BackgroundScheduler
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每隔5秒执行一次
scheduler.add_job(job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
# 立即执行任务
scheduler.run_job(job)
# 主线程等待一段时间后结束
time.sleep(20)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们使用scheduler.run_job(job)方法立即执行了任务。

9. 调度器持久化

在实际应用中，我们可能需要将调度器的配置保存到文件中，以便在下次启动时恢复。

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
import time
# 创建数据库存储
jobstores = {
 'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
# 创建后台调度器，并指定jobstores参数
scheduler = BackgroundScheduler(jobstores=jobstores)
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每隔5秒执行一次
scheduler.add_job(job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
# 主线程等待一段时间后结束
time.sleep(20)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们创建了一个数据库存储jobstores，并在创建后台调度器时指定了jobstores参数。这样，在调度器运行过程中，任务的配置将会被持久化到数据库中。

10. 任务监听器

APScheduler提供了任务监听器，用于监听任务的状态变化。我们可以通过add_listener方法添加监听器，并在任务状态发生变化时进行相应的处理。

from apscheduler.schedulers.background import BackgroundScheduler
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每隔5秒执行一次
scheduler.add_job(job, 'interval', seconds=5)
# 定义任务监听器
def my_listener(event):
 if event.exception:
     print("任务执行过程中发生异常：", str(event.exception))
 else:
     print("任务执行成功")
     # 添加任务监听器
scheduler.add_listener(my_listener, mask='all')
# 启动调度器
scheduler.start()
# 主线程等待一段时间后结束
time.sleep(20)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们创建了一个任务监听器my_listener，并在任务执行过程中通过if...else...判断是否出现异常。然后通过scheduler.add_listener(my_listener, mask='all')方法添加了监听器。

11. 移除定时任务

如果我们希望在调度器运行过程中移除某个定时任务，可以使用scheduler.remove_job(job_id)方法。

from apscheduler.schedulers.background import BackgroundScheduler
import time
# 创建后台调度器
scheduler = BackgroundScheduler()
# 定义任务函数
def job():
 print("定时任务执行：", time.strftime("%Y-%m-%d %H:%M:%S"))
 # 添加定时任务，每隔5秒执行一次，并获取任务ID
job_id = scheduler.add_job(job, 'interval', seconds=5).id
# 启动调度器
scheduler.start()
# 主线程等待一段时间后移除定时任务
time.sleep(10)
scheduler.remove_job(job_id)
# 主线程等待一段时间后结束
time.sleep(10)
# 关闭调度器
scheduler.shutdown()
print("主线程结束")

在上述代码中，我们通过scheduler.add_job(job, 'interval', seconds=5).id获取了定时任务的ID，并使用scheduler.remove_job(job_id)移除了定时任务。

总结

通过本文的介绍，我们学习了APScheduler库的基本用法，包括创建定时任务、定时任务触发器、任务存储、并发执行、阻塞和非阻塞调度器、错误处理、立即执行任务、调度器持久化、任务监听器和移除定时任务等。APScheduler为Python开发者提供了一个强大的定时任务调度框架，使得在Python中实现定时任务变得非常简单和高效。掌握APScheduler的使用将为我们的项目和程序带来很大的便利。

以上就是Python高效定时任务处理APScheduler库深入学习的详细内容，更多关于Python APScheduler定时任务的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

pytorch实现图像识别(实战)
这篇文章主要介绍了pytorch实现图像识别(实战),文章主要分享实现代码，但也具有一定的参考价值，需要的小伙伴可以才可以一下，希望对你有所帮助
2022-02-02
Python是怎样处理json模块的
这篇文章主要介绍了Python是怎样处理json模块的，文中讲解非常细致，代码帮助大家更好的理解和学习，感兴趣的朋友可以了解下
2020-07-07
Python 获得命令行参数的方法(推荐)
本篇将介绍python中sys, getopt模块处理命令行参数的方法，本文给大家介绍的非常详细，具有参考借鉴价值，需要的朋友参考下吧
2018-01-01
python快速安装OpenCV的步骤记录
这篇文章主要给大家介绍了关于python快速安装OpenCV的相关资料，文中通过图文介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2021-02-02
Python一键安装全部依赖包的方法
今天小编就为大家分享一篇Python一键安装全部依赖包的方法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2019-08-08
Python+OpenCV实现角度测量的示例代码
本文介绍如何使用python语言实现角度测量，程序包括鼠标选点、直线斜率计算、角度计算三个子程序和一个主程序，感兴趣的可以了解一下
2022-03-03
python筛选出两个文件中重复行的方法
这篇文章主要为大家详细介绍了python筛选出两个文件中重复行的方法，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2018-05-05
详解Python使用apscheduler定时执行任务
在平常的工作中几乎有一半的功能模块都需要定时任务来推动，例如项目中有一个定时统计程序，定时爬出网站的URL程序，定时检测钓鱼网站的程序等等，都涉及到了关于定时任务的问题，所以就找到了python的定时任务模块
2022-03-03
使用python在本地电脑上快速处理数据
这篇文章主要介绍了使用python在本地电脑上快速处理数据的相关资料,需要的朋友可以参考下
2017-06-06
python 批量修改 labelImg 生成的xml文件的方法
这篇文章主要介绍了python 批量修改 labelImg 生成的xml文件的方法，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-09-09