Python队列案例如何实现队列存储

本文目录导读：

1. 基础队列实现
2. 使用list实现的简单队列
3. 优先队列实现
4. 实际应用案例
5. 完整示例主程序

我来为您展示Python中实现队列存储的多种方法,包括基础队列、优先队列和实际应用案例。

基础队列实现

使用collections.deque（推荐）

from collections import deque
import time
import random
class BasicQueue:
    """基础队列 - 使用deque实现"""
    def __init__(self):
        self.queue = deque()
    def enqueue(self, item):
        """入队"""
        self.queue.append(item)
        print(f"入队: {item}")
    def dequeue(self):
        """出队"""
        if not self.is_empty():
            item = self.queue.popleft()
            print(f"出队: {item}")
            return item
        print("队列为空，无法出队")
        return None
    def peek(self):
        """查看队首元素"""
        if not self.is_empty():
            return self.queue[0]
        return None
    def is_empty(self):
        """判断队列是否为空"""
        return len(self.queue) == 0
    def size(self):
        """获取队列大小"""
        return len(self.queue)
    def display(self):
        """显示队列内容"""
        print(f"队列内容: {list(self.queue)}")
        print(f"队列大小: {self.size()}")
# 测试基础队列
def test_basic_queue():
    print("=" * 40)
    print("基础队列测试")
    print("=" * 40)
    q = BasicQueue()
    # 入队操作
    for i in range(1, 6):
        q.enqueue(f"任务{i}")
    q.display()
    # 出队操作
    print("\n开始出队...")
    while not q.is_empty():
        q.dequeue()
        time.sleep(0.5)
    q.display()
# test_basic_queue()

使用list实现的简单队列

class ListQueue:
    """使用list实现的简单队列"""
    def __init__(self):
        self.queue = []
    def enqueue(self, item):
        self.queue.append(item)
    def dequeue(self):
        if not self.is_empty():
            return self.queue.pop(0)
        return None
    def is_empty(self):
        return len(self.queue) == 0
    def __str__(self):
        return f"Queue: {self.queue}"

优先队列实现

import heapq
class PriorityQueue:
    """优先队列 - 使用heapq实现"""
    def __init__(self):
        self.queue = []
        self.counter = 0  # 用于相同优先级时的稳定性
    def enqueue(self, item, priority):
        """入队，priority越小优先级越高"""
        heapq.heappush(self.queue, (priority, self.counter, item))
        self.counter += 1
        print(f"入队: {item} (优先级: {priority})")
    def dequeue(self):
        """出队，返回优先级最高的元素"""
        if not self.is_empty():
            priority, _, item = heapq.heappop(self.queue)
            print(f"出队: {item} (优先级: {priority})")
            return item
        return None
    def peek(self):
        """查看队首元素"""
        if not self.is_empty():
            return self.queue[0][2]
        return None
    def is_empty(self):
        return len(self.queue) == 0
    def size(self):
        return len(self.queue)
# 测试优先队列
def test_priority_queue():
    print("\n" + "=" * 40)
    print("优先队列测试")
    print("=" * 40)
    pq = PriorityQueue()
    # 不同优先级的任务
    tasks = [
        ("普通任务A", 3),
        ("紧急任务", 1),
        ("普通任务B", 3),
        ("重要任务", 2),
        ("低优先级任务", 5)
    ]
    for task, priority in tasks:
        pq.enqueue(task, priority)
    print("\n开始处理任务...")
    while not pq.is_empty():
        pq.dequeue()
        time.sleep(0.5)
# test_priority_queue()

实际应用案例

案例1：消息队列模拟

class MessageQueue:
    """消息队列 - 模拟发布/订阅模式"""
    def __init__(self, name="default"):
        self.name = name
        self.queue = deque()
        self.subscribers = []
        self.processed_count = 0
    def publish(self, message):
        """发布消息"""
        message_id = f"MSG-{time.time()}-{random.randint(1000, 9999)}"
        msg_data = {
            'id': message_id,
            'content': message,
            'timestamp': time.time()
        }
        self.queue.append(msg_data)
        print(f"[{self.name}] 发布消息: {message_id[:20]}... -> {message}")
        return message_id
    def subscribe(self, callback):
        """订阅消息"""
        self.subscribers.append(callback)
        print(f"[{self.name}] 新增订阅者: {callback.__name__}")
    def consume(self):
        """消费一条消息"""
        if not self.queue:
            return None
        message = self.queue.popleft()
        self.processed_count += 1
        # 通知所有订阅者
        for subscriber in self.subscribers:
            subscriber(message)
        return message
    def get_stats(self):
        """获取队列统计信息"""
        return {
            'name': self.name,
            'pending': len(self.queue),
            'processed': self.processed_count,
            'subscribers': len(self.subscribers)
        }
# 消息处理函数
def logger(message):
    print(f"  [日志] 记录消息: {message['content']}")
def email_sender(message):
    print(f"  [邮件] 发送邮件: {message['content']}")
def data_analyst(message):
    print(f"  [分析] 分析数据: {message['content']}")
def test_message_queue():
    print("\n" + "=" * 40)
    print("消息队列模拟")
    print("=" * 40)
    mq = MessageQueue("订单系统")
    # 注册订阅者
    mq.subscribe(logger)
    mq.subscribe(email_sender)
    mq.subscribe(data_analyst)
    # 发布消息
    messages = [
        "新订单 #12345",
        "订单 #12346 已支付",
        "订单 #12345 已发货",
        "用户反馈: 产品质量很好"
    ]
    for msg in messages:
        mq.publish(msg)
        print()
    # 消费消息
    print("开始消费消息...")
    while mq.queue:
        mq.consume()
        print("-" * 30)
        time.sleep(0.5)
    # 显示统计信息
    stats = mq.get_stats()
    print(f"\n队列统计: {stats}")
# test_message_queue()

案例2：任务调度器

class TaskScheduler:
    """任务调度器 - 带优先级的任务队列"""
    def __init__(self, max_workers=3):
        self.task_queue = PriorityQueue()
        self.max_workers = max_workers
        self.completed_tasks = []
        self.task_id = 0
    def add_task(self, name, duration, priority=5):
        """添加任务"""
        self.task_id += 1
        task = {
            'id': self.task_id,
            'name': name,
            'duration': duration,
            'priority': priority,
            'added_time': time.time()
        }
        self.task_queue.enqueue(task, priority)
        print(f"添加任务: {name} (ID: {self.task_id}, 耗时: {duration}s, 优先级: {priority})")
        return self.task_id
    def execute_next_task(self):
        """执行下一个任务"""
        if self.task_queue.is_empty():
            print("没有待执行的任务")
            return None
        task = self.task_queue.dequeue()
        print(f"\n开始执行: {task['name']} (优先级: {task['priority']})")
        # 模拟任务执行
        time.sleep(min(task['duration'], 2))  # 最多等待2秒
        task['completed_time'] = time.time()
        task['actual_duration'] = task['completed_time'] - task['added_time']
        self.completed_tasks.append(task)
        print(f"✓ 完成: {task['name']} (耗时: {task['actual_duration']:.2f}s)")
        return task
    def run_all(self):
        """执行所有任务"""
        print(f"\n开始执行所有任务 (最大并发: {self.max_workers})")
        print("=" * 40)
        while not self.task_queue.is_empty():
            self.execute_next_task()
            print("-" * 30)
        print("\n所有任务执行完毕!")
        self.show_results()
    def show_results(self):
        """显示执行结果"""
        print("\n执行结果汇总:")
        print("=" * 40)
        for task in self.completed_tasks:
            print(f"  [{task['id']}] {task['name']} - {task['actual_duration']:.2f}s (优先级: {task['priority']})")
        total_time = sum(t['actual_duration'] for t in self.completed_tasks)
        print(f"\n总任务数: {len(self.completed_tasks)}")
        print(f"总耗时: {total_time:.2f}s")
def test_task_scheduler():
    print("\n" + "=" * 40)
    print("任务调度器测试")
    print("=" * 40)
    scheduler = TaskScheduler()
    # 添加不同优先级的任务
    scheduler.add_task("备份数据库", 3, priority=1)
    scheduler.add_task("发送通知邮件", 2, priority=3)
    scheduler.add_task("清理临时文件", 1, priority=5)
    scheduler.add_task("生成报表", 4, priority=2)
    scheduler.add_task("更新缓存", 2, priority=4)
    # 执行所有任务
    scheduler.run_all()
# test_task_scheduler()

案例3：银行排队系统

class Customer:
    """客户类"""
    def __init__(self, name, service_type, is_vip=False):
        self.name = name
        self.service_type = service_type
        self.is_vip = is_vip
        self.arrival_time = time.time()
        self.wait_time = 0
    def __str__(self):
        return f"{'VIP ' if self.is_vip else ''}{self.name} ({self.service_type})"
class BankQueueSystem:
    """银行排队系统"""
    def __init__(self):
        self.normal_queue = deque()  # 普通客户
        self.vip_queue = deque()     # VIP客户
        self.served_customers = []    # 已服务客户
        self.counter_number = 0       # 当前叫号
    def add_customer(self, name, service_type, is_vip=False):
        """添加客户到队列"""
        customer = Customer(name, service_type, is_vip)
        if is_vip:
            self.vip_queue.append(customer)
            queue_name = "VIP队列"
        else:
            self.normal_queue.append(customer)
            queue_name = "普通队列"
        print(f"{customer} 进入{queue_name}")
        self.display_queue_status()
        return customer
    def serve_next(self):
        """服务下一个客户（VIP优先）"""
        # VIP客户优先
        if self.vip_queue:
            customer = self.vip_queue.popleft()
        elif self.normal_queue:
            customer = self.normal_queue.popleft()
        else:
            print("没有客户在等待")
            return None
        # 计算等待时间
        customer.wait_time = time.time() - customer.arrival_time
        self.counter_number += 1
        print(f"\n柜台 {self.counter_number}: 服务 {customer}")
        print(f"等待时间: {customer.wait_time:.1f}秒")
        # 模拟服务过程
        service_time = random.uniform(1, 3)
        time.sleep(min(service_time, 2))  # 最多等待2秒
        customer.service_time = service_time
        self.served_customers.append(customer)
        print(f"✓ {customer} 服务完成 (耗时: {service_time:.1f}秒)")
        self.display_queue_status()
        return customer
    def display_queue_status(self):
        """显示队列状态"""
        print(f"\n当前队列状态:")
        if self.vip_queue:
            print(f"  VIP队列: {len(self.vip_queue)}人")
            for c in self.vip_queue:
                print(f"    - {c}")
        else:
            print("  VIP队列: 空")
        if self.normal_queue:
            print(f"  普通队列: {len(self.normal_queue)}人")
            for c in list(self.normal_queue)[:3]:  # 只显示前3个
                print(f"    - {c}")
            if len(self.normal_queue) > 3:
                print(f"    ... 还有{len(self.normal_queue)-3}人")
        else:
            print("  普通队列: 空")
        print()
def test_bank_queue():
    print("\n" + "=" * 40)
    print("银行排队系统")
    print("=" * 40)
    bank = BankQueueSystem()
    # 添加各种客户
    customers = [
        ("张三", "存款", False),
        ("李四", "取款", False),
        ("王老板", "理财", True),  # VIP客户
        ("赵六", "转账", False),
        ("刘总", "贷款", True),   # VIP客户
        ("陈七", "开户", False),
        ("孙八", "销户", False)
    ]
    print("开始排队...")
    for name, service, is_vip in customers:
        bank.add_customer(name, service, is_vip)
        time.sleep(0.3)
    print("\n开始服务客户...")
    print("=" * 40)
    # 服务所有客户
    for i in range(len(customers)):
        bank.serve_next()
        time.sleep(0.5)
    # 显示统计信息
    print("\n" + "=" * 40)
    print("服务统计:")
    print("-" * 40)
    total_wait = sum(c.wait_time for c in bank.served_customers)
    avg_wait = total_wait / len(bank.served_customers) if bank.served_customers else 0
    print(f"总服务客户数: {len(bank.served_customers)}")
    print(f"平均等待时间: {avg_wait:.1f}秒")
    print(f"VIP客户数: {sum(1 for c in bank.served_customers if c.is_vip)}")
# test_bank_queue()

完整示例主程序

# 取消注释下面的函数来运行对应的测试
if __name__ == "__main__":
    print("Python队列实现示例")
    print("=" * 50)
    # 测试基础队列
    test_basic_queue()
    # 测试优先队列
    test_priority_queue()
    # 测试消息队列
    test_message_queue()
    # 测试任务调度器
    test_task_scheduler()
    # 测试银行排队系统
    test_bank_queue()

这些示例展示了：

1. 基础队列: 使用deque和list实现FIFO队列
2. 优先队列: 使用heapq实现基于优先级的队列
3. 实际应用: 消息队列、任务调度器、银行排队系统

您可以根据实际需求选择合适的队列实现方式来存储和管理数据。