Python队列及拓展：从基础Queue到高级应用

1. 队列Queue基础概念队列Queue是一种先进先出FIFO, First In First Out的线性数据结构类似于现实生活中的排队。在Python中队列是线程安全的常用于多线程编程中的生产者-消费者模型。2. Python标准库中的队列模块Python的queue模块提供了多种队列实现2.1 Queue基本的先进先出队列import queue 创建队列 q queue.Queue(maxsize3) # 设置最大容量 入队 q.put(A) q.put(B) q.put(C) 出队 print(q.get()) # 输出: A print(q.get()) # 输出: B 队列大小 print(q.qsize()) # 输出: 12.2 LifoQueue后进先出队列栈import queue lifo_q queue.LifoQueue() lifo_q.put(A) lifo_q.put(B) lifo_q.put(C) print(lifo_q.get()) # 输出: C后进先出2.3 PriorityQueue优先级队列import queue pri_q queue.PriorityQueue() pri_q.put((3, Low priority)) pri_q.put((1, High priority)) pri_q.put((2, Medium priority)) print(pri_q.get()) # 输出: (1, High priority) print(pri_q.get()) # 输出: (2, Medium priority)3. collections.deque双端队列collections.deque是Python中高效的双端队列实现支持从两端快速添加和删除元素from collections import deque 创建双端队列 d deque([B, C, D]) 从左侧添加 d.appendleft(A) print(d) # 输出: deque([A, B, C, D]) 从右侧添加 d.append(E) print(d) # 输出: deque([A, B, C, D, E]) 从左侧弹出 print(d.popleft()) # 输出: A 从右侧弹出 print(d.pop()) # 输出: E 限制最大长度 limited_d deque(maxlen3) limited_d.extend([1, 2, 3]) limited_d.append(4) print(limited_d) # 输出: deque([2, 3, 4], maxlen3)4. 队列在多线程中的应用队列是线程间通信的安全方式import queue import threading import time def producer(q): for i in range(5): time.sleep(0.5) q.put(f产品{i}) print(f生产者生产: 产品{i}) def consumer(q): while True: item q.get() if item is None: # 终止信号 break print(f消费者消费: {item}) q.task_done() 创建队列 q queue.Queue() 创建线程 prod_thread threading.Thread(targetproducer, args(q,)) cons_thread threading.Thread(targetconsumer, args(q,)) 启动线程 prod_thread.start() cons_thread.start() 等待生产者完成 prod_thread.join() 发送终止信号 q.put(None) cons_thread.join()5. 高级队列应用与拓展5.1 异步队列asyncio.Queue用于异步编程的队列import asyncio async def producer(queue): for i in range(3): await asyncio.sleep(1) await queue.put(f异步任务{i}) print(f生产: 异步任务{i}) async def consumer(queue): while True: item await queue.get() if item is None: break print(f消费: {item}) queue.task_done() async def main(): queue asyncio.Queue() # 创建生产者和消费者任务 prod_task asyncio.create_task(producer(queue)) cons_task asyncio.create_task(consumer(queue)) 等待生产者完成 await prod_task 发送终止信号 await queue.put(None) await cons_task 等待队列清空 await queue.join() asyncio.run(main())5.2 消息队列中间件集成Python可以集成RabbitMQ、Redis等消息队列# RabbitMQ示例需要pika库 import pika 连接RabbitMQ connection pika.BlockingConnection( pika.ConnectionParameters(localhost) ) channel connection.channel() 声明队列 channel.queue_declare(queuehello) 发送消息 channel.basic_publish( exchange, routing_keyhello, bodyHello RabbitMQ! ) print(消息已发送) connection.close()5.3 自定义优先级队列import heapq class CustomPriorityQueue: def init(self): self._queue [] self._index 0 def push(self, item, priority): heapq.heappush(self._queue, (priority, self._index, item)) self._index 1 def pop(self): return heapq.heappop(self._queue)[-1] def is_empty(self): return len(self._queue) 0 使用示例 pq CustomPriorityQueue() pq.push(任务A, 2) pq.push(任务B, 1) pq.push(任务C, 3) print(pq.pop()) # 输出: 任务B优先级最高 print(pq.pop()) # 输出: 任务A6. 性能比较与选择建议队列类型特点适用场景queue.Queue线程安全FIFO多线程编程生产者-消费者queue.LifoQueue线程安全LIFO需要栈结构的线程安全场景queue.PriorityQueue线程安全按优先级任务调度优先级处理collections.deque高效双端操作需要频繁两端操作单线程场景asyncio.Queue异步支持异步编程协程间通信7. 最佳实践与注意事项线程安全在多线程环境中使用queue模块的队列避免死锁合理设置超时时间使用put(timeout...)和get(timeout...)资源管理及时调用task_done()和join()性能考虑单线程场景优先使用collections.deque错误处理处理queue.Empty和queue.Full异常8. 总结Python提供了丰富的队列实现从基础的queue.Queue到高效的collections.deque再到支持异步的asyncio.Queue。选择适合的队列类型可以显著提升程序性能和可维护性。在实际开发中应根据具体需求线程安全、性能要求、功能特性选择合适的队列实现。