Python deque (双端队列)

最后修改于 2025 年 4 月 2 日

deque（发音为“deck”）是 Python collections 模块中的一个双端队列，它支持在两端进行高效的添加和弹出操作。与列表相比，这些操作的时间复杂度为 O(1)。本指南涵盖了 deque 的创建、操作、性能特征和实际应用。Deques 是队列、栈和滑动窗口算法的理想选择。

基本 deque 操作

Deques 支持两端的多样化操作，并具有一致的性能。它们可以为空初始化，也可以从可迭代对象初始化，并带有可选的最大长度。本示例演示了核心的 deque 操作。理解这些基本知识是有效利用 deques 的关键。

from collections import deque

# Create a deque
d = deque([1, 2, 3])  # Initialize with elements
print(f"Initial deque: {d}")  # deque([1, 2, 3])

# Append elements
d.append(4)      # Add to right end
d.appendleft(0)  # Add to left end
print(f"After appends: {d}")  # deque([0, 1, 2, 3, 4])

# Pop elements
right = d.pop()       # Remove from right
left = d.popleft()    # Remove from left
print(f"After pops: {d}, popped: {left}, {right}")  # deque([1, 2, 3]), 0, 4

# Additional example: Max length
limited = deque(maxlen=3)
for i in range(5):
    limited.append(i)
    print(limited)  # Last 3 elements: deque([2, 3, 4], maxlen=3)

# Additional example: Rotation
d = deque([1, 2, 3, 4, 5])
d.rotate(2)   # Right rotation
print(d)      # deque([4, 5, 1, 2, 3])
d.rotate(-1)  # Left rotation
print(d)      # deque([5, 1, 2, 3, 4])

该示例展示了基本的 deque 创建和修改。append/appendleft 用于添加元素，而 pop/popleft 用于移除元素。maxlen 参数创建了一个有界 deque，当 deque 满时会丢弃旧元素。

旋转操作可以循环移动元素——正数向右旋转，负数向左旋转。与列表不同，deque 在两端的插入和删除操作的效率与大小无关。这使得 deques 成为 FIFO/LIFO 结构的理想选择。

性能特征

与列表的 O(n) 复杂度相比，Deques 在两端的添加/弹出操作提供了 O(1) 的性能。本示例对 deque 和 list 的性能进行了基准测试。了解这些差异有助于选择正确的数据结构。

from collections import deque
import timeit

# Test data sizes
sizes = [1000, 10000, 100000]

print(f"{'Size':<10}{'deque append':<20}{'list append':<20}")
for size in sizes:
    # Time deque appendleft
    d_time = timeit.timeit(
        'd.appendleft(0)', 
        setup=f'from collections import deque; d = deque(range({size}))',
        number=10000
    )
    
    # Time list insert(0)
    l_time = timeit.timeit(
        'l.insert(0, 0)', 
        setup=f'l = list(range({size}))',
        number=10000
    )
    
    print(f"{size:<10}{d_time:<20.5f}{l_time:<20.5f}")

# Additional example: Middle access
print("\nMiddle access performance:")
d = deque(range(100000))
l = list(range(100000))

%timeit d[50000]  # Slower than list
%timeit l[50000]  # Faster O(1) access

基准测试显示，deque 在左侧添加元素时具有一致的 O(1) 性能，而列表的 insert(0) 随着大小的增长会变慢至 O(n)。Deques 针对末端操作进行了优化，但在中间访问时性能为 O(n)。

列表在随机访问（lst[index]）方面更快，因为 deques 是作为双向链表实现的。当需要频繁地在两端添加/删除元素时，选择 deques；当需要随机访问或固定长度操作时，选择列表。

队列实现

Deques 自然地实现了 FIFO（队列）和 LIFO（栈）结构。本示例展示了线程安全的队列实现。由于其高效的末端操作，Deques 是这些模式的理想选择。

from collections import deque
import threading

# FIFO Queue (First-In-First-Out)
class Queue:
    def __init__(self):
        self._items = deque()
        self._lock = threading.Lock()
    
    def enqueue(self, item):
        with self._lock:
            self._items.append(item)
    
    def dequeue(self):
        with self._lock:
            return self._items.popleft()
    
    def size(self):
        with self._lock:
            return len(self._items)

# LIFO Stack (Last-In-First-Out)
class Stack:
    def __init__(self):
        self._items = deque()
    
    def push(self, item):
        self._items.append(item)
    
    def pop(self):
        return self._items.pop()
    
    def peek(self):
        return self._items[-1] if self._items else None

# Usage
q = Queue()
q.enqueue('a')
q.enqueue('b')
print(q.dequeue())  # 'a'

s = Stack()
s.push('x')
s.push('y')
print(s.pop())      # 'y'

# Additional example: Bounded queue
class BoundedQueue:
    def __init__(self, max_size):
        self._items = deque(maxlen=max_size)
    
    def enqueue(self, item):
        if len(self._items) == self._items.maxlen:
            raise Exception("Queue full")
        self._items.append(item)
    
    def dequeue(self):
        return self._items.popleft()

bq = BoundedQueue(2)
bq.enqueue(1)
bq.enqueue(2)
# bq.enqueue(3)  # Raises Exception

Queue 类使用 append（入队）和 popleft（出队）实现线程安全的 FIFO 操作。Stack 使用 append/pop 来实现 LIFO 行为。两者都利用了 deque 高效的末端操作。

BoundedQueue 展示了如何使用 maxlen 来创建固定大小的队列。当 deque 满时，它会自动丢弃旧元素，但本示例强制抛出异常。这些模式在生产者-消费者场景中很常见。

滑动窗口算法

Deques 在滑动窗口问题上表现出色，在这些问题中，您需要在序列上维护一个元素窗口。本示例演示了一个最大滑动窗口实现。Deques 有效地维护了窗口的不变量。

from collections import deque

def max_sliding_window(nums, k):
    """Return max of each sliding window of size k."""
    q = deque()
    result = []
    
    for i, num in enumerate(nums):
        # Remove indices outside current window
        while q and q[0] <= i - k:
            q.popleft()
        
        # Remove smaller elements from back
        while q and nums[q[-1]] <= num:
            q.pop()
        
        q.append(i)
        
        # Window is fully formed
        if i >= k - 1:
            result.append(nums[q[0]])
    
    return result

# Example usage
nums = [1, 3, -1, -3, 5, 3, 6, 7]
k = 3
print(max_sliding_window(nums, k))  # [3, 3, 5, 5, 6, 7]

# Additional example: Moving average
def moving_average(stream, window_size):
    q = deque(maxlen=window_size)
    for num in stream:
        q.append(num)
        yield sum(q) / len(q)

print(list(moving_average([10, 20, 30, 40, 50], 3)))  # [10.0, 15.0, 20.0, 30.0, 40.0]

max_sliding_window 函数在 deque q 中维护索引，使得 nums[q[0]] 始终是当前窗口中的最大值。窗口外的元素会从前面移除，而比新索引小的元素会在添加新索引之前从后面移除。

moving_average 生成器展示了一个更简单的滑动窗口应用，使用 maxlen 来维护窗口大小。与手动操作列表相比，Deques 使这些算法更加简洁高效。

多线程与 deque

在适当同步的情况下，Deques 可以在多线程应用程序中使用。本示例展示了一个线程安全的生产者-消费者模式。虽然 deque 操作是原子的，但复合操作需要适当的锁定。

from collections import deque
import threading
import time
import random

class ProducerConsumer:
    def __init__(self, max_size):
        self.buffer = deque(maxlen=max_size)
        self.lock = threading.Lock()
        self.not_empty = threading.Condition(self.lock)
        self.not_full = threading.Condition(self.lock)
    
    def produce(self, item):
        with self.not_full:
            while len(self.buffer) == self.buffer.maxlen:
                self.not_full.wait()
            self.buffer.append(item)
            self.not_empty.notify()
    
    def consume(self):
        with self.not_empty:
            while not self.buffer:
                self.not_empty.wait()
            item = self.buffer.popleft()
            self.not_full.notify()
            return item

# Test
def producer(pc, items):
    for item in items:
        time.sleep(random.random())
        pc.produce(item)
        print(f"Produced: {item}")

def consumer(pc, count):
    for _ in range(count):
        time.sleep(random.random() * 2)
        item = pc.consume()
        print(f"Consumed: {item}")

pc = ProducerConsumer(3)
producer_thread = threading.Thread(
    target=producer, args=(pc, ['A', 'B', 'C', 'D', 'E'])
)
consumer_thread = threading.Thread(
    target=consumer, args=(pc, 5)
)

producer_thread.start()
consumer_thread.start()
producer_thread.join()
consumer_thread.join()

ProducerConsumer 类使用有界 deque 和条件变量来协调生产者和消费者。生产者在缓冲区满时等待，消费者在缓冲区空时等待。通知确保了在条件变化时线程能够被唤醒。

虽然 deque 操作对于单个操作是线程安全的，但复合操作（如“检查-然后执行”）需要同步。该示例展示了用于线程协调的适当锁定，以防止多线程场景中的竞态条件。

Deque 与其他数据结构

Deques 与列表、队列和其他集合相比，具有不同的权衡。本示例将 deques 与替代方案进行了比较。选择正确的数据结构取决于您的访问模式和性能需求。

from collections import deque
from queue import Queue, LifoQueue
import timeit

# Deque vs List for queue operations
print("Queue operations (popleft vs pop(0)):")
d = deque(range(100000))
l = list(range(100000))

%timeit d.popleft()      # Much faster
%timeit l.pop(0)         # Slow for large lists

# Deque vs queue.Queue
print("\nqueue.Queue vs deque:")
q = Queue()
d = deque()

%timeit q.put(1); q.get()  # Thread-safe but slower
%timeit d.append(1); d.popleft()  # Faster but not thread-safe

# Deque vs LifoQueue for stacks
print("\nStack operations:")
s = LifoQueue()
d = deque()

%timeit s.put(1); s.get()  # Thread-safe stack
%timeit d.append(1); d.pop()  # Faster alternative

# Additional example: Deque as circular buffer
def circular_buffer():
    d = deque(maxlen=10)
    for i in range(15):
        d.append(i)
    return list(d)  # Last 10 elements

print(circular_buffer())  # [5, 6, 7, 8, 9, 10, 11, 12, 13, 14]

在队列操作（popleft 对比 pop(0)）方面，Deques 的性能优于列表，而在栈操作方面则与列表相当。queue.Queue 和 LifoQueue 是线程安全的，但在单线程使用时比 deque 慢。

使用 maxlen 时，Deques 可以用作循环缓冲区，自动丢弃旧元素。对于线程安全场景，请使用 queue 模块类；否则，在单线程端操作方面，deque 通常提供更好的性能。

最佳实践

当需要频繁地在两端添加/删除元素时，请使用 deques 而不是列表。对于线程安全，请添加适当的同步或使用 queue 模块。为有界集合设置 maxlen 以防止无界增长。对于队列/栈操作，请优先使用 deque 的内置方法，而不是手动操作列表。为滑动窗口算法和类似模式考虑使用 deque。

资料来源

从以下资源了解更多信息：Python deque 文档、Python 时间复杂度 和 Real Python deque 指南。

作者

我叫 Jan Bodnar，是一名充满热情的程序员，拥有丰富的编程经验。我自 2007 年开始撰写编程文章。迄今为止，我已撰写了 1400 多篇文章和 8 本电子书。我在编程教学方面拥有十多年的经验。

列出所有 Python 教程。