一文学会队列入门：Python数据结构与算法

Python 2023-09-30 09:03 182 0

队列(Queue)是一种特殊的线性数据结构，其操作遵循先进先出(FIFO)的原则，即最先添加到队列中的元素最先被移除。

队列的基本概念

队列的基本操作包括：入队(Enqueue)将元素添加到队列的尾部，和出队(Dequeue)从队列的头部移除元素。在Python中，我们可以使用列表来简单地模拟队列，但为了效率更高，我们经常使用 collections 模块中的 deque 类来实现队列。

from collections import deque

# 创建一个队列
queue = deque()

# 入队操作
queue.append(10)
queue.append(20)
queue.append(30)

# 此时队列的状态为 [10, 20, 30]

出队操作

从队列的头部移除元素。

# 出队操作
first_element = queue.popleft()  # 移除并返回头部元素，结果是 10

# 此时队列的状态为 [20, 30]

队列的辅助操作

(1) 查看队首和队尾元素

# 查看队首元素
front_element = queue[0]  # 结果是 20

# 查看队尾元素
rear_element = queue[-1]  # 结果是 30

(2) 检查队列是否为空

is_empty = not bool(queue)  # 如果队列为空，结果为 True

(3) 获取队列的大小

size = len(queue)  # 结果是 2，因为队列中有两个元素

优先队列

优先队列是一种特殊的队列，其中每个元素都有一个与之相关的优先级。Python的heapq模块提供了实现优先队列的工具。

import heapq

# 创建一个空的优先队列
priority_queue = []

# 入队操作
heapq.heappush(priority_queue, (1, "Task 1"))  # 数字1表示优先级
heapq.heappush(priority_queue, (3, "Task 3"))
heapq.heappush(priority_queue, (2, "Task 2"))

# 出队操作（按优先级）
task = heapq.heappop(priority_queue)  # 结果是 (1, "Task 1")

双端队列

deque 不仅可以作为一个队列使用，还可以支持从两端添加和删除元素，因此被称为双端队列。

dq = deque()

# 从头部和尾部添加元素
dq.appendleft(10)
dq.append(20)

# 从头部和尾部移除元素
dq.popleft()  # 结果是 10
dq.pop()      # 结果是 20

实战案例：任务调度

假设我们有一个打印机，需要处理一系列的打印任务。任务有不同的优先级，并且需要在有限的时间内完成。我们可以使用队列来模拟这个过程。

from random import randint

class PrintTask:
    def __init__(self, priority):
        self.priority = priority
        self.time_needed = randint(1, 5)  # 随机生成所需时间

    def tick(self):
        """减少任务所需的时间"""
        self.time_needed -= 1

    def is_done(self):
        """检查任务是否完成"""
        return self.time_needed <= 0

# 创建任务队列
tasks = deque()

# 生成10个随机任务
for _ in range(10):
    p = randint(1, 5)
    tasks.append(PrintTask(p))

# 处理任务
while tasks:
    current_task = tasks.popleft()
    current_task.tick()
    print(f"Processing task with priority {current_task.priority}... Time left: {current_task.time_needed}")

    if not current_task.is_done():
        tasks.append(current_task)
    else:
        print(f"Task with priority {current_task.priority} is done!")