
Python 双端队列(简称 Deque)是一种并不常见的集合类型。在网上搜索资料时,你会发现关于列表、字典和元组的内容比比皆是,但有关 deque 的资料却寥寥无几。
Deque(也可读作 “deck”)是 Python 中一种有趣且实用的集合类型。它与其他数据结构的不同之处在于:仅会保留你设定数量及以内的元素,绝不会超出上限。
Deque到底是什么?
它遵循先进先出(FIFO) 规则工作。一旦双端队列已满,若再追加新元素,它就会舍弃最左侧第一个元素,并把新元素添加到最右侧。
我们通过几个基础示例来理解这种数据容器。首先,从集合导入:from collections import deque
创建Deque
接下来我们创建一个简单的双端队列,并设置最大长度为 3。
# Create a new deque with 3 (or less) elements
my_deck = deque(maxlen = 3)
# Adding one element
my_deck.append(1)
my_deck.append(2)
my_deck.append(3)
# View
my_deck
# deque([1, 2, 3])
很好。当我们的Deque已满后,留意一下再往里面添加新值时会发生什么。最左侧的第一个元素会被自动丢弃,腾出空间给右侧新加入的元素。
# It drops the first element and adds the new one at the end.
my_deck.append('extra')
deque([2, 3, 'extra'])
它遵循先进先出(FIFO)规则运行,原理和双端队列一致。
向左侧添加元素
要记住,这种数据结构名叫双端队列,因此你也可以从左侧追加或批量扩展元素。这种情况下,队列会自动舍弃最右侧的元素。
# Append to left
my_deck.appendleft('left')
# [OUT]: deque(['left', 2, 3])
# Extend to left
my_deck.extendleft(['d', 'd'])
#[OUT]: deque(['d', 'd', 'left'])
元素轮转
你还可以对元素进行轮转操作,将它们向左或向右移动一位(或多位)。
# Create a new deque with 3 (or less) elements
my_deck = deque(maxlen = 3)
# Adding one element
my_deck.extend([1,2,3]) # deque([1, 2, 3])
# Rotating the elements by one position to the right
my_deck.rotate() # deque([3, 1, 2])
# Rotate to the left
my_deck.rotate(-1) # deque([1, 2, 3])
除此之外,除了轮转操作,还可以轻松将双端队列元素整体反转。
# New deck
my_deck.extend([1, 2, 3])
#[OUT]: deque([1, 2, 3])
# Reverse deck
my_deck.reverse()
# [OUT]: deque([3, 2, 1])
删除元素
在双端队列中,你可以从左侧、右侧删除元素,也可以按元素值进行删除。
# New deck
my_deck = deque(maxlen = 3)
my_deck.extend([1, 2, 3])
# [OUT]: deque([1, 2, 3])
# Remove item from the left
my_deck.popleft()
# [OUT]: deque([2, 3])
#---
# New deck
my_deck = deque(maxlen = 3)
my_deck.extend([1, 2, 3])
# [OUT]: deque([1, 2, 3])
# Remove item from the right
my_deck.pop()
# [OUT]: deque([1, 2])
#---
# New deck
my_deck = deque(maxlen = 3)
my_deck.extend([1, 'a', 3])
# [OUT]: deque([1, 'a', 3])
# Remove item by name
my_deck.remove('a')
# [OUT]: deque([1, 3])
你还可以清空双端队列,移除其中所有元素。
my_deck.clear()
#[OUT]: deque([])
应用场景
1. 最近搜索记录(内存管理)
普通列表可以无限制扩容,而 Deque 拥有 maxlen 最大长度参数。
这一特性非常适合最近浏览、最近搜索记录这类功能场景:只需保留最新的 N 条数据,无需手动删除旧数据。
# Keep only the last 3 user searches
search_history = deque(maxlen=3)
search_history.append("Python tutorials")
search_history.append("Machine Learning")
search_history.append("Data Science")
search_history.append("Deep Learning") # "Python tutorials" is automatically removed
print(list(search_history))
# Output: ['Machine Learning', 'Data Science', 'Deep Learning']
2. 实时数据流与移动平均值
在数据科学或物联网场景中,我们经常需要对数据流(例如温度传感器数据、股票价格)计算移动平均值。使用 Deque 可以高效维护数据的滑动窗口。
def moving_average(stream, window_size=5):
window = deque(maxlen=window_size)
for val in stream:
window.append(val)
if len(window) == window_size:
yield sum(window) / window_size
# Usage: Calculating average of a sensor reading stream
data_stream = [20, 21, 20, 22, 23, 25, 24]
print(list(moving_average(data_stream, window_size=3)))
3. 多线程任务队列(线程安全)
在 CPython 中,Deque 还有一个隐藏优势:append() 和 popleft() 操作是线程安全的。这让它非常适合实现简易的生产者 - 消费者模式:一个线程负责添加任务,另一个线程负责执行任务。
import threading
from collections import deque
task_queue = deque()
def producer():
for i in range(5):
task_queue.append(f"Task {i}") # Thread-safe append
def consumer():
while True:
try:
task = task_queue.popleft() # Thread-safe pop
print(f"Processing {task}")
except IndexError:
break
我们来看实际效果演示。
# Generate Tasks
producer()
task_queue
# [OUT] deque(['Task 0', 'Task 1', 'Task 2', 'Task 3', 'Task 4'])
# Consume Tasks
consumer()
# [OUT]
# Processing Task 0
# Processing Task 1
# Processing Task 2
# Processing Task 3
# Processing Task 4
# Check Queue
task_queue
# [OUT] deque([])
