Python sched 模块

最后修改于 2025 年 3 月 10 日

Python 中的 sched 模块是一个轻量级的工具，用于调度事件在特定时间运行。它非常适合单线程应用程序中的周期性作业或延迟操作等任务。本教程通过清晰的示例介绍其用法。

与线程或多进程不同，sched 简单且非阻塞，使其成为满足基本定时需求的理想选择，而无需增加复杂性。

基本用法

此示例演示了如何安排具有延迟的任务。

import sched
import time

def print_message():
    print("Hello, World!")

scheduler = sched.scheduler(time.time, time.sleep)
scheduler.enter(5, 1, print_message)  # Run after 5 seconds
scheduler.run()  # Start the scheduler

在这里，scheduler.enter 设置 print_message 函数在 5 秒后运行。 scheduler.run 按顺序执行所有计划的事件。

重复事件

此示例创建一个每 5 秒重复一次的任务。

import sched
import time

def print_message():
    print(f"Tick at {time.ctime()}")
    scheduler.enter(5, 1, print_message)  # Reschedule

scheduler = sched.scheduler(time.time, time.sleep)
scheduler.enter(5, 1, print_message)  # First run
scheduler.run()  # Runs indefinitely

该函数重新调度自身，形成一个循环。请注意，scheduler.run 会阻塞直到手动停止。

优先级和参数

此示例使用优先级并将参数传递给任务。

import sched
import time

def print_message(msg):
    print(msg)

scheduler = sched.scheduler(time.time, time.sleep)
scheduler.enter(5, 2, print_message, ("Low priority",))  # Priority 2
scheduler.enter(5, 1, print_message, ("High priority",))  # Priority 1
scheduler.run()

如果时间重叠，则优先级较低的数字首先运行。 argument 元组将数据传递给该函数。

取消计划的事件

此示例在计划的任务运行之前将其取消。

import sched
import time

def print_message():
    print("This won't run!")

scheduler = sched.scheduler(time.time, time.sleep)
event = scheduler.enter(5, 1, print_message)
scheduler.cancel(event)  # Remove from queue
scheduler.run()  # Nothing happens

scheduler.cancel 接受一个事件对象并将其删除，如果尚未运行，则阻止执行。

高级：使用绝对时间

此示例计划在确切时间执行任务。

import sched
import time

def print_message():
    print(f"Event at {time.ctime()}")

scheduler = sched.scheduler(time.time, time.sleep)
run_time = time.time() + 5  # 5 seconds ahead
scheduler.enterabs(run_time, 1, print_message)
scheduler.run()

scheduler.enterabs 使用绝对时间（例如，Unix 时间戳）而不是相对延迟。

新示例：具有异常处理的任务

此示例安全地处理计划任务中的错误。

import sched
import time

def risky_task():
    try:
        result = 10 / 0  # Simulate error
    except ZeroDivisionError:
        print("Caught an error!")

scheduler = sched.scheduler(time.time, time.sleep)
scheduler.enter(3, 1, risky_task)  # Run after 3 seconds
scheduler.run()

将任务包装在 try-except 中可以防止调度程序因未处理的异常而崩溃。

新示例：计划的文件清理

此示例计划文件删除任务。

import sched
import time
import os

def cleanup_temp_file():
    file = "temp.txt"
    if os.path.exists(file):
        os.remove(file)
        print(f"Deleted {file}")
    else:
        print(f"{file} not found")

scheduler = sched.scheduler(time.time, time.sleep)
scheduler.enter(5, 1, cleanup_temp_file)
scheduler.run()

这模拟了一个清理作业，对于临时文件或日志很有用。检查文件是否存在以避免错误。

sched 的最佳实践

• 保持任务轻量： 用于简单，非阻塞的作业。
• 处理错误： 在任务中添加异常处理。
• 设置优先级： 对同一时间事件使用优先级。
• 取消事件： 删除不需要的任务以节省资源。
• 避免阻塞： 不要用于长时间运行的任务。
• 记录活动： 记录事件以进行调试和监视。

来源

Python sched 模块文档

本教程探讨了 sched 模块，其中包含重复任务，优先级和文件清理等示例。遵循最佳实践以实现可靠的调度。

作者

我叫 Jan Bodnar，是一位充满热情的程序员，拥有多年的经验。自 2007 年以来，我撰写了超过 1400 篇文章和 8 本电子书，并拥有超过 8 年的编程教学经验。

列出所有 Python 教程。