Python双端队列deque实现指南：基本用法、线程安全与旋转操作

江屿

发布时间：2026-10-14 | 4人浏览过

原创

本文详细介绍Python中collections.deque双端队列的实现与应用。内容涵盖基本创建与填充、线程安全特性、任意方向旋转操作以及最大长度限制机制。通过具体代码示例，解析append、pop、popleft及rotate等核心方法的使用场景与注意事项，帮助开发者高效处理数据流与并发任务。

前言

双端队列（deque）允许从队列的任意一端添加或删除元素。栈和队列可视为双端队列的特例，其输入与输出操作被限制在单一端点。

基本用法

首先介绍容器类collections.deque()函数的基础使用方法。具体实现代码如下：


import collections

c = collections.deque('abcdefg')
print("输出双端队列：", c)
print("双端队列的长度：", len(c))
print("前端值：", c[0])
print("末端值：", c[-1])

执行上述代码后的运行结果如下：

填充

得益于双端队列的特性，支持从任意一端进行元素的添加或删除。以下代码分别演示了两端的添加与删除操作：


import collections

c = collections.deque()
#不使用构造函数初始化
c.extend("abcdefg")
#右端（末端）添加
c.append('h')
print(c)
#左端添加（前端）添加
c.extendleft('i')
print(c)
#末尾删除
c.pop()
print(c)
#前端删除
c.popleft()
print(c)
#随便删除
c.remove('c')
print(c)

运行上述代码后的效果如下：

与使用list数组类似，通过append()方法添加元素时，默认从右端（末端）开始。若需从前端添加，可使用extendleft()函数。删除操作方面，pop()函数从右端（末尾）移除元素，popleft()从左端移除。如需删除指定值，可直接使用remove()方法。

线程安全

双端队列具备线程安全特性，实际应用中支持在不同线程中同时从两端消费队列内容。具体代码实现如下：


import collections
import threading
import time

def getItem(lor, method):
    while True:
        try:
            next = method()
        except IndexError:
            break
        else:
            print("{0}:{1}".format(lor, next))
            time.sleep(0.1)
    print('{0}:None'.format(lor))
    return

c = collections.deque("abcdefg")
t1 = threading.Thread(target=getItem, args=('Left', c.popleft))
t2 = threading.Thread(target=getItem, args=('Right', c.popleft))
t1.start()
t2.start()
t1.join()
t2.join()

运行上述代码后的效果如下：

上述代码中，两个线程交替删除元素，直至双端队列deque为空。结果显示，没有重复元素被删除。

旋转

双端队列deque的另一重要功能是按任意方向旋转，从而跳过部分元素。

例如，将deque双端队列向右旋转（使用正旋转值）会从右端取出元素并移动至左端。同理，向左旋转（使用负值）则从左端将元素移动至右端。

通过以下代码即可清晰理解：


import collections

a = collections.deque("abcdefg")
b = collections.deque("abcdefg")
c = collections.deque("abcdefg")
print(a)
b.rotate(2)
print(b)
c.rotate(-2)
print(c)

运行上述代码后的效果如下：

结果显示，b的前两个字母被移动至前面，c的前两个字母被移动至后面。

限制双端队列大小

在实际的双端队列操作中，可以设置deque实例的最大长度，确保其不超过指定大小。这种机制在查找长度不确定的数据流中最后n个元素时非常有用。

首先查看一段示例代码：


import collections
import random

c1 = collections.deque(maxlen=5)
c2 = collections.deque(maxlen=3)

for i in range(8):
    r = random.randint(1, 100)
    print(r)
    c1.append(r)
    c2.append(r)
print(c1)
print(c2)

运行上述代码后的效果如下：