编程Python

Python queue 和 deque 的用法

字数统计: 917字   |   阅读时长: 3分
, ,

摘要: Python 双端队列操作

【对算法，数学，计算机感兴趣的同学，欢迎关注我哈，阅读更多原创文章】
我的网站：潮汐朝夕的生活实验室
我的公众号：算法题刷刷
我的知乎：潮汐朝夕
我的github：FennelDumplings
我的leetcode：FennelDumplings

本文通过用 Python 实现 极值栈与极值队列 中的 MaxQueue，来看一下 collections 模块中 deque 以及 queue 模块中 Queue 的用法。

collections.deque

deque 是双端队列，同时支持在任意一端压入和弹出元素。

deque 时一种序列容器，因此也支持 list 的一些操作，例如用 __getitem__() 检查内容，确定长度。下面是示例代码：

1
2
3
4
5
6
from collections import deque

deq = deque("abcdefg")
len(deq)        # 长度
deq[-1]         # 下标访问
deq.remove('c') # list 的方法

压入

尾部压入的操作与 list 相同：

1
2
deq.append()
deq.extend()

头部压入是 list 没有的：

1
2
deq.appendleft()
deq.extendleft()

其中 extendleft 迭代处理其输入，对各个元素完成与 appendleft 相同的处理。

弹出

  • 尾部弹出：pop()。
  • 头部弹出：popleft()

旋转

deque 可以按照任意方向旋转，从而跳过一些元素。

1
deq = deque(range(10))

通过一个例子看一下，若初始时 deq 内容为 deque([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])。

右旋转会从右端(尾部)取元素，将其移到左端(头部)：

1
2
deq.rotate(2)
# 结果为 deque([8, 9, 0, 1, 2, 3, 4, 5, 6, 7])

左旋转会从左端(头部)取元素，将其移到右端(尾部)：

1
2
deq.rotate(-2)
# 结果为 deque([2, 3, 4, 5, 6, 7, 8, 9, 0, 1])

限制大小

deque 可以指定一个最大长度。当队列达到指定长度时，新压入元素会弹出现有元素。

线程安全

双端队列是线程安全的，因此可以在不同线程中同时从两端消费队列的内容。示例代码在文章 Python标准库-数据结构 中。

queue 模块

queue 模块提供了适用于多线程编程的 FIFO 数据结构。可以用于在生产者和消费者线程间安全地传递消息或数据，会为调用者处理锁。

Queue

Queue 实现了 FIFO 容器，队尾压入操作为 put(x)、队头弹出操作为 get()。多线程中使用 Queue 的例子可以参考文章 实现简易的播客下载客户端。

LifoQueue

LifoQueue 是一个 LIFO 容器，也就是栈，压入和弹出操作依然是 put(x)、get()。

PriorityQueue

PriorityQueue 是一个优先级队列，压入和弹出操作依然是 put(x)、get()。多线程消费 PriorityQueue中的任务的示例代码参考文章 Python标准库-数据结构

MaxQueue

在队列的基本功能的基础上，可以随时查询队列中的最大值。逻辑和 C++ 代码模板参考文章 极值栈与极值队列。

这里直接用 Python 实现一遍代码模板，其中同时用到了 queue 和 deque。

代码 (模板 Queue + deque)

注意：Queue 不支持仅访问队头而不弹出，如果要的话可以用 list 来模拟队列。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from queue import Queue
from collections import deque

class MaxQueue:
    def __init__(self):
        self.q = Queue()
        self.deq_max = deque()

    def push_back(self, x: int) -> None:
        self.q.put(x)
        while self.deq_max and self.deq_max[-1] < x:
            self.deq_max.pop()
        self.deq_max.append(x)

    def pop_front(self) -> None:
        if not self.q: 
            return
        top_item = self.q.get()
        if top_item == self.deq_max[0]:
            self.deq_max.popleft()

    def getMax(self) -> int:
        if not self.q:
            return -1
        return self.deq_max[0]

测试

239. 滑动窗口最大值

1
2
3
4
5
6
7
8
9
10
11
12
13
class Solution:
    def maxSlidingWindow(self, nums: List[int], k: int) -> List[int]:
        max_q = MaxQueue()
        n = len(nums)
        result = [0 for _ in range(n - k + 1)]
        for i in range(k):
            max_q.push_back(nums[i])
        result[0] = max_q.getMax()
        for i in range(k, n):
            max_q.pop_front()
            max_q.push_back(nums[i])
            result[i - k + 1] = max_q.getMax()
        return result
Share