📚 数据结构概述
| 数据结构 | 特点 | 用途 |
|---|
| 列表(List) | 可变、有序 | 存储有序数据集合 |
| 元组(Tuple) | 不可变、有序 | 存储不可变数据 |
| 字典(Dict) | 键值对、可变 | 存储映射关系 |
| 集合(Set) | 无序、唯一 | 去重、集合运算 |
| 栈(Stack) | 先进后出(LIFO) | 递归、表达式求值 |
| 队列(Queue) | 先进先出(FIFO) | 缓存、消息队列 |
| 堆(Heap) | 完全二叉树 | 优先队列 |
🎯 任务管理示例
from collections import deque
import heapq
from typing import List, Tuple, Dict, Set
class Task:
def __init__(self, id: int, name: str, priority: int):
self.id = id
self.name = name
self.priority = priority
def __lt__(self, other):
return self.priority < other.priority
class TaskManager:
def __init__(self):
self.tasks = {} # 字典 存储任务
self.completed_tasks = set() # 集合 存储完成的任务
self.task_queue = deque() # 任务队列 双端队列
self.priority_queue = [] # 堆 优先队列
self.stack = [] # 栈 后进先出
def add_task(self, task: Task):
self.tasks[task.id] = task
self.task_queue.append(task)
heapq.heappush(self.priority_queue, task)
self.stack.append(task)
def assign_task(self) -> Tuple[int, str]:
task = self.task_queue.popleft()
return task.id, task.name
def complete_task(self, task_id: int):
task = self.tasks[task_id]
self.completed_tasks.add(task)
self.stack.remove(task)
def task_overview(self) -> Dict[str, List[Tuple[int, str]]]:
overview = {
"task queue": [(task.id, task.name) for task in self.task_queue],
"priority_queue": [(task.id, task.name, task.priority) for task in self.priority_queue],
"stack tasks": [(task.id, task.name) for task in self.stack],
"completed tasks": [(task.id, task.name) for task in self.completed_tasks],
}
return overview
if __name__ == "__main__":
task_manager = TaskManager()
task1 = Task(1, "第一个任务", 3)
task2 = Task(2, "第二个任务", 1)
task3 = Task(3, "第三个任务", 2)
task_manager.add_task(task1)
task_manager.add_task(task2)
task_manager.add_task(task3)
assigned_task = task_manager.assign_task()
print(f"Assigned Task: {assigned_task}")
task_manager.complete_task(1)
overview = task_manager.task_overview()
print("Task Overview:")
for key, value in overview.items():
print(f"{key}: {value}")
📈 heapq模块
import heapq
heap = []
# 添加元素
heapq.heappush(heap, 3)
heapq.heappush(heap, 1)
heapq.heappush(heap, 4)
heapq.heappush(heap, 1)
heapq.heappush(heap, 2)
print(heap) # [1, 1, 4, 3, 2]
# 移除最小元素
heapq.heappop(heap)
print(heap)
# 将列表转为堆
heap = [3, 1, 4, 1]
heapq.heapify(heap)
print(heap)
🔗 进阶任务管理（带链表功能）
from collections import deque
import heapq
from typing import List, Tuple, Dict, Set
class Task:
def __init__(self, id: int, name: str, priority: int, predecessor: List[int] = None, successor: List[int] = None):
self.id = id
self.name = name
self.priority = priority
self.predecessor = predecessor if predecessor is not None else []
self.successor = successor if successor is not None else []
def __lt__(self, other):
return self.priority < other.priority
class TaskManager:
def __init__(self):
self.tasks = {}
self.completed_tasks = set()
self.task_queue = deque()
self.priority_queue = []
self.stack = []
self.last_completed_task = None
def add_task(self, task: Task):
self.tasks[task.id] = task
self.task_queue.append(task)
heapq.heappush(self.priority_queue, task)
self.stack.append(task)
# 更新前驱任务的后继任务
for predecessor_id in task.predecessor:
predecessor = self.tasks[predecessor_id]
if task.id not in predecessor.successor:
predecessor.successor.append(task.id)
def assign_task(self) -> Tuple[int, str]:
task = self.task_queue.popleft()
return task.id, task.name
def complete_task(self, task_id: int):
task = self.tasks[task_id]
self.completed_tasks.add(task)
# 更新后继任务的前驱任务
for successor_id in task.successor:
successor = self.tasks[successor_id]
successor.predecessor.remove(task.id)
if self.last_completed_task is not None:
self.stack.remove(self.last_completed_task)
self.last_completed_task = task
def task_overview(self) -> Dict[str, List[Tuple[int, str, int, List[int], List[int]]]]:
overview = {
"task queue": [(task.id, task.name, task.priority, task.predecessor, task.successor) for task in self.task_queue],
"priority_queue": [(task.id, task.name, task.priority, task.predecessor, task.successor) for task in self.priority_queue],
"stack tasks": [(task.id, task.name, task.priority, task.predecessor, task.successor) for task in self.stack],
"completed tasks": [(task.id, task.name, task.priority, task.predecessor, task.successor) for task in self.completed_tasks],
}
return overview