知识前提
利用数组实现
普通队列
优化队列
循环队列
利于优化
利用链表实现
普通链表队列
不易优化
队列二要素:入队,出队
头结点:
- 删除指定位置元素
- 判断队列是否为空
尾结点:
- 指定新元素的插入位置
- 通过尾结点判断队列是否满了,如果满了,就禁止插入
- 判断队列是否为空
顺序队列
基础队列
class Queue:
def __init__(self, capacity: int) -> None:
self.n = capacity
self.items = [-1]*capacity
self.head = 0
self.tail = 0
def enqueue(self, data):
if self.n == self.tail:
return False
self.items[self.tail] = data
self.tail += 1
return True
def dequeue(self):
if self.head == self.tail:
return None
value = self.items[self.head]
self.head += 1
return value
def test_queue():
a = Queue(10)
a.enqueue("10")
a.enqueue("20")
deque_item = a.dequeue()
assert deque_item == "10"
a.enqueue("30")
assert a.items[a.head] == "20"
assert a.items[a.tail - 1] == "30"
if __name__ == "__main__":
test_queue()
加入数据迁移的队列
上一种队列会越用越小,就需要对数据进行迁移,实现队列的自动整理。
class Queue:
def __init__(self, capacity: int) -> None:
self.n = capacity
self.items = [-1]*capacity
self.head = 0
self.tail = 0
def enqueue(self, data):
if self.n == self.tail:
if self.head == 0:
return False
for i in range(self.head, self.tail):
self.items[i-self.head] = self.items[i]
self.tail -= self.head
self.head = 0
self.items[self.tail] = data
self.tail += 1
return True
def dequeue(self):
if self.head == self.tail:
return None
value = self.items[self.head]
self.head += 1
return value
def test_queue():
a = Queue(3)
a.enqueue("10")
a.enqueue("20")
a.enqueue("30")
result = a.enqueue("40")
assert not result
deque_item = a.dequeue()
assert deque_item == "10"
a.enqueue("30")
assert a.items[0] == "20"
assert a.items[2] == "30"
if __name__ == "__main__":
test_queue()
链式队列
class Queue:
def __init__(self) -> None:
self.head = None
self.tail = None
def enqueue(self, data):
if self.tail is None:
new_node = self.Node(data)
self.tail = new_node
self.head = new_node
else:
self.tail.next = self.Node(data)
self.tail = self.tail.next
def dequeue(self):
if self.head is None:
return None
value = self.head.data
self.head = self.head.next
return value
class Node:
def __init__(self, data) -> None:
self.data = data
self.next = None
def test_queue():
a = Queue()
a.enqueue("10")
a.enqueue("20")
a.enqueue("30")
deque_item = a.dequeue()
assert deque_item == "10"
assert a.head.data == "20"
assert a.head.next.data == "30"
if __name__ == "__main__":
test_queue()
循环队列
class Queue:
def __init__(self, capacity) -> None:
self.head = 0
self.tail = 0
self.n = capacity
self.items = [-1]*capacity
def enqueue(self, data):
if (self.tail + 1) % self.n == self.head:
return False
self.items[self.tail] = data
self.tail = (self.tail + 1) % self.n
return True
def dequeue(self):
if self.tail == self.head:
return None
value = self.items[self.head]
self.head = (self.head + 1) % self.n
return value
def test_queue():
a = Queue(3)
a.enqueue("10")
a.enqueue("20")
result = a.enqueue("30")
assert not result
a.dequeue()
a.enqueue("30")
assert a.items[2] == "30"
result = a.enqueue("10")
assert not result
if __name__ == "__main__":
test_queue()
阻塞队列
入队和出队可以等待
并发队列
支持多线程的阻塞队列