sched 是 Python 标准库中的一个轻量级事件调度器模块，用于安排将来的任务执行。它允许在给定时间（或延迟）后执行函数，常用于编写定时执行的脚本或事件模拟程序。

该模块基于用户传入的时间函数（如 time.time）和延迟函数（如 time.sleep）进行调度，不依赖系统级事件，因此适用于单线程、阻塞型的调度任务。

常见应用场景：（1）实现延迟任务（如 10 秒后执行某函数）。（2）构建简单的定时任务系统（适用于脚本或演示）。（3）控制任务执行顺序和优先级。（4）在模拟环境中精确安排事件顺序（如事件模拟器）。（5）替代 time.sleep() 实现非阻塞型任务调度（结合 threading）。

◆  ◆  ◆核心概念

sched.scheduler 是调度器的核心类，其构造函数如下：sched.scheduler(timefunc, delayfunc)timefunc：返回当前时间戳的函数，一般使用 time.time。delayfunc：延迟一定秒数的函数，一般使用 time.sleep。

该调度器内部维护一个事件队列（优先队列），每个事件都包含：（1）执行时间（时间戳）（2）优先级（数值越小越先执行）（3）执行的函数及参数

调度器调用 .run() 方法时，会按照任务的时间顺序顺次执行。

◆  ◆  ◆应用举例

例 1：延迟执行一次性任务（基础用法）import schedimport time

# 创建调度器：使用 time.time 作为时间函数，time.sleep 作为延迟函数scheduler = sched.scheduler(time.time, time.sleep)

# 待执行的任务函数def say_hello():    print("5 秒后执行：Hello, Python!")

# 安排任务：5 秒后执行，优先级为 1scheduler.enter(delay=5, priority=1, action=say_hello)

print("等待任务执行...")scheduler.run()例 2：安排多个任务并设置优先级import schedimport time

scheduler = sched.scheduler(time.time, time.sleep)

def task(name):    print(f"{time.strftime('%X')} 执行任务: {name}")

# 延迟时间相同，通过优先级控制执行顺序scheduler.enter(3, 2, task, argument=("任务 B",))  # 优先级 2scheduler.enter(3, 1, task, argument=("任务 A",))  # 优先级 1，先执行scheduler.enter(3, 3, task, argument=("任务 C",))  # 优先级 3

print("等待调度执行...")scheduler.run()例 3：每 2 秒执行一次任务（手动循环添加）import schedimport time

scheduler = sched.scheduler(time.time, time.sleep)

def repeated_task():    print(f"{time.strftime('%X')}：定时任务正在执行")    # 再次安排自己，达到循环执行效果    scheduler.enter(2, 1, repeated_task)

# 首次安排scheduler.enter(0, 1, repeated_task)scheduler.run()提示：由于 scheduler.run() 是阻塞的，该代码会永久运行，直到你手动停止（如 Ctrl + C）。

例 4：配合 threading 实现非阻塞调度器import schedimport timeimport threading

scheduler = sched.scheduler(time.time, time.sleep)

def say_hi():    print(f"{time.strftime('%X')}：你好，调度器！")

def run_scheduler():    scheduler.run()

# 安排任务 4 秒后执行scheduler.enter(4, 1, say_hi)

# 启动调度器线程，不阻塞主线程thread = threading.Thread(target=run_scheduler)thread.start()

# 主线程继续执行for i in range(3):    print(f"{time.strftime('%X')}：主线程运行中...")    time.sleep(1)例 5：定时发送“心跳”信息（模拟系统监控）import schedimport time

scheduler = sched.scheduler(time.time, time.sleep)

def heartbeat():    print(f"{time.strftime('%X')}：发送心跳信号")    # 每 10 秒发送一次    scheduler.enter(10, 1, heartbeat)

# 启动心跳scheduler.enter(0, 1, heartbeat)scheduler.run()

◆  ◆  ◆常用函数速览

cancel(event)

将该事件从调度器队列中移除。参数：event：由 enter() 或 enterabs() 返回的事件对象返回：无异常：如果事件已经执行或不存在，抛出 ValueError

empty()

判断当前调度器是否还有待执行任务。参数：无返回：布尔值，若为空返回 True

enter(delay, priority, action, argument=(), kwargs={})

在当前时间 + delay 时执行指定函数。参数：delay：延迟的秒数priority：任务优先级（整数，数值越小越优先）action：待执行的函数argument：传给函数的位置参数（元组）kwargs：传给函数的关键字参数（字典）返回：事件对象，可用于取消该任务

enterabs(time, priority, action, argument=(), kwargs={})

在指定的时间戳执行指定函数。参数：time：绝对时间戳（如 time.time() + 5）其余参数与 enter() 相同返回：事件对象

queue

获取当前调度队列中的所有事件（列表）。返回：一个按执行时间和优先级排序的事件对象列表

run(blocking=True)

按顺序运行所有任务，直到队列为空或被中断。参数：blocking（默认 True）：是否阻塞执行。若为 False，仅执行当前已到时间的任务返回：无

◆  ◆  ◆补充说明

1、sched 设计用于单线程同步调度，不适合精确计时或高并发任务。

2、若需后台持续调度，建议与 threading.Thread 配合使用。

3、若需功能更强的调度工具，可使用 threading.Timer 或第三方库 schedule、APScheduler。

4、所有任务执行均发生在 .run() 调用时，队列不会自动运行。

更多相关技术内容咨询欢迎前往并持续关注好学星城论坛了解详情。

想高效系统的学习Python编程语言，推荐大家关注一个微信公众号：Python编程学习圈。每天分享行业资讯、技术干货供大家阅读，关注即可免费领取整套Python入门到进阶的学习资料以及教程，感兴趣的小伙伴赶紧行动起来吧。