迹忆客 专注技术分享

本篇文章将介绍 Python 中的 Schedule 包，以在特定时间间隔后定期安排作业。

Schedule是Python中的一个轻量级进程调度程序库，用于安排任务以指定的时间间隔定期运行。

我们可以使用人类友好的语法调用函数或任何可调用对象来自动执行任务，例如发送电子邮件、设置提醒、定期获取货币或比特币的价格。

在 Python 中安装 Schedule 库

我们必须首先使用以下命令安装此软件包才能使用它。

#Python 3.x
pip install schedule

为了安排作业，我们使用以下语法。

语法：

#Python 3.x
Schedule.every(interval).[timeframe].do(function)

参数:

• interval 可以是任意整数。
• 时间范围可以是秒、分钟、小时、天，甚至是工作日的名称，如星期一、星期二等。
• do() 方法将函数名称作为调度的参数。

使用秒作为间隔来定期安排作业

我们在下面的代码中有一个名为任务的方法，计划每五秒定期运行一次。 我们将时间间隔（在本例中为五秒）定义为 every() 方法的参数。

我们指定要调度的函数作为 do() 方法的参数，该方法在机器到达指定时间后自动运行。 最后，我们在 while 循环中执行 run_pending() 函数来连续运行计划的作业。

在输出中，消息“Task is running”按照编程每五秒一次显示一次。

示例代码：

#Python 3.x
import schedule as s
import time as t
def task():
    print("Task is running")
s.every(5).seconds.do(task)
while True:
    s.run_pending()
    t.sleep(1)

输出:

#Python 3.x
Task is running
Task is running
Task is running
Task is running
Task is running

使用分钟、小时和天作为间隔来定期安排作业

如果我们想将间隔更改为任何指定的分钟、小时或一周中的几天，我们可以编写这样的代码。 输出将根据设置的时间持续显示消息。

示例代码：

#Python 3.x
import schedule as s
import time as t
def task():
    print("Task is running")
s.every(5).minutes.do(task)
s.every(10).hours.do(task)
s.every().monday.do(task)
while True:
    s.run_pending()
    t.sleep(1)

输出:

#Python 3.x
Task is running
Task is running
Task is running

使用 HH:MM 格式的时间作为间隔来定期安排作业

我们还可以设置 HH:MM:SS 格式的特定时间作为函数的参数。 该消息将在每个指定的时间间隔后定期显示。

示例代码：

#Python 3.x
import schedule as s
import time as t
def task():
    print("Task is running")
s.every().day.at("04:21").do(task)
s.every().monday.at("12:00").do(task)
while True:
    s.run_pending()
    t.sleep(1)

输出:

#Python 3.x
Task is running

使用函数装饰器来安排作业

函数装饰器接受函数作为输入，添加一些功能，然后返回一个函数。 我们将在这里使用 @repeat 来安排任务函数。

示例代码：

#Python 3.x
from schedule import every, run_pending, repeat
import time as t
@repeat(every(5).seconds)
def task():
    print("Task is running")
while True:
    run_pending()
    t.sleep(1)

输出:

#Python 3.x
Task is running
Task is running
Task is running

一次运行一个作业

我们可以通过在调度实例中使用 CancelJob 来取消作业，从而立即运行作业。

示例代码：

#Python 3.x
import schedule as s
import time as t
def task():
    print("Task is running")
    return s.CancelJob
s.every(5).seconds.do(task)
while True:
    s.run_pending()
    t.sleep(1)

输出:

#Python 3.x
Task is running

取消所有作业

我们可以使用调度程序实例的 clear() 方法取消调度程序中的所有作业。 在这里，我们使用 get_jobs() 方法从调度程序打印调度的作业。

示例代码：

#Python 3.x
import schedule as s
import time as t
def task():
    print("Task is running")
s.every(5).seconds.do(task)
print("Jobs:", s.get_jobs())
s.clear()
print("Jobs:", s.get_jobs())

输出:

#Python 3.x
Jobs: [Every 5 seconds do task() (last run: [never], next run: 2022-04-08 16:59:37)]
Jobs: []