直接用 heapq 实现优先级队列需小心时间戳，因为当优先级相同时，heapq 会比较元组后续元素；若 task 不可比较（如 dict），将抛出 TypeError。
为什么直接用 heapq 实现优先级队列要小心时间戳？
Python 的 heapq 只保证最小堆，不支持自定义比较逻辑，所以当多个任务优先级相同时，它会尝试比较元组里的下一个元素——比如你写成 (priority, task)，一旦 task 是不可比较对象（如 dict、自定义类），就会抛出 TypeError: '。
常见错误写法：heapq.heappush(heap, (priority, task)) —— 看似简洁，实则埋雷。
正确做法是插入一个唯一递增序号，把三元组变成 (priority, insertion_count, task)：
import heapq
<p>class PriorityQueue:
def <strong>init</strong>(self):
self._heap = []
self._count = 0 # 保证同一 priority 下按插入顺序排序</p><pre class='brush:python;toolbar:false;'>def push(self, item, priority):
heapq.heappush(self._heap, (priority, self._count, item))
self._count += 1
def pop(self):
return heapq.heappop(self._heap)[-1]</pre>
heapq.heappushpop 和 heapq.heapreplace 该选哪个？
两者都用于“推入+弹出”，但行为不同，选错会导致任务丢失或顺序错乱。
立即学习“Python免费学习笔记（深入）”；
-
heapq.heappushpop(heap, item)：先 push，再 pop 最小元素。如果item比原堆顶还小，那它自己会被立刻弹出，实际没进队列。 -
heapq.heapreplace(heap, item)：先 pop 堆顶，再 pushitem。无论item多大，它一定进队列，且堆大小不变。
典型场景：限流任务队列（最多存 N 个待执行任务）。
要保留最高优先级的 N 个任务？用 heapreplace；要尝试插入再看是否挤掉最不重要的？用 pushpop。
如何让任务支持延迟执行（类似 schedule_at(timestamp)）？
heapq 本身不处理时间，但你可以把时间戳作为优先级字段。关键点在于：优先级越小越先执行，所以得用绝对时间戳（如 time.time()），而不是倒计时。
示例：
import heapq
import time
<h1>推迟 5 秒执行</h1><p>delayed_task = ("send_email", time.time() + 5)
heapq.heappush(heap, (delayed_task[1], count, delayed_task[0]))
count += 1消费端需轮询或配合 select/epoll 做等待，不能靠 heapq 自动触发。简单轮询写法：
while heap:
next_time, _, task = heap[0] # peek
if time.time() >= next_time:
heapq.heappop(heap)
execute(task)
else:
time.sleep(max(0.01, next_time - time.time()))注意：time.time() 在某些系统上精度有限，高频率调度建议用 time.monotonic()。
多线程环境下直接用 heapq 安全吗？
不安全。heapq 的所有操作（heappush、heappop、heapify）都不是原子的。两个线程同时 heappush 同一个列表，可能破坏堆结构，导致后续 heappop 返回错误元素甚至崩溃。
必须加锁，但别锁整个操作——否则吞吐量骤降。推荐粒度控制：
- 对
push和pop单独加threading.Lock - 避免在锁内做耗时操作（如序列化、网络请求）
- 如果只是生产者/消费者模型，考虑用
queue.PriorityQueue（它内部已封装锁和条件变量）
queue.PriorityQueue 底层就是 heapq + 锁，接口更安全，但无法直接访问底层堆做 peek 或批量修改——这点容易被忽略。
