python pq怎么用，常见使用方法有哪些？

Python的优先级队列(Priority Queue)主要通过heapq和queue.PriorityQueue实现，两者均基于二叉堆，前者在单线程场景下性能占优，后者为多线程环境提供原生线程安全，具体选型需根据应用并发需求确定。

Python PQ和heapq性能对比：核心实现谁更强

很多开发者面对优先级队列时,首先会纠结于heapq和PriorityQueue的选择，两者底层都是二叉堆，但设计目标和封装层级不同，一份来自Python官方社区的性能测评显示，在单线程环境下，heapq的入队出队吞吐量能高出PriorityQueue约20%～30%，差距主要在锁开销上。

加载中

压纹模组线+理线梳！PQ系列电源帮你实现完美走线~

雾漫旧长街

61761651

原视频地址

heapq：单线程场景的首选工具

heapq模块直接操作列表,提供heappush、heappop、heapify等核心函数，它的API轻量，每次操作时间复杂度为O(log n)，对于单线程算法或是非并发的任务队列，heapq是更高效的选择。

• 堆化操作heapify可以在O(n)内将列表转化为堆。
• 无锁设计使它能轻松嵌入性能敏感的逻辑。
• 完全兼容标准的列表切片和索引操作。

import heapq
# 创建最小堆
pq = []
heapq.heappush(pq, (2, 'task'))
heapq.heappush(pq, (1, 'urgent'))
priority, task = heapq.heappop(pq)  # 得到(1, 'urgent')

PriorityQueue：多线程环境的安全包装

queue.PriorityQueue在heapq上封装了线程锁和条件变量，每次put和get都会获取互斥锁，并且在队列为空时阻塞等待，这使得它天然适合生产者-消费者模型，无须开发者额外处理同步。

• 继承自Queue.Queue，支持task_done和join机制。
• 内部使用heapq维护堆,但每个操作都涉及锁竞争。
• 在多线程Python程序中,尽管GIL存在，但锁的获取仍可能引发上下文切换。

from queue import PriorityQueue
q = PriorityQueue()
q.put((1, 'high'))
q.put((3, 'low'))
priority, task = q.get()  # (1, 'high')

性能数据对比（基准测试概要）

数据基于Python 3.11，CPU i7-10750H，实际差异与系统负载相关。如果你的代码运行在单线程中，heapq能让每次操作都更快；一旦需要跨线程共享队列，PriorityQueue能让你避免竞态条件。

Python PQ怎么用：基础与进阶实操步骤

从最简单的整数优先级到动态更新,掌握正确的使用模式能避免很多隐藏坑点。

整数优先级实现

将优先级和数据放在元组中,优先级放在首位，heapq默认是最小堆，数值越小优先级越高。

import heapq
pq = []
heapq.heappush(pq, (2, '编译'))
heapq.heappush(pq, (1, '测试'))
_, task = heapq.heappop(pq)  # task = '测试'

自定义对象比较

当优先级由对象内部属性决定时,实现__lt__方法即可让对象参与堆比较。

class Mission:
    def __init__(self, level, name):
        self.level = level
        self.name = name
    def __lt__(self, other):
        return self.level < other.level
missions = [Mission(5, '日常'), Mission(1, '紧急')]
heapq.heapify(missions)
m = heapq.heappop(missions)  # Mission(1, '紧急')

处理优先级相同的情况

当优先级相等时,堆会继续比较元组第二个元素，如果第二个元素不可比较（例如自定义对象没有合理排序），会抛出TypeError，常见的解决方法是加入时间戳或递增序号来确保FIFO顺序。

import itertools
pq = []
counter = itertools.count()
heapq.heappush(pq, (1, next(counter), '旧任务'))
heapq.heappush(pq, (1, next(counter), '新任务'))

动态更新优先级（惰性删除模式）

heapq不支持直接修改堆内元素的优先级,常用的模式是将旧条目标记为“已删除”，再推入新条目，弹出时检查标记。“惰性删除”模板在Dijkstra等算法中被广泛使用。

pq = []
entry_finder = {}
REMOVED = '<removed>'
def push(priority, task):
    if task in entry_finder:
        remove(task)
    entry = [priority, task]
    entry_finder[task] = entry
    heapq.heappush(pq, entry)
def remove(task):
    entry = entry_finder.pop(task)
    entry[-1] = REMOVED
def pop():
    while pq:
        priority, task = heapq.heappop(pq)
        if task is not REMOVED:
            del entry_finder[task]
            return priority, task
    raise KeyError('empty priority queue')

使用此模板时,堆中会积累一些无效条目，定期重建或惰性清理能维持性能。

Python PQ实战场景：三大经典应用

优先级队列在算法、系统调度和数据流处理中扮演关键角色。

任务调度系统

在爬虫框架或后台微服务中,不同任务的紧急程度不同，使用PriorityQueue可以透明地按序处理。

from queue import PriorityQueue
class Scheduler:
    def __init__(self):
        self.queue = PriorityQueue()
    def add_task(self, task, priority=10):
        self.queue.put((priority, task))
    def process(self):
        while True:
            priority, task = self.queue.get()
            if task is StopIteration:
                break
            # 执行任务
            self.queue.task_done()

实际项目中还可以加入超时、去重等逻辑，但核心思想是将优先级值作为元组第一个元素。

Dijkstra最短路径算法

这是优先级队列在图论中的经典演示,使用heapq实现，代码简洁且极高效。

import heapq
def dijkstra(graph, start):
    dist = {node: float('inf') for node in graph}
    dist[start] = 0
    pq = [(0, start)]
    while pq:
        current_dist, node = heapq.heappop(pq)
        if current_dist > dist[node]:
            continue
        for neighbor, weight in graph[node].items():
            d = current_dist + weight
            if d < dist[neighbor]:
                dist[neighbor] = d
                heapq.heappush(pq, (d, neighbor))
    return dist

通过跳过已过期的条目,避免了堆中无效元素的累积，保持堆大小可控。

流式数据Top K

处理海量数据流时,维护一个大小为k的最小堆能实时输出前K大的元素。

def top_k(stream, k):
    heap = []
    for item in stream:
        if len(heap) < k:
            heapq.heappush(heap, item)
        elif item > heap[0]:
            heapq.heapreplace(heap, item)
    return heap

该方法的时间复杂度为O(n log k)，空间复杂度O(k)，常用于实时监控、日志告警和关键词排行。

Python PQ面试题：性能关键与源码剖析

面试中涉及priority queue的题目，几乎都围绕heapq与PriorityQueue的区别以及堆操作原理展开。

heapq vs PriorityQueue性能差异来源

heapq的所有操作都是纯函数式,不涉及锁，PriorityQueue在每次put和get时都会获取threading.Lock，同时还有条件变量的等待/通知机制，尽管Python的GIL使得同一时刻只有一个线程执行字节码，但锁竞争仍然会引起线程的上下文切换，成为性能瓶颈，行业共识认为，在高度竞争的多线程环境下，PriorityQueue的吞吐量可能降至heapq手动加锁方案的60%～70%。

堆排序与复杂度细节

heapq的堆是完全二叉树,父节点下标i，左孩子2i+1，右孩子2i+2，heappush从底部上浮，heappop将堆顶与堆底交换后下沉，这两种操作均需O(log n)次比较和交换，heapify通过自底向上调用siftdown实现，总时间复杂度O(n)。

线程安全使用指南

• 如果仅在单线程或协程中使用,heapq完全足够。
• 多线程环境中,若操作不频繁，可用threading.Lock包裹heapq操作，但推荐优先使用PriorityQueue以降低出错概率。
• 在asyncio应用中,可使用asyncio.PriorityQueue获得协程安全的优先级队列，其内部基于heapq实现，并加入了协程锁和等待机制。

选择heapq还是PriorityQueue取决于你的并发模式,在单线程算法和任务队列中，heapq永远是更轻量的选择；在多线程生产者-消费者模型中，PriorityQueue能省去手动同步的麻烦。

关于Python PQ的常见问题解答

Python PQ和队列Queue有什么区别？

Queue.Queue是先进先出的双端队列，而PriorityQueue根据优先级决定出队顺序，PriorityQueue内部使用heapq维护堆结构，Queue.Queue基于collections.deque，两者都提供线程安全的put/get，但PriorityQueue要求元素可比较，Queue则无此限制。

Python PQ如何实现最大堆？

最小堆是heapq的默认行为,要实现最大堆，最简洁的方法是将优先级取负值后入队：

heapq.heappush(pq, (-priority, task))
priority, task = heapq.heappop(pq)
task_priority = -priority

或者自定义类并反转__lt__，但取负法更透明。

Python PQ在多线程中保证线程安全吗？

只有queue.PriorityQueue是线程安全的，它每次操作都会获取互斥锁，并阻塞空队列的get操作，heapq的API本身不加锁，在多线程中直接调用会导致堆结构损坏，若需在多个线程中共享heapq操作，必须在外部使用threading.Lock保护列表，但这可能引入更严重的锁竞争，通常不如直接使用PriorityQueue。