前言：数据结构的重要性

在前七篇文章中，我们系统学习了Python的基本语法、数据类型、控制结构。现在，我们将深入探讨两种基础但极其重要的数据结构——栈（Stack）和队列（Queue）。这两种数据结构是计算机科学中最基本的数据组织方式，几乎在所有软件系统中都有广泛应用。

栈和队列代表了两种不同的数据管理原则：后进先出（LIFO）和先进先出（FIFO）。理解它们的特性、实现方式和应用场景，对于编写高效、可维护的程序至关重要。本文将深入解析栈和队列的各个方面，通过丰富的实战案例，帮助你掌握这两种基础数据结构的原理和应用。

---

一、 栈（Stack）：后进先出的数据结构

栈是一种遵循后进先出（Last-In-First-Out, LIFO）原则的数据结构。最后添加的元素最先被移除，就像一叠盘子，你只能从最上面取放。

1.1 栈的基本概念与操作

栈支持以下基本操作：

• push：将元素添加到栈顶
• pop：移除并返回栈顶元素
• peek/top：返回栈顶元素但不移除
• is_empty：检查栈是否为空
• size：返回栈中元素的数量

实战案例9-1：栈的基本实现

class Stack:
    """使用列表实现栈数据结构"""
    
    def __init__(self):
        self.items = []
    
    def push(self, item):
        """将元素压入栈顶"""
        self.items.append(item)
    
    def pop(self):
        """移除并返回栈顶元素"""
        if not self.is_empty():
            return self.items.pop()
        raise IndexError("弹出空栈")
    
    def peek(self):
        """返回栈顶元素但不移除"""
        if not self.is_empty():
            return self.items[-1]
        raise IndexError("查看空栈")
    
    def is_empty(self):
        """检查栈是否为空"""
        return len(self.items) == 0
    
    def size(self):
        """返回栈的大小"""
        return len(self.items)
    
    def __str__(self):
        """返回栈的字符串表示"""
        return f"Stack({self.items})"

# 测试栈的基本功能
stack = Stack()
print(f"栈是否为空: {stack.is_empty()}")  # True

# 压入元素
stack.push(10)
stack.push(20)
stack.push(30)
print(f"压入元素后: {stack}")  # Stack([10, 20, 30])
print(f"栈顶元素: {stack.peek()}")  # 30
print(f"栈大小: {stack.size()}")  # 3

# 弹出元素
popped = stack.pop()
print(f"弹出的元素: {popped}")  # 30
print(f"弹出后栈: {stack}")  # Stack([10, 20])

# 继续弹出
stack.pop()
stack.pop()
print(f"栈是否为空: {stack.is_empty()}")  # True

# 尝试弹出空栈
try:
    stack.pop()
except IndexError as e:
    print(f"错误: {e}")  # 弹出空栈

1.2 栈的应用场景

栈在计算机科学中有广泛的应用，以下是一些常见场景：

实战案例9-2：栈的实际应用

# 1. 括号匹配检查
def is_balanced(expression):
    """检查括号是否匹配"""
    stack = Stack()
    matching_brackets = {')': '(', '}': '{', ']': '['}
    
    for char in expression:
        if char in '({[':
            stack.push(char)
        elif char in ')}]':
            if stack.is_empty() or stack.pop() != matching_brackets[char]:
                return False
    
    return stack.is_empty()

# 测试括号匹配
test_expressions = [
    "((()))",        # True
    "{[()]}",        # True
    "({[]})",        # True
    "((())",         # False
    "([)]",          # False
    "{{{}}}"         # True
]

print("括号匹配检查:")
for expr in test_expressions:
    result = is_balanced(expr)
    print(f"'{expr}': {result}")

# 2. 浏览器前进后退功能
class Browser:
    """模拟浏览器前进后退功能"""
    
    def __init__(self):
        self.back_stack = Stack()
        self.forward_stack = Stack()
        self.current_page = None
    
    def navigate(self, url):
        """导航到新页面"""
        if self.current_page:
            self.back_stack.push(self.current_page)
        self.current_page = url
        self.forward_stack = Stack()  # 清空前进栈
        print(f"导航到: {url}")
    
    def back(self):
        """后退到上一页面"""
        if not self.back_stack.is_empty():
            self.forward_stack.push(self.current_page)
            self.current_page = self.back_stack.pop()
            print(f"后退到: {self.current_page}")
        else:
            print("无法后退")
    
    def forward(self):
        """前进到下一页面"""
        if not self.forward_stack.is_empty():
            self.back_stack.push(self.current_page)
            self.current_page = self.forward_stack.pop()
            print(f"前进到: {self.current_page}")
        else:
            print("无法前进")
    
    def show_current(self):
        """显示当前页面"""
        print(f"当前页面: {self.current_page}")

# 测试浏览器功能
print("\n浏览器模拟:")
browser = Browser()
browser.navigate("google.com")
browser.navigate("python.org")
browser.navigate("github.com")

browser.back()      # 后退到 python.org
browser.back()      # 后退到 google.com
browser.forward()   # 前进到 python.org
browser.navigate("stackoverflow.com")  # 导航到新页面
browser.forward()   # 无法前进（前进栈被清空）

# 3. 撤销/重做功能
class TextEditor:
    """模拟文本编辑器的撤销/重做功能"""
    
    def __init__(self):
        self.text = ""
        self.undo_stack = Stack()
        self.redo_stack = Stack()
    
    def write(self, new_text):
        """写入文本"""
        if self.text:
            self.undo_stack.push(self.text)
        self.text += new_text
        self.redo_stack = Stack()  # 清空重做栈
        print(f"当前文本: {self.text}")
    
    def undo(self):
        """撤销上一次操作"""
        if not self.undo_stack.is_empty():
            self.redo_stack.push(self.text)
            self.text = self.undo_stack.pop()
            print(f"撤销后: {self.text}")
        else:
            print("无法撤销")
    
    def redo(self):
        """重做上一次操作"""
        if not self.redo_stack.is_empty():
            self.undo_stack.push(self.text)
            self.text = self.redo_stack.pop()
            print(f"重做后: {self.text}")
        else:
            print("无法重做")

# 测试文本编辑器
print("\n文本编辑器模拟:")
editor = TextEditor()
editor.write("Hello")
editor.write(" World")
editor.write("!")
editor.undo()  # 撤销最后一个"!"
editor.undo()  # 撤销" World"
editor.redo()  # 重做" World"
editor.write(" Python")  # 写入新文本
editor.redo()  # 无法重做（重做栈被清空）

1.3 使用 collections.deque 实现栈

Python的collections模块提供了deque（双端队列），它可以高效地实现栈功能。

实战案例9-3：使用deque实现栈

from collections import deque

class DequeStack:
    """使用deque实现栈"""
    
    def __init__(self):
        self.items = deque()
    
    def push(self, item):
        """将元素压入栈顶"""
        self.items.append(item)
    
    def pop(self):
        """移除并返回栈顶元素"""
        if not self.is_empty():
            return self.items.pop()
        raise IndexError("弹出空栈")
    
    def peek(self):
        """返回栈顶元素但不移除"""
        if not self.is_empty():
            return self.items[-1]
        raise IndexError("查看空栈")
    
    def is_empty(self):
        """检查栈是否为空"""
        return len(self.items) == 0
    
    def size(self):
        """返回栈的大小"""
        return len(self.items)
    
    def __str__(self):
        """返回栈的字符串表示"""
        return f"DequeStack({list(self.items)})"

# 比较列表和deque实现的性能
import time

def test_stack_performance(stack_class, n=100000):
    """测试栈实现的性能"""
    stack = stack_class()
    start = time.time()
    
    # 压入n个元素
    for i in range(n):
        stack.push(i)
    
    # 弹出所有元素
    while not stack.is_empty():
        stack.pop()
    
    end = time.time()
    return end - start

# 测试性能
n = 100000
list_time = test_stack_performance(Stack, n)
deque_time = test_stack_performance(DequeStack, n)

print(f"列表实现耗时: {list_time:.4f}秒")
print(f"deque实现耗时: {deque_time:.4f}秒")
print(f"deque比列表快 {list_time/deque_time:.2f}倍")

1.4 栈在算法中的应用

栈在许多算法中都有重要应用，特别是深度优先搜索（DFS）和表达式求值。

实战案例9-4：栈在算法中的应用

# 1. 深度优先搜索（DFS）
def dfs(graph, start):
    """使用栈实现深度优先搜索"""
    visited = set()
    stack = [start]
    result = []
    
    while stack:
        vertex = stack.pop()
        if vertex not in visited:
            visited.add(vertex)
            result.append(vertex)
            # 将邻居按逆序压入栈，以保持原始顺序
            for neighbor in reversed(graph.get(vertex, [])):
                if neighbor not in visited:
                    stack.append(neighbor)
    
    return result

# 测试DFS
graph = {
    'A': ['B', 'C'],
    'B': ['D', 'E'],
    'C': ['F'],
    'D': [],
    'E': ['F'],
    'F': []
}

print("深度优先搜索:")
print(dfs(graph, 'A'))  # ['A', 'B', 'D', 'E', 'F', 'C']

# 2. 表达式求值（逆波兰表示法）
def evaluate_rpn(expression):
    """计算逆波兰表达式"""
    stack = Stack()
    operators = {
        '+': lambda a, b: a + b,
        '-': lambda a, b: a - b,
        '*': lambda a, b: a * b,
        '/': lambda a, b: a / b
    }
    
    for token in expression.split():
        if token in operators:
            # 弹出两个操作数
            b = stack.pop()
            a = stack.pop()
            # 执行运算并将结果压入栈
            result = operators[token](a, b)
            stack.push(result)
        else:
            # 将操作数压入栈
            stack.push(float(token))
    
    return stack.pop()

# 测试逆波兰表达式求值
expressions = [
    "3 4 +",           # 3 + 4 = 7
    "5 1 2 + 4 * + 3 -",  # 5 + ((1 + 2) × 4) - 3 = 14
    "10 6 9 3 + -11 * / * 17 + 5 +"  # 复杂表达式
]

print("\n逆波兰表达式求值:")
for expr in expressions:
    try:
        result = evaluate_rpn(expr)
        print(f"'{expr}' = {result}")
    except Exception as e:
        print(f"计算 '{expr}' 时出错: {e}")

# 3. 中缀表达式转后缀表达式（逆波兰表示法）
def infix_to_postfix(expression):
    """将中缀表达式转换为后缀表达式"""
    precedence = {'+': 1, '-': 1, '*': 2, '/': 2, '^': 3}
    stack = Stack()
    output = []
    
    for token in expression.split():
        if token.isdigit() or token.replace('.', '').isdigit():
            # 数字直接添加到输出
            output.append(token)
        elif token == '(':
            # 左括号压入栈
            stack.push(token)
        elif token == ')':
            # 右括号：弹出直到遇到左括号
            while not stack.is_empty() and stack.peek() != '(':
                output.append(stack.pop())
            stack.pop()  # 弹出左括号
        else:
            # 运算符：弹出优先级更高或相等的运算符
            while (not stack.is_empty() and stack.peek() != '(' and
                   precedence.get(token, 0) <= precedence.get(stack.peek(), 0)):
                output.append(stack.pop())
            stack.push(token)
    
    # 弹出栈中所有剩余运算符
    while not stack.is_empty():
        output.append(stack.pop())
    
    return ' '.join(output)

# 测试中缀转后缀
infix_expressions = [
    "3 + 4",
    "( 1 + 2 ) * 4",
    "3 + 4 * 2 / ( 1 - 5 ) ^ 2"
]

print("\n中缀表达式转后缀表达式:")
for expr in infix_expressions:
    postfix = infix_to_postfix(expr)
    print(f"'{expr}' -> '{postfix}'")

---

二、 队列（Queue）：先进先出的数据结构

队列是一种遵循先进先出（First-In-First-Out, FIFO）原则的数据结构。最先添加的元素最先被移除，就像现实生活中的排队。

2.1 队列的基本概念与操作

队列支持以下基本操作：

• enqueue：将元素添加到队列末尾
• dequeue：移除并返回队列前端的元素
• front/peek：返回队列前端的元素但不移除
• is_empty：检查队列是否为空
• size：返回队列中元素的数量

实战案例9-5：队列的基本实现

class Queue:
    """使用列表实现队列数据结构"""
    
    def __init__(self):
        self.items = []
    
    def enqueue(self, item):
        """将元素添加到队列末尾"""
        self.items.append(item)
    
    def dequeue(self):
        """移除并返回队列前端的元素"""
        if not self.is_empty():
            return self.items.pop(0)
        raise IndexError("出队空队列")
    
    def front(self):
        """返回队列前端的元素但不移除"""
        if not self.is_empty():
            return self.items[0]
        raise IndexError("查看空队列")
    
    def is_empty(self):
        """检查队列是否为空"""
        return len(self.items) == 0
    
    def size(self):
        """返回队列的大小"""
        return len(self.items)
    
    def __str__(self):
        """返回队列的字符串表示"""
        return f"Queue({self.items})"

# 测试队列的基本功能
queue = Queue()
print(f"队列是否为空: {queue.is_empty()}")  # True

# 入队元素
queue.enqueue(10)
queue.enqueue(20)
queue.enqueue(30)
print(f"入队元素后: {queue}")  # Queue([10, 20, 30])
print(f"队首元素: {queue.front()}")  # 10
print(f"队列大小: {queue.size()}")  # 3

# 出队元素
dequeued = queue.dequeue()
print(f"出队的元素: {dequeued}")  # 10
print(f"出队后队列: {queue}")  # Queue([20, 30])

# 继续出队
queue.dequeue()
queue.dequeue()
print(f"队列是否为空: {queue.is_empty()}")  # True

# 尝试出队空队列
try:
    queue.dequeue()
except IndexError as e:
    print(f"错误: {e}")  # 出队空队列

2.2 使用 collections.deque 实现高效队列

使用列表实现队列时，出队操作（pop(0)）的时间复杂度为O(n)，而使用deque可以实现O(1)时间复杂度的出队操作。

实战案例9-6：使用deque实现高效队列

from collections import deque

class DequeQueue:
    """使用deque实现高效队列"""
    
    def __init__(self):
        self.items = deque()
    
    def enqueue(self, item):
        """将元素添加到队列末尾"""
        self.items.append(item)
    
    def dequeue(self):
        """移除并返回队列前端的元素"""
        if not self.is_empty():
            return self.items.popleft()
        raise IndexError("出队空队列")
    
    def front(self):
        """返回队列前端的元素但不移除"""
        if not self.is_empty():
            return self.items[0]
        raise IndexError("查看空队列")
    
    def is_empty(self):
        """检查队列是否为空"""
        return len(self.items) == 0
    
    def size(self):
        """返回队列的大小"""
        return len(self.items)
    
    def __str__(self):
        """返回队列的字符串表示"""
        return f"DequeQueue({list(self.items)})"

# 比较列表和deque实现的性能
def test_queue_performance(queue_class, n=100000):
    """测试队列实现的性能"""
    queue = queue_class()
    start = time.time()
    
    # 入队n个元素
    for i in range(n):
        queue.enqueue(i)
    
    # 出队所有元素
    while not queue.is_empty():
        queue.dequeue()
    
    end = time.time()
    return end - start

# 测试性能
n = 10000  # 减少测试规模，因为列表实现较慢
list_time = test_queue_performance(Queue, n)
deque_time = test_queue_performance(DequeQueue, n)

print(f"列表实现耗时: {list_time:.4f}秒")
print(f"deque实现耗时: {deque_time:.4f}秒")
print(f"deque比列表快 {list_time/deque_time:.2f}倍")

2.3 队列的应用场景

队列在计算机科学中有广泛的应用，特别是任务调度和广度优先搜索。

实战案例9-7：队列的实际应用

# 1. 广度优先搜索（BFS）
def bfs(graph, start):
    """使用队列实现广度优先搜索"""
    visited = set()
    queue = DequeQueue()
    queue.enqueue(start)
    visited.add(start)
    result = []
    
    while not queue.is_empty():
        vertex = queue.dequeue()
        result.append(vertex)
        
        for neighbor in graph.get(vertex, []):
            if neighbor not in visited:
                visited.add(neighbor)
                queue.enqueue(neighbor)
    
    return result

# 测试BFS
print("广度优先搜索:")
print(bfs(graph, 'A'))  # ['A', 'B', 'C', 'D', 'E', 'F']

# 2. 任务调度系统
class TaskScheduler:
    """简单的任务调度系统"""
    
    def __init__(self):
        self.task_queue = DequeQueue()
    
    def add_task(self, task_name, priority=0):
        """添加任务到队列"""
        self.task_queue.enqueue((priority, task_name))
        print(f"添加任务: {task_name} (优先级: {priority})")
    
    def process_tasks(self):
        """处理所有任务"""
        print("开始处理任务...")
        while not self.task_queue.is_empty():
            priority, task = self.task_queue.dequeue()
            print(f"处理任务: {task} (优先级: {priority})")
            # 模拟任务处理时间
            time.sleep(0.1)
        print("所有任务处理完成")

# 测试任务调度
print("\n任务调度系统:")
scheduler = TaskScheduler()
scheduler.add_task("备份数据库", 2)
scheduler.add_task("发送邮件", 1)
scheduler.add_task("生成报告", 3)
scheduler.add_task("清理缓存", 1)

# 在实际应用中，这里可能会使用多线程或异步处理
# scheduler.process_tasks()

# 3. 消息队列模拟
class MessageQueue:
    """简单的消息队列模拟"""
    
    def __init__(self):
        self.messages = DequeQueue()
        self.subscribers = set()
    
    def subscribe(self, subscriber_id):
        """订阅消息队列"""
        self.subscribers.add(subscriber_id)
        print(f"订阅者 {subscriber_id} 已加入")
    
    def unsubscribe(self, subscriber_id):
        """取消订阅消息队列"""
        if subscriber_id in self.subscribers:
            self.subscribers.remove(subscriber_id)
            print(f"订阅者 {subscriber_id} 已离开")
    
    def publish(self, message):
        """发布消息到队列"""
        self.messages.enqueue(message)
        print(f"发布消息: {message}")
    
    def consume(self, subscriber_id):
        """消费消息"""
        if subscriber_id not in self.subscribers:
            print(f"错误: {subscriber_id} 未订阅")
            return None
        
        if not self.messages.is_empty():
            message = self.messages.dequeue()
            print(f"订阅者 {subscriber_id} 消费消息: {message}")
            return message
        else:
            print(f"没有消息可供消费")
            return None

# 测试消息队列
print("\n消息队列模拟:")
mq = MessageQueue()
mq.subscribe("consumer1")
mq.subscribe("consumer2")

mq.publish("Hello World!")
mq.publish("How are you?")
mq.publish("Goodbye!")

mq.consume("consumer1")  # 消费第一条消息
mq.consume("consumer2")  # 消费第二条消息
mq.consume("consumer1")  # 消费第三条消息
mq.consume("consumer2")  # 没有消息了

2.4 优先队列（Priority Queue）

优先队列是一种特殊类型的队列，其中每个元素都有优先级，优先级高的元素先出队。

实战案例9-8：优先队列的实现与应用

import heapq

class PriorityQueue:
    """使用heapq实现优先队列"""
    
    def __init__(self):
        self.heap = []
        self.index = 0  # 用于处理相同优先级的情况
    
    def push(self, item, priority=0):
        """将元素添加到优先队列"""
        # 使用三元组 (priority, index, item) 确保相同优先级时按插入顺序处理
        heapq.heappush(self.heap, (priority, self.index, item))
        self.index += 1
    
    def pop(self):
        """移除并返回优先级最高的元素"""
        if not self.is_empty():
            priority, index, item = heapq.heappop(self.heap)
            return item
        raise IndexError("弹出空优先队列")
    
    def is_empty(self):
        """检查优先队列是否为空"""
        return len(self.heap) == 0
    
    def size(self):
        """返回优先队列的大小"""
        return len(self.heap)
    
    def __str__(self):
        """返回优先队列的字符串表示"""
        items = [f"({p}: {i})" for p, idx, i in sorted(self.heap)]
        return f"PriorityQueue({items})"

# 测试优先队列
print("优先队列测试:")
pq = PriorityQueue()
pq.push("任务A", 3)
pq.push("任务B", 1)
pq.push("任务C", 2)
pq.push("任务D", 1)  # 相同优先级

print(f"优先队列: {pq}")
print(f"弹出: {pq.pop()}")  # 任务B (优先级1)
print(f"弹出: {pq.pop()}")  # 任务D (优先级1，但后插入)
print(f"弹出: {pq.pop()}")  # 任务C (优先级2)
print(f"弹出: {pq.pop()}")  # 任务A (优先级3)

# 使用优先队列实现Dijkstra算法（简化版）
def dijkstra(graph, start):
    """使用优先队列实现Dijkstra算法（简化版）"""
    distances = {node: float('infinity') for node in graph}
    distances[start] = 0
    pq = PriorityQueue()
    pq.push(start, 0)
    
    while not pq.is_empty():
        current_node = pq.pop()
        
        for neighbor, weight in graph[current_node].items():
            distance = distances[current_node] + weight
            
            if distance < distances[neighbor]:
                distances[neighbor] = distance
                pq.push(neighbor, distance)
    
    return distances

# 测试Dijkstra算法
graph = {
    'A': {'B': 1, 'C': 4},
    'B': {'A': 1, 'C': 2, 'D': 5},
    'C': {'A': 4, 'B': 2, 'D': 1},
    'D': {'B': 5, 'C': 1}
}

print("\nDijkstra算法:")
shortest_paths = dijkstra(graph, 'A')
for node, distance in shortest_paths.items():
    print(f"从A到{node}的最短距离: {distance}")

---

三、 双端队列（Deque）

双端队列（Deque，Double-Ended Queue）是一种允许从两端添加和移除元素的数据结构，结合了栈和队列的特性。

实战案例9-9：双端队列的应用

from collections import deque

# 创建双端队列
d = deque([1, 2, 3, 4])
print(f"初始双端队列: {d}")

# 从右侧添加元素
d.append(5)
print(f"右侧添加后: {d}")  # deque([1, 2, 3, 4, 5])

# 从左侧添加元素
d.appendleft(0)
print(f"左侧添加后: {d}")  # deque([0, 1, 2, 3, 4, 5])

# 从右侧移除元素
right = d.pop()
print(f"右侧移除: {right}, 剩余: {d}")  # 5, deque([0, 1, 2, 3, 4])

# 从左侧移除元素
left = d.popleft()
print(f"左侧移除: {left}, 剩余: {d}")  # 0, deque([1, 2, 3, 4])

# 旋转双端队列
d.rotate(2)
print(f"向右旋转2位: {d}")  # deque([3, 4, 1, 2])

d.rotate(-1)
print(f"向左旋转1位: {d}")  # deque([4, 1, 2, 3])

# 限制双端队列大小
limited_deque = deque(maxlen=3)
for i in range(5):
    limited_deque.append(i)
    print(f"添加{i}后: {limited_deque}")

# 输出:
# 添加0后: deque([0], maxlen=3)
# 添加1后: deque([0, 1], maxlen=3)
# 添加2后: deque([0, 1, 2], maxlen=3)
# 添加3后: deque([1, 2, 3], maxlen=3)  # 自动移除最左边的0
# 添加4后: deque([2, 3, 4], maxlen=3)  # 自动移除最左边的1

# 使用双端队列实现回文检查
def is_palindrome(word):
    """检查字符串是否是回文"""
    d = deque(word.lower())
    while len(d) > 1:
        if d.popleft() != d.pop():
            return False
    return True

# 测试回文检查
test_words = ["radar", "python", "madam", "level", "algorithm"]
print("\n回文检查:")
for word in test_words:
    result = is_palindrome(word)
    print(f"'{word}': {result}")

---

四、 综合实战：打印任务调度系统

现在，让我们创建一个综合性的打印任务调度系统，应用本章所学的栈和队列知识。

项目目标：

• 模拟打印任务的生成和调度
• 支持不同优先级的打印任务
• 实现多种调度算法（FIFO、优先级）
• 统计打印任务的等待时间和处理时间
• 提供性能分析和报告生成

代码实现：

class PrintTask:
    """打印任务类"""
    
    def __init__(self, task_id, pages, priority=0):
        self.task_id = task_id
        self.pages = pages
        self.priority = priority  # 优先级：0=普通, 1=重要, 2=紧急
        self.submit_time = time.time()
        self.start_time = None
        self.end_time = None
    
    def process_time(self):
        """计算处理时间（每页需要0.5秒）"""
        return self.pages * 0.5
    
    def wait_time(self):
        """计算等待时间"""
        if self.start_time:
            return self.start_time - self.submit_time
        return time.time() - self.submit_time
    
    def total_time(self):
        """计算总时间"""
        if self.end_time:
            return self.end_time - self.submit_time
        return None
    
    def __lt__(self, other):
        """定义比较操作，用于优先队列"""
        return self.priority > other.priority  # 优先级高的先处理
    
    def __str__(self):
        return f"任务{self.task_id}({self.pages}页, 优先级{self.priority})"

class Printer:
    """打印机类"""
    
    def __init__(self, name):
        self.name = name
        self.current_task = None
        self.busy_until = 0
        self.completed_tasks = []
    
    def is_busy(self):
        """检查打印机是否忙碌"""
        return time.time() < self.busy_until
    
    def start_task(self, task):
        """开始处理任务"""
        self.current_task = task
        task.start_time = time.time()
        processing_time = task.process_time()
        self.busy_until = time.time() + processing_time
        print(f"{self.name} 开始处理 {task}，预计需要 {processing_time:.1f} 秒")
    
    def check_status(self):
        """检查打印机状态"""
        if self.is_busy():
            remaining = self.busy_until - time.time()
            print(f"{self.name} 正在处理 {self.current_task}，剩余 {remaining:.1f} 秒")
        else:
            print(f"{self.name} 空闲")
    
    def complete_task(self):
        """完成任务"""
        if self.current_task and not self.is_busy():
            self.current_task.end_time = time.time()
            self.completed_tasks.append(self.current_task)
            print(f"{self.name} 完成 {self.current_task}，总耗时 {self.current_task.total_time():.1f} 秒")
            self.current_task = None

class PrintScheduler:
    """打印任务调度系统"""
    
    def __init__(self, algorithm="fifo"):
        self.task_queue = deque()
        self.priority_queue = PriorityQueue()
        self.algorithm = algorithm  # "fifo" 或 "priority"
        self.next_task_id = 1
        self.printers = []
    
    def add_printer(self, printer):
        """添加打印机"""
        self.printers.append(printer)
    
    def submit_task(self, pages, priority=0):
        """提交打印任务"""
        task = PrintTask(self.next_task_id, pages, priority)
        self.next_task_id += 1
        
        if self.algorithm == "priority":
            self.priority_queue.push(task, -priority)  # 负号使优先级高的先出队
        else:
            self.task_queue.append(task)
        
        print(f"提交 {task}，当前队列长度: {self.queue_length()}")
    
    def queue_length(self):
        """返回队列长度"""
        if self.algorithm == "priority":
            return self.priority_queue.size()
        else:
            return len(self.task_queue)
    
    def get_next_task(self):
        """获取下一个任务"""
        if self.algorithm == "priority":
            if not self.priority_queue.is_empty():
                return self.priority_queue.pop()
        else:
            if self.task_queue:
                return self.task_queue.popleft()
        return None
    
    def assign_tasks(self):
        """分配任务给空闲的打印机"""
        for printer in self.printers:
            if not printer.is_busy():
                task = self.get_next_task()
                if task:
                    printer.start_task(task)
    
    def run_simulation(self, duration=60):
        """运行模拟"""
        print(f"\n开始打印任务调度模拟（算法: {self.algorithm}，持续时间: {duration}秒）")
        start_time = time.time()
        
        # 模拟任务提交
        task_submission_times = [5, 8, 12, 15, 18, 22, 25, 30, 35, 40]
        task_specs = [(10, 0), (5, 1), (3, 2), (8, 0), (2, 1), (15, 0), (4, 2), (6, 0), (7, 1), (1, 2)]
        
        while time.time() - start_time < duration:
            # 提交任务
            current_time = time.time() - start_time
            for i, (submit_time, (pages, priority)) in enumerate(zip(task_submission_times, task_specs)):
                if current_time >= submit_time and i < self.next_task_id - 1:
                    # 已经提交过
                    continue
                if current_time >= submit_time and i == self.next_task_id - 1:
                    self.submit_task(pages, priority)
            
            # 分配任务
            self.assign_tasks()
            
            # 检查打印机状态
            for printer in self.printers:
                printer.check_status()
                printer.complete_task()
            
            time.sleep(1)  # 每秒检查一次
        
        # 生成报告
        self.generate_report()
    
    def generate_report(self):
        """生成性能报告"""
        all_tasks = []
        for printer in self.printers:
            all_tasks.extend(printer.completed_tasks)
        
        if not all_tasks:
            print("没有完成任务")
            return
        
        total_wait_time = sum(task.wait_time() for task in all_tasks)
        total_process_time = sum(task.process_time() for task in all_tasks)
        avg_wait_time = total_wait_time / len(all_tasks)
        avg_total_time = (total_wait_time + total_process_time) / len(all_tasks)
        
        print("\n=== 打印任务调度报告 ===")
        print(f"调度算法: {self.algorithm}")
        print(f"完成任务数: {len(all_tasks)}")
        print(f"平均等待时间: {avg_wait_time:.2f} 秒")
        print(f"平均处理时间: {avg_total_time:.2f} 秒")
        print(f"总等待时间: {total_wait_time:.2f} 秒")
        print(f"总处理时间: {total_process_time:.2f} 秒")
        
        # 按优先级分组统计
        priority_groups = {0: [], 1: [], 2: []}
        for task in all_tasks:
            priority_groups[task.priority].append(task)
        
        for priority, tasks in priority_groups.items():
            if tasks:
                avg_wait = sum(t.wait_time() for t in tasks) / len(tasks)
                print(f"优先级{priority}任务: {len(tasks)}个，平均等待时间: {avg_wait:.2f}秒")

# 运行模拟
print("打印任务调度系统模拟:")
scheduler_fifo = PrintScheduler("fifo")
scheduler_priority = PrintScheduler("priority")

# 添加打印机
scheduler_fifo.add_printer(Printer("打印机1"))
scheduler_priority.add_printer(Printer("打印机1"))

# 运行FIFO算法模拟
scheduler_fifo.run_simulation(30)

# 运行优先级算法模拟
scheduler_priority.run_simulation(30)

---

总结与展望

通过本篇文章的深入学习，我们全面掌握了Python中的栈和队列：

1. 栈（Stack）：

   • 理解了后进先出（LIFO）的原则
   • 掌握了栈的基本操作：push、pop、peek
   • 学会了使用列表和deque实现栈
   • 了解了栈在算法和实际应用中的使用

2. 队列（Queue）：

   • 理解了先进先出（FIFO）的原则
   • 掌握了队列的基本操作：enqueue、dequeue、front
   • 学会了使用列表和deque实现高效队列
   • 了解了队列在任务调度和BFS算法中的应用

3. 优先队列（Priority Queue）：

   • 理解了优先级调度的概念
   • 学会了使用heapq实现优先队列
   • 了解了优先队列在算法中的应用

4. 双端队列（Deque）：

   • 掌握了双端队列的特性和操作
   • 了解了双端队列的应用场景

5. 综合应用：

   • 通过打印任务调度系统，将所学知识融会贯通
   • 学会了如何模拟真实世界的系统
   • 实现了完整的调度和性能分析功能

栈和队列是计算机科学中最基础的数据结构，几乎在所有软件系统中都有应用。掌握好这两种数据结构，对于编写高效、可维护的程序至关重要。

思考题：

1. 在什么情况下应该使用栈而不是队列？反之呢？
2. 使用列表实现队列时，为什么出队操作（pop(0)）的时间复杂度是O(n)？
3. 优先队列和普通队列的主要区别是什么？在什么场景下应该使用优先队列？
4. 如何实现一个固定大小的队列，当队列满时自动丢弃最旧的数据？
5. 在打印任务调度系统的基础上，还可以添加哪些功能来增强其实用性？