当前位置：首页 > news >正文

【Python】 Python Schedule 模块：轻量级的定时任务调度库

news 来源：原创 2024/9/20 16:37:21

我听见有人猜
你是敌人潜伏的内线
和你相知多年
我确信对你的了解
你舍命救我画面
一一在眼前浮现
司空见惯了鲜血
你忘记你本是娇娆的红颜
感觉你我彼此都那么依恋
🎵 许嵩《内线》

在日常开发中，我们经常需要执行一些定时任务，例如定时备份数据库、定时发送邮件提醒、定时抓取数据等。在 Python 中，虽然可以使用传统的 threading 模块结合 time.sleep() 来实现简单的定时任务，但它的灵活性和可读性较差。如果你需要一个轻量级且易于使用的调度库，那么 schedule 模块将是一个不错的选择。

1. 什么是 schedule？

schedule 是一个非常简洁、轻量的 Python 库，它的主要功能就是用于设置定时任务。通过几行代码，我们就可以实现定时执行函数，并且支持多种时间间隔（如秒、分钟、小时、天、周等），同时其 API 设计非常直观，适合快速开发应用中的简单调度任务。

2. 安装 schedule

在开始之前，我们首先需要安装这个模块。你可以使用 pip 来安装：

pip install schedule

安装完成后，你就可以在你的 Python 代码中引入 schedule 模块，并开始使用它来调度任务。

3. 基本使用

schedule 的使用非常简单，核心的思想是通过 do() 方法来定义任务，然后用时间调度方法如 every() 来定义任务的执行频率。最后，通过 run_pending() 进行任务的检查与执行。

3.1 简单任务调度示例

下面是一个简单的例子，每隔 10 秒执行一次任务：

import schedule
import timedef job():print("任务执行中...")# 每隔 10 秒执行一次任务
schedule.every(10).seconds.do(job)while True:# 检查是否有任务需要执行schedule.run_pending()time.sleep(1)

在这个例子中，我们定义了一个 job() 函数，这个函数会被调度为每隔 10 秒执行一次。schedule.every(10).seconds.do(job) 用于设置任务的时间间隔，而 schedule.run_pending() 用于每次循环中检查并运行等待执行的任务。

3.2 多个任务调度

schedule 支持同时调度多个任务，下面是一个例子：

import schedule
import timedef job1():print("任务 1 执行中...")def job2():print("任务 2 执行中...")# 每 5 秒执行一次 job1
schedule.every(5).seconds.do(job1)# 每 10 秒执行一次 job2
schedule.every(10).seconds.do(job2)while True:schedule.run_pending()time.sleep(1)

在这个示例中，我们设置了两个任务，job1 每隔 5 秒执行一次，而 job2 每隔 10 秒执行一次。

4. 高级用法

除了简单的秒级调度，schedule 还支持基于分钟、小时、天、星期的调度，甚至还可以设置每天的特定时间点执行任务。

4.1 基于时间单位的调度

schedule 提供了丰富的时间单位来配置任务的调度，包括：秒（seconds）、分钟（minutes）、小时（hours）、天（days）、周（weeks）等。下面是一些常见用法：

import scheduledef job():print("任务执行中...")# 每 10 分钟执行一次
schedule.every(10).minutes.do(job)# 每小时执行一次
schedule.every().hour.do(job)# 每天执行一次
schedule.every().day.at("10:30").do(job)# 每周一执行一次
schedule.every().monday.do(job)

4.2 以特定时间间隔执行任务

除了固定的时间单位外，schedule 还支持按天或周的时间间隔执行任务。例如，你可以设置每隔 3 天执行一次任务，或者每隔两周的某天执行任务：

# 每隔 3 天执行一次
schedule.every(3).days.do(job)# 每两周的周三执行一次
schedule.every(2).weeks.on('wednesday').do(job)

5. 停止任务

在某些情况下，你可能需要停止特定的任务。可以通过 cancel_job() 来取消任务的调度。示例：

import scheduledef job():print("任务执行中...")# 设置任务
task = schedule.every(5).seconds.do(job)# 取消任务
schedule.cancel_job(task)

6. 完整示例

为了展示 schedule 的灵活性，我们来看一个更完整的示例。这个例子中会定时发送提醒，并在特定的时间执行备份任务：

import schedule
import timedef send_reminder():print("发送提醒邮件...")def backup_database():print("备份数据库...")# 每隔 30 分钟发送一次提醒
schedule.every(30).minutes.do(send_reminder)# 每天凌晨 2:00 备份数据库
schedule.every().day.at("02:00").do(backup_database)while True:schedule.run_pending()time.sleep(1)