自制Python任务调度模块-MySchedule

type

status

date

slug

summary

自制Python任务调度模块-MySchedule

前言

工作中需要写一个Python工具的每日定时任务功能，自然想到用子线程来执行任务调度。业务需求中该定时任务需要反复取消、恢复、设置参数、重新安排。

起初采用了sched库，在子线程中递归调用任务，但是发现了一些坑点，简言之：

sched本身不具备线程管理功能，需要把sched.run()放到子线程中执行。

sched.run()是阻塞式的，在run的过程中如果重置了任务队列（sched.queue），不能重新run，即无法立即重新安排任务。

原因是scheduler已经在run了就会阻塞子线程（底层是time.sleep，阻塞式且不可中断）。

浅看了下sched库实现的源码，发现其底层用的还是while True + time.sleep：

另外，sched库用二叉堆（优先队列）heapq按照任务时间先后给任务排序的做法还是很有参考价值的~

正篇

于是决定自己写一个更契合业务需求的模块——自带子线程、支持任务中断的MyScheduler。

MyScheduler特性

MyScheduler 是一个轻量级的调度器模块，用于管理和执行定时任务。它基于优先队列和线程管理，支持动态添加任务、中断任务执行以及重置调度器。

优先级任务：

使用 queue.PriorityQueue 存储任务，确保任务按执行时间排序。

支持动态添加任务，任务按延迟时间独立计时。

线程管理：

使用单独的线程 (task_thread) 监控任务队列并执行任务。

提供 start() 和 stop() 方法，用于控制调度器线程的生命周期。

事件中断：

使用 threading.Event() 实现任务执行的中断和重置。

支持动态重置调度器，清空任务队列并中断当前任务的延迟执行。

线程通信：

使用 queue.PriorityQueue 作为任务载体，主线程放入或清空、子线程取出并执行；threading.Event() 实现线程间的通信。

供外部调用的方法

add_task(task, delay)

功能：向任务队列中添加任务。

参数：

task (callable)：要执行的任务函数。
delay (float)：任务的延迟时间（秒）。

clear_tasks()

功能：清空任务队列。

reset()

功能：重置调度器，清空任务队列并中断当前任务的延迟执行。

start()

功能：启动调度器线程。

stop()

功能：停止调度器线程。

get_next_task_time()

功能：获取下一个任务的执行时间。

str：下一个任务的执行时间字符串，格式为 "YYYY-MM-DD HH:MM:SS"，如果队列为空则返回 "当前没有任务"。

实现细节

优先队列排序任务

🐈

使用queue.PriorityQueue作为任务队列的容器，其底层使用heapq模块来维护一个最小堆，内部通过锁机制（threading.Lock）确保多线程环境下的数据一致性。

在heapq中，任务的排序是通过比较元组的第一个元素（即execute_time）来实现的。当将(execute_time, task)放入PriorityQueue时execute_time越小的任务会排在越前面。

线程执行支持中断

threading.Event() 是一个简单的线程同步工具，它允许一个线程通知另一个线程某个事件已经发生，用来处理阻塞和中断：

.wait()：阻塞线程，直到指定的延迟时间或事件被触发。

.set()：中断阻塞，允许线程重置或重新安排任务。

与time.sleep对比：

特性	`time.sleep`	`threading.Event().wait`
是否可中断	不可中断	可中断
中断方式	无	通过 `.set()`
返回值	无	`True`（被中断）或 `False`（超时）
适用场景	简单的定时等待	需要响应外部事件的等待

子线程的线程方法监听主线程中是否.set()

主线程提供重置和停止线程的接口，都需要.set()

完整代码

完整代码居然不足100行，之后有空会参考sched的实现追加一些新接口，使用案例可以参考GitHub上的schedule.py。