不能直接用 heapq 存储 (timestamp, func, args)，因为当 timestamp 相同时，Python 会比较 func；而函数对象默认不可比较，触发 TypeError。
为什么不能直接用 heapq 存储函数和时间戳
直接把 (timestamp, func, args) 塞进 heapq 看似合理，但一旦 func 不可比较（比如是普通函数对象），heapq 在堆化或弹出时会报 TypeError: '。Python 的堆不支持对元组中第二个元素（函数）做大小比较，哪怕第一个元素（时间戳）已足够排序。
解决方法是引入一个单调递增的计数器，把元组变成 (timestamp, counter, func, args)。只要 counter 严格递增，即使时间戳相同，也能避免比较函数对象：
import heapq
import time
<p>class TaskScheduler:
def <strong>init</strong>(self):
self._heap = []
self._counter = 0 # 避免时间戳相同时比较函数</p><pre class="brush:php;toolbar:false;">def schedule(self, delay_sec, func, *args):
timestamp = time.time() + delay_sec
heapq.heappush(self._heap, (timestamp, self._counter, func, args))
self._counter += 1
如何安全地运行到期任务而不阻塞主线程
定时调度器常被误写成「死循环 + sleep」，但 time.sleep() 无法响应新任务插入，且精度差、无法中断。更稳妥的做法是：每次只 sleep 到最近一个任务的触发时刻，期间允许外部调用 schedule() 插入新任务。
关键点在于：每次 pop 前先检查堆顶是否已到期，未到期就 sleep 差值；到期则 pop 并执行——且必须在 try/except 中运行任务，防止一个崩溃导致整个调度器停摆：
- 用
heapq.heappop()只在确认到期后才调用，避免提前破坏堆结构 - sleep 时间用
max(0, next_time - time.time())，防止因系统时钟跳变或任务执行过长导致负值 - 任务执行需包裹
try/except BaseException（不是Exception），因为用户代码可能 raiseSystemExit或KeyboardInterrupt
怎样支持取消已调度但未执行的任务
原生 heapq 不支持删除中间元素，暴力遍历删除再重建堆效率低（O(n)），且破坏线程安全性。实用解法是「惰性删除」：给每个任务加唯一 task_id，维护一个 set 记录已取消的 ID，在 pop 出任务时先检查 ID 是否在取消集合中。
python-docx Skill功能概述python-docx Skill是一项面向实际任务的技能，主要用于本Skill提供使用python-docx生成专业Word文档的标准方法和最佳实践；生成安全服务方案文档；核心要点生成技术架构设计文档；生成任何需要专业排版的Word文档；核心库 : python-docx；使用与执行辅助库 : docx.shared , docx.enum , docx.oxml.ns；标准代码模板；1. 文档初始化；2. 字体设置（必须！它将相关步骤、工具调用和结果整理方式集
示例逻辑片段：
def cancel(self, task_id):
self._cancelled.add(task_id)
<p>def run_once(self):
while self._heap:
timestamp, counter, func, args = self._heap[0] # peek only
if timestamp > time.time():
break
heapq.heappop(self._heap)
if counter in self._cancelled:
continue # skip cancelled task
try:
func(*args)
except BaseException:
pass # log if needed
</p>
注意：这里用 counter 作 task_id 是可行的（因它全局唯一且不可重用），但若需显式管理任务生命周期，建议改用 uuid.uuid4() 并额外存储映射关系。
为什么不要在任务函数里调用 time.sleep() 或阻塞 I/O
调度器本身是单线程的，任何阻塞操作都会卡住后续所有任务。比如一个任务里写了 requests.get(...)，整个调度器就停在那儿等响应，其他定时任务全部延迟。
真实场景应切换为异步模型：asyncio.PriorityQueue 替代 heapq，任务用 async def 定义，并用 asyncio.create_task() 启动。同步版调度器只适合轻量、纯 CPU 的短任务（如刷新缓存、打日志）。若必须调用阻塞 API，至少用 loop.run_in_executor() 托管到线程池。
容易被忽略的是：即使用了 asyncio，也要避免在 __aenter__ 或回调中做耗时计算——优先队列的「优先」只体现在入队顺序，不保证执行不被抢占。真正的实时性需要 OS 级调度或专用服务（如 APScheduler 的 BackgroundScheduler 配合线程池）。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
