超越列表:用 Python Deque实现实时滑动窗口

2026年05月09日 由 alex 发表 4159 0

blog2-1


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([])

文章来源:https://towardsdatascience.com/beyond-lists-using-python-deque-for-real-time-sliding-windows/
欢迎关注ATYUN官方公众号
商务合作及内容投稿请联系邮箱:bd@atyun.com
评论 登录
热门职位
Maluuba
20000~40000/月
Cisco
25000~30000/月 深圳市
PilotAILabs
30000~60000/年 深圳市
写评论取消
回复取消