不能直接用 multiprocessing.queue 包装 priorityqueue，因其内部含不可序列化的锁且非进程安全；推荐用 manager().list() 模拟，或高吞吐时改用 redis zset。
为什么不能直接用 multiprocessing.Queue 包装 PriorityQueue
因为 multiprocessing.Queue 只能传输可序列化对象，而 PriorityQueue 内部依赖 heapq 和锁（_thread.Lock），本身不可 pickle —— 直接传入会报 TypeError: can't pickle _thread.lock objects。更关键的是，即使绕过 pickle 问题，原生 PriorityQueue 也不是进程安全的：多个进程并发调用 put() 或 get() 会破坏堆结构，导致数据错乱或死锁。
用 multiprocessing.Manager() + list 模拟优先队列
这是最轻量、兼容性最好的方案，适合中低频写入场景（每秒百次级）。核心思路是用 Manager().list() 存储 (priority, item) 元组，每次 put() 后手动 heapq.heapify()，get() 前加锁并 heapq.heappop()。
实操建议：
- 必须用
Manager().Lock()而非threading.Lock()，后者跨进程无效 -
heapq.heapify()是 O(n) 操作，频繁put()时性能明显下降；若插入频率高，改用bisect.insort()维护有序列表（但需自行处理 priority 相同时的稳定排序） - 示例关键片段：
from multiprocessing import Manager import heapq <p>class ManagedPriorityQueue: def <strong>init</strong>(self, manager): self._data = manager.list() self._lock = manager.Lock()</p><pre class="brush:php;toolbar:false;"><pre class="brush:php;toolbar:false;">def put(self, item, priority=0): with self._lock: self._data.append((priority, item)) heapq.heapify(self._data) # 注意：这里不是 heappush！ def get(self): with self._lock: if not self._data: raise IndexError("pop from empty priority queue") return heapq.heappop(self._data)[1]
用 multiprocessing.Queue + 自定义序列化 + 外部排序逻辑
适用于高吞吐场景（如每秒数千任务），但要求所有 item 本身可 pickle，且业务能接受“插入不立即排序，消费时按优先级批量拉取”的模型。
实操建议：
- 用
multiprocessing.Queue原生传输 (priority, item)，不封装成PriorityQueue实例 - 消费者进程启动后，先
queue.get_nowait()拉取全部待处理项到本地list，再heapq.heapify()，然后循环heappop()—— 避免每次get()都跨进程通信 - 注意
queue.empty()不可靠，必须用异常捕获判断队列结束：try: ... except queue.Empty: break - 若需严格 FIFO 保序（相同 priority 时），在 tuple 中加入递增序列号：
(priority, seq, item)，否则 CPython 的 tuple 比较可能因内存地址引发不稳定排序
第三方方案：queuelib 或 redis-py + heap
当业务已用 Redis 或需要强一致性、持久化、跨语言支持时，redis-py + ZSET 是更健壮的选择 —— ZSET 原生支持优先级排序与原子操作。
实操建议：
- 用
zadd(key, {item: priority})插入，zrange(key, 0, 0, withscores=True)查最小，zremrangebyrank(key, 0, 0)原子弹出 - 避免用
zrange+zrem两步操作，中间可能被其他进程修改；必须用 Lua 脚本封装成原子动作 -
queuelib库提供PickleFifoDiskQueue等磁盘队列，但无内置优先级支持，需自行扩展_push()方法调用heapq—— 这类方案调试成本高，仅推荐日志/离线任务等非实时场景
真正难的不是实现堆逻辑，而是权衡“排序时机”和“锁粒度”：早排序（插入时）加重写负担，晚排序（消费时）增加内存占用和延迟。多数情况，用 Manager().list() + 显式锁 + 小批量 heapify() 已足够；一旦出现明显性能瓶颈，就该切到 Redis，别硬啃 multiprocessing 的边界。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
