Skip to content

Python Deque：使用collections.deque实现快速双端队列

Soren Atelier
Name
Soren Atelier

Updated on

使用Python列表作为队列或栈看起来很自然——直到你开始从左侧pop。调用list.pop(0)是一个O(n)操作，因为每个剩余元素都必须向前移动一个位置。在处理数百万项的循环中，这个隐藏成本将快速管道变成了龟速。同样的问题影响list.insert(0, x)，它也是O(n)。对于任何触及序列两端的工作负载，普通列表都会成为性能陷阱。

Python的collections.deque（双端队列，发音为"deck"）通过提供两端O(1)的append和pop操作来解决这个问题。它被实现为固定长度块的双向链表，在每端提供常数时间性能的同时还支持索引访问。本指南涵盖了你需要了解的关于deque的一切：创建、核心操作、有界队列、线程安全和实用模式。

📚

    创建Deque

    from collections import deque
 
d = deque([1, 2, 3, 4, 5])
print(d)  # deque([1, 2, 3, 4, 5])
 
letters = deque("hello")
print(letters)  # deque(['h', 'e', 'l', 'l', 'o'])
 
d = deque()
 
d = deque(maxlen=3)
d.append(1)
d.append(2)
d.append(3)
d.append(4)  # 右端添加会移除最左端的项
print(d)  # deque([2, 3, 4], maxlen=3)

    核心操作

    Deque为序列两端提供对称的操作。

    from collections import deque
 
d = deque([2, 3, 4])
 
d.append(5)        # 添加到右端
d.appendleft(1)    # 添加到左端
print(d)  # deque([1, 2, 3, 4, 5])
 
right = d.pop()       # 从右端移除
left = d.popleft()    # 从左端移除
print(right)  # 5
print(left)   # 1
 
d = deque([3, 4, 5])
d.extend([6, 7, 8])        # 向右端添加多个元素
d.extendleft([2, 1, 0])    # 向左端添加（注意：顺序反转）
print(d)  # deque([0, 1, 2, 3, 4, 5, 6, 7, 8])

    注意extendleft逐个处理元素，将每个元素插入到位置0。结果顺序与输入可迭代对象相反。

    其他有用的方法

    from collections import deque
 
d = deque([1, 2, 3, 2, 4, 2, 5])
 
print(d.count(2))   # 3
print(d.index(3))   # 2
d.remove(2)          # 移除第一个出现
d.insert(1, 99)      # 在位置插入
d.reverse()          # 原地反转
d.clear()            # 清除所有元素

    使用rotate()旋转

    rotate()方法将deque向右旋转n步。负数n向左旋转。

    from collections import deque
 
d = deque([1, 2, 3, 4, 5])
 
d.rotate(2)    # 向右旋转2步
print(d)       # deque([4, 5, 1, 2, 3])
 
d.rotate(-3)   # 向左旋转3步
print(d)       # deque([2, 3, 4, 5, 1])

    使用maxlen的有界Deque

    有界deque是最实用的功能之一。通过设置maxlen，你创建一个固定大小的缓冲区，当容量达到时自动驱逐最旧的项。

    滑动窗口/移动平均

    from collections import deque
 
def moving_average(data, window_size):
    """Calculate moving average using a bounded deque."""
    window = deque(maxlen=window_size)
    averages = []
    for value in data:
        window.append(value)
        if len(window) == window_size:
            averages.append(sum(window) / window_size)
    return averages
 
prices = [100, 102, 104, 103, 105, 107, 106, 108, 110, 109]
print(moving_average(prices, 3))

    文件尾部（最后N行）

    from collections import deque
 
def tail(filename, n=10):
    """Return the last n lines of a file."""
    with open(filename) as f:
        return deque(f, maxlen=n)

    这非常高效。有界deque读取整个文件，但只在内存中保留最后n行。

    Deque作为队列（FIFO）

    使用append()入队，popleft()出队：

    from collections import deque
 
queue = deque()
queue.append("first")
queue.append("second")
queue.append("third")
 
print(queue.popleft())    # first
print(queue.popleft())    # second
print(queue.popleft())    # third

    这是deque真正优于列表的地方。使用列表时，list.pop(0)是O(n)。使用deque时，popleft()是O(1)。

    Deque作为栈（LIFO）

    使用append()压入，pop()从同一端弹出：

    from collections import deque
 
stack = deque()
stack.append("first")
stack.append("second")
stack.append("third")
 
print(stack.pop())    # third
print(stack.pop())    # second
print(stack.pop())    # first

    性能比较：Deque vs 列表

    操作dequelist备注
    append(x)（右）O(1)O(1)摊销两者右端追加都快
    appendleft(x) / insert(0, x)O(1)O(n)Deque显著更快
    pop()（右）O(1)O(1)摊销两者右端弹出都快
    popleft() / pop(0)O(1)O(n)Deque显著更快
    extend(iterable)O(k)O(k)摊销k = 可迭代对象长度
    rotate(k)O(k)N/A列表没有等价操作
    索引访问 d[i]O(n)O(1)随机访问列表胜出
    len()O(1)O(1)两者都内部跟踪长度
    in（成员测试）O(n)O(n)两者都需要线性扫描

    基准测试

    from collections import deque
import time
 
def benchmark_left_operations(n):
    """Compare popleft/appendleft performance."""
    data = list(range(n))
    start = time.perf_counter()
    while data:
        data.pop(0)
    list_time = time.perf_counter() - start
 
    data = deque(range(n))
    start = time.perf_counter()
    while data:
        data.popleft()
    deque_time = time.perf_counter() - start
 
    print(f"n={n:>8,}")
    print(f"  list.pop(0):     {list_time:.4f}s")
    print(f"  deque.popleft(): {deque_time:.4f}s")
    print(f"  Speedup: {list_time / deque_time:.1f}x")
 
benchmark_left_operations(100_000)

    线程安全

    Python的deque提供线程安全的append()、appendleft()、pop()和popleft()操作。这些方法在CPython中是原子的，因为它们在GIL保护的单个字节码指令内执行。

    对于需要阻塞行为的生产系统，请使用queue.Queue代替。

    实用模式

    BFS图遍历

    from collections import deque
 
def bfs(graph, start):
    """Breadth-first search using deque as queue."""
    visited = set()
    queue = deque([start])
    order = []
    while queue:
        node = queue.popleft()
        if node not in visited:
            visited.add(node)
            order.append(node)
            queue.extend(
                neighbor for neighbor in graph[node]
                if neighbor not in visited
            )
    return order
 
graph = {
    'A': ['B', 'C'],
    'B': ['A', 'D', 'E'],
    'C': ['A', 'F'],
    'D': ['B'],
    'E': ['B', 'F'],
    'F': ['C', 'E']
}
 
print(bfs(graph, 'A'))  # ['A', 'B', 'C', 'D', 'E', 'F']

    滑动窗口最大值

    from collections import deque
 
def sliding_window_max(nums, k):
    """Find maximum in each sliding window of size k."""
    result = []
    window = deque()  # 存储有用元素的索引
    for i, num in enumerate(nums):
        while window and window[0] < i - k + 1:
            window.popleft()
        while window and nums[window[-1]] < num:
            window.pop()
        window.append(i)
        if i >= k - 1:
            result.append(nums[window[0]])
    return result
 
nums = [1, 3, -1, -3, 5, 3, 6, 7]
print(sliding_window_max(nums, 3))
# [3, 3, 5, 5, 6, 7]

    撤销历史

    有界deque非常适合作为自动遗忘最旧操作的撤销历史：

    from collections import deque
 
class TextEditor:
    def __init__(self, max_undo=50):
        self.content = ""
        self._undo_stack = deque(maxlen=max_undo)
 
    def type_text(self, text):
        self._undo_stack.append(self.content)
        self.content += text
 
    def delete_last(self, n=1):
        self._undo_stack.append(self.content)
        self.content = self.content[:-n]
 
    def undo(self):
        if self._undo_stack:
            self.content = self._undo_stack.pop()
 
editor = TextEditor(max_undo=10)
editor.type_text("Hello")
editor.type_text(" World")
print(editor.content)   # Hello World
editor.undo()
print(editor.content)   # Hello

    在Jupyter中测试数据结构性能

    在对deque和列表进行基准测试或为特定工作负载调整maxlen时，交互式笔记本非常有价值。RunCell (opens in a new tab)提供了一个AI驱动的Jupyter环境，你可以在其中分析数据结构操作、可视化性能差异并获得智能优化建议。

    常见问题

    Python中的deque是什么？

    deque（双端队列）是Python collections模块中的一种数据结构，支持两端O(1)的append和pop操作。与从前端插入或删除需要O(n)时间的列表不同，deque在任一端提供常数时间性能。

    什么时候应该使用deque而不是list？

    当你需要在序列两端进行快速插入或删除时使用deque，例如实现队列（FIFO）、维护滑动窗口或进行BFS遍历。当你需要按索引快速随机访问时使用list（列表O(1) vs deque O(n)）。

    Python deque是线程安全的吗？

    是的，append()、appendleft()、pop()和popleft()方法在CPython中是线程安全的。这些操作是原子的，因为它们在GIL保护的单个字节码指令内完成。但是，复合操作不是原子的，因此对于需要阻塞语义的生产者-消费者模式，请使用queue.Queue。

    deque中的maxlen做什么？

    maxlen参数创建一个具有固定最大大小的有界deque。当deque已满并添加新项时，对端的项会自动被丢弃。这对于实现滚动窗口、最近活动日志和撤销历史很有用。

    deque与queue.Queue的比较？

    collections.deque设计用于速度。queue.Queue在内部封装了deque，但添加了锁和条件变量等线程同步原语。单线程代码使用deque。当你在多线程生产者-消费者场景中需要阻塞的get()和put()操作时，使用queue.Queue。

    总结

    Python的collections.deque是标准库中最有用却被低估的数据结构之一。它消除了列表左侧操作的O(n)惩罚，通过maxlen提供内置的有界缓冲，提供线程安全的原子操作，并作为队列、栈、滑动窗口和图遍历算法的基础。

    关键决策很简单：如果你的代码触及序列的两端，使用deque。如果你需要按索引快速随机访问，使用list。对于大多数队列和栈模式，deque应该是你的默认选择。

    📚