Python中的5种队列数据结构，你知道几个？

1、基础队列：列表实现

在Python中，最直接实现队列数据结构的方式莫过于使用内置的列表（list）。尽管这不是最高效的解决方案 ，但其简洁性使其成为学习队列概念的起点。

1.1 列表作为队列的简单应用

列表提供了append()方法用于在尾部添加元素（相当于队列的入队操作），以及pop(0)方法来移除并返回列表的第一个元素（相当于队列的出队操作）。

•入队（Enqueue）：使用list.append()方法在列表的末尾添加元素。这对应于队列中新元素的加入，遵循先进先出的原则。

•出队（Dequeue）：标准的列表不直接支持队列的头部移除操作 ，通常使用list.pop(0)来模拟。这种方法虽然实现了功能，但是效率低下，特别是当队列很大时，因为每次出队操作都需要移动后续的所有元素。

下面是一个基本的队列操作示例：

queue = []

# 入队操作

queue.append("任务A")

queue.append("任务B")

print("初始队列:", queue)

# 出队操作

task = queue.pop(0)

print("处理任务:", task)

print("处理后队列:", queue)

输出结果：

初始队列: ['任务A', '任务B']

处理任务: 任务A

处理后队列: ['任务B']1.2 进阶：循环队列模拟

标准列表实现的队列存在一个问题，即当频繁执行出队操作时 ，由于pop(0)操作的时间复杂度为O(n)，效率较低。通过模拟循环队列 ，我们可以减少这种开销。这里采用列表的两端作为队首和队尾，同时维护两个指针分别记录队首和队尾的位置。

class CircularQueue:

def __init__(self):

self.queue = []

self.head = 0

self.tail = 0

def enqueue(self, item):

self.queue.append(item)

self.tail += 1

def dequeue(self):

if self.is_empty():

return None

item = self.queue[self.head]

self.head = (self.head + 1) % len(self.queue)

if self.head == self.tail:  # 清理已处理的元素，防止内存泄漏

self.queue = []

self.head = 0

self.tail = 0

return item

def is_empty(self):

return self.head == self.tail and not self.queue

cq = CircularQueue()

cq.enqueue("任务1")

cq.enqueue("任务2")

print("出队任务:", cq.dequeue())

print("当前队列状态:", cq.queue)

输出结果：

出队任务: 任务1

当前队列状态: ['任务2']1.3 缺点与优化建议

虽然上述模拟循环队列的方法解决了效率问题，但它依然不是最优解。使用列表作为底层实现的主要缺点在于，随着队列操作的进行，频繁的插入和删除可能导致内存碎片。

•效率问题：由于Python列表是基于数组实现的，当执行list.pop(0)操作时，需要移动所有后续元素来填补空位，时间复杂度为O(n) ，在大规模数据处理时效率极低。

•循环利用问题：真实世界中的队列往往期望能够高效地循环利用存储空间，而列表实现的队列在不断出队后 ，无法自动复用前面的空间，导致内存使用效率不高。

为了进一步提升性能和内存效率，Python标准库提供了collections.deque，它专为此类场景设计 ，具有O(1)的平均时间复杂度。对于追求极致性能的应用，推荐直接使用deque替代列表来实现队列，尤其是在处理大量数据或高并发场景下。此外，对于多线程或多进程环境，考虑使用queue模块提供的线程安全队列类型，以避免数据竞争问题。

2、高效队列：collections.deque

在Python标准库中，collections.deque（双端队列）是一种优化的队列实现 ，特别适用于需要高效插入和删除元素的场景。相较于使用列表模拟队列，deque提供了更优的性能和灵活性。

2.1 deque介绍与优势

deque，全称双端队列（double-ended queue），是一种允许在其两端进行添加和删除操作的数据结构。它克服了列表作为队列使用时的主要瓶颈——出队操作的低效。deque通过使用双向链表实现 ，确保了在两端执行插入和删除操作的时间复杂度均为O(1)。

•快速操作：无论是头部还是尾部 ，入队（append/appendleft）和出队(pop/popleft)都非常迅速。

•内存效率：自动管理内存，无需像列表那样移动大量元素。

•灵活应用：不仅限于队列 ，还可以当作栈或双端队列使用。

2.2 实现快速入队与出队

使用deque实现队列的基本操作非常直观，下面的示例展示了如何使用deque进行快速的入队和出队：

from collections import deque

queue = deque()

# 入队操作

queue.append("任务1")

queue.append("任务2")

print("初始队列:", queue)

# 出队操作

task = queue.popleft()

print("处理任务:", task)

print("处理后队列:", queue)

输出结果：

初始队列: deque(['任务1', '任务2'])

处理任务: 任务1

处理后队列: deque(['任务2'])

除了基础的入队和出队操作，deque还支持一些进阶操作 ，增强了其实用性：

•限制大小：可以创建一个固定大小的deque，当达到容量上限时，新元素的加入会导致最旧的元素自动弹出 ，类似于循环队列。

from collections import deque

# 创建一个最大长度为3的deque

limited_deque = deque(maxlen=3)

limited_deque.append(1)

limited_deque.append(2)

limited_deque.append(3)

limited_deque.append(4)  # 此时1会被自动移除

print(limited_deque)  # 输出: deque([2, 3, 4], maxlen=3)

•双端访问：既可以从左侧添加或移除元素（appendleft和popleft） ，也可以从右侧操作 ，提供了更多操作灵活性。

2.3 性能提升实例演示

考虑一个场景 ，我们需要频繁地在序列的开始和结束添加元素，并从中提取最早添加的元素。使用deque相比列表，可以显著提高性能。

import time

from collections import deque

# 使用deque

start_time = time.time()

deque_example = deque()

for i in range(100000):

deque_example.append(i)

if len(deque_example) > 60000:

deque_example.popleft()

end_time = time.time()

deque_duration = end_time - start_time

# 使用列表

start_time = time.time()

list_example = []

for i in range(100000):

list_example.append(i)

if len(list_example) > 60000:

list_example.pop(0)

end_time = time.time()

list_duration = end_time - start_time

print(f"Deque操作耗时: {deque_duration}秒")

print(f"列表操作耗时: {list_duration}秒")

在上述示例中，通过比较deque和列表在相同操作下的执行时间，可以直观地看出deque在处理这类需求时的性能优势。

3、并发安全：queue模块

3.1 多线程/进程安全队列

在涉及多线程或多进程的并发程序设计中，数据同步和线程安全是必须面对的挑战。Python的queue模块为此提供了线程安全的队列类 ，包括Queue,LifoQueue（后进先出队列），以及PriorityQueue（优先级队列）。这些队列类内部实现了锁机制 ，确保了在多线程环境下的安全访问，避免了数据竞争和一致性问题。

3.2 queue模块核心类解析

• **FIFO队列 (Queue)**：最基本的形式 ，遵循先进先出（FIFO）原则 ，提供了put()方法用于插入元素到队列尾部，以及get()方法从队列头部移除并返回一个元素。

• **LIFO队列 (LifoQueue)**：后进先出队列，类似于堆栈，最近添加的元素将是最先被获取的元素，通过put()和get()操作实现。

• **优先级队列 (PriorityQueue)**：元素带有优先级的队列，每次出队都是优先级最高的元素。元素被赋予一个优先级数字，越小的数字代表越高的优先级。通过put()存放带有优先级的项，get()取出优先级最高的项。

3.3 生产者消费者模型实例

生产者消费者模型是一种经典的设计模式，常用于处理多线程或多进程间的协作。下面是一个使用queue模块实现的简单生产者消费者模型示例：

import threading

import time

from queue import Queue

def producer(queue):

"""生产者线程，向队列中添加任务"""

for i in range(10):

item = f"产品{i}"

print(f"生产了 {item}")

queue.put(item)

time.sleep(1)  # 模拟生产间隔

def consumer(queue):

"""消费者线程 ，从队列中获取并处理任务"""

while True:

item = queue.get()

if item is None:  # 使用None作为停止信号

break

print(f"消费了 {item}")

time.sleep(2)  # 模拟消费间隔

# 创建共享的线程安全队列

task_queue = Queue()

# 分别启动生产者和消费者线程

producer_thread = threading.Thread(target=producer, args=(task_queue,))

consumer_thread = threading.Thread(target=consumer, args=(task_queue,))

producer_thread.start()

consumer_thread.start()

producer_thread.join()  # 等待生产者线程完成

print("生产结束，发送停止信号")

task_queue.put(None)  # 向队列发送结束信号

consumer_thread.join()  # 等待消费者线程完成

此示例中，producer线程不断向队列中添加“产品”，而consumer线程则不断地从队列中取出并“消费”这些产品。通过队列作为中介 ，两者之间实现了有效的协同工作 ，同时保证了线程安全。

4、高级数据结构：堆与优先队列 PriorityQueue

4.1 何为优先队列及其应用场景

优先队列是一种特殊类型的队列，其中每个元素都有一个优先级。在这样的队列中，出队操作总是移除优先级最高的元素。优先队列广泛应用于各种场景，如任务调度、事件驱动系统、Dijkstra算法中的最短路径计算、优先级中断处理等。它确保了关键或紧急的任务能够得到及时处理。

4.2 使用heapq模块实现优先队列

Python的heapq模块提供了堆队列算法的实现，可以方便地用来构建优先队列。堆是一个近似完全二叉树的结构，满足父节点的值小于或等于（在最小堆中）其子节点的值。利用heapq，我们可以高效地实现优先队列的核心操作——插入元素（heappush）和获取并移除最小元素（heappop）。

示例代码如下：

import heapq

class PriorityQueue:

def __init__(self):

self._queue = []

self._index = 0  # 用于自定义优先级的计数器

def push(self, item, priority=0):

"""将元素插入队列，可选优先级"""

heapq.heappush(self._queue, (-priority, self._index, item))  # 使用负优先级进行最小堆排序

self._index += 1

def pop(self):

"""移除并返回优先级最高的元素"""

return heapq.heappop(self._queue)[-1]  # 返回元素部分

pq = PriorityQueue()

pq.push("任务A", 2)

pq.push("任务B", 1)

pq.push("任务C", 3)

print(pq.pop())  # 应输出优先级最高的任务C

print(pq.pop())  # 接下来是任务A4.3 自定义比较函数优化优先级处理

在某些情况下 ，优先级可能基于复杂的逻辑，而不仅仅是简单的数值比较。虽然可以通过转换策略间接实现 ，但直接提供自定义比较函数给heapq模块更为灵活。不过，heapq本身不直接支持传入自定义比较函数 ，而是依赖于元素的自然顺序或显式指定的负优先级值。因此，为了实现这一需求 ，通常是在元素中包含或绑定比较逻辑。

例如，如果任务类本身定义了比较操作，那么直接使用该类的实例即可：

class Task:

def __init__(self, name, priority):

self.name = name

self.priority = priority

def __lt__(self, other):

return self.priority < other.priority  # 定义优先级比较

# 使用Task实例作为优先队列的元素

pq = []

heapq.heapify(pq)  # 将列表转换为堆结构

heapq.heappush(pq, Task("任务D", 5))

heapq.heappush(pq, Task("任务E", 9))

print(heapq.heappop(pq).name)  # 应输出优先级最低的任务D

通过这种方式，我们可以更灵活地控制优先级的比较逻辑，使优先队列更加适应复杂的应用场景。

4.4 实时调度与Top-K问题解决

•实时调度：在操作系统和网络通信等领域 ，堆可实现高效的任务调度算法，如事件驱动编程中的事件调度，能确保优先级高的任务优先得到处理。

•Top-K问题：当需要找出一组数据中的前K个最大或最小元素时，堆是一个理想的数据结构。通过维护一个大小为K的最小堆（找最大K个元素）或最大堆（找最小K个元素） ，可以在O(NlogK)的时间复杂度内完成。

示例：使用heapq求解Top-3最大数。

import heapq

numbers = [1, 8, 9, 5, 12, 7, 15, 20, 3]

top_k = 3

# 使用nlargest函数直接求解Top-K问题

top_three = heapq.nlargest(top_k, numbers)

print(f"Top-{top_k}最大数：{top_three}")

# 输出结果：

# Top-3最大数：[20, 15, 12]

通过heapq模块，开发者可以轻松实现复杂的堆相关应用 ，满足对高效数据排序、实时处理以及特定问题求解的需求。

5、扩展视野：协程和asyncio.Queue

5.1 协程基础与async/await

协程（Coroutine）是Python中用于异步编程的一种高级控制流特性。它们允许非阻塞地执行多个任务，提高了I/O密集型程序的效率。从Python 3.5起，引入了async/await语法 ，大大简化了异步编程模型。

async关键字用于声明一个协程函数（coroutine function），而await用于等待一个异步操作的结果，直到该操作完成。这意味着在等待期间 ，控制权可以交给其他协程继续执行，从而实现了并发执行的效果。

5.2 异步编程中的队列应用

在异步编程领域，asyncio.Queue是asyncio模块提供的一个线程安全的队列 ，特别适合在协程之间传递数据。它支持异步地将数据放入队列（put()）和从中取出数据（get()） ，非常适合实现生产者-消费者模型，尤其在处理I/O密集型任务时，能显著提高程序性能。

示例：使用asyncio.Queue处理并发请求。

import asyncio

import aiohttp

async def fetch(session, url, queue):

async with session.get(url) as response:

data = await response.text()

await queue.put(data)

async def process_queue(queue):

while True:

data = await queue.get()

if data is None:

break

print(f"处理响应数据：{data[:50]}...")

queue.task_done()

async def main(urls):

queue = asyncio.Queue()

async with aiohttp.ClientSession() as session:

tasks = [fetch(session, url, queue) for url in urls]

asyncio.create_task(process_queue(queue))

await asyncio.gather(*tasks)

await queue.join()  # 等待所有任务完成

urls = ["http://example.com"] * 10  # 示例URL列表

asyncio.run(main(urls))

此代码示例展示了如何使用asyncio.Queue来收集并发HTTP请求的响应数据 ，并通过另一个协程异步处理这些数据。

5.3 提升I/O密集型任务性能

下面是一个使用asyncio.Queue和协程提升I/O密集型任务性能的示例。假设我们有多个URL需要并发下载，每个下载任务视为一个异步操作，我们可以利用asyncio和asyncio.Queue来高效地管理这些任务。

import asyncio

import aiohttp

async def download(url, queue):

async with aiohttp.ClientSession() as session:

async with await session.get(url) as response:

content = await response.text()

print(f"Downloaded {url} - {len(content)} bytes")

await queue.put(url)  # 完成后通知队列

async def main(urls):

queue = asyncio.Queue()

tasks = [download(url, queue) for url in urls]

# 启动所有任务

await asyncio.gather(*tasks)

# 等待所有任务完成

while not queue.empty():

await queue.get()  # 阻塞直到所有任务完成通知

urls = [

"https://example.com",

"https://example.org",

"https://example.net"

]

asyncio.run(main(urls))

在这个例子中，每个URL的下载任务由单独的协程处理，它们并发执行 ，充分利用了异步I/O ，提升了整体的下载效率。asyncio.Queue在这里起到了协调作用，确保所有下载任务完成后才结束程序 ，展示了在异步编程中队列的高效应用。

6、总结与展望

探索Python数据结构中的队列艺术，从基础列表到高效deque，进阶至并发安全queue模块 ，乃至高级优先队列与异步编程中的asyncio.Queue，本文全面剖析各类队列特性和应用场景。列表简单直观 ，deque性能优越，queue模块保障线程安全，PriorityQueue解决优先级问题 ，而asyncio.Queue助力I/O密集型任务飞速提升。未来Python队列发展或将融合更多智能化调度策略 ，持续学习与实践则是解锁高效并发处理的关键。