ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 数据库 > 数据库开发技术 > python 队列系：Queue，LifoQueue，PriorityQueue，deque 初步了解

python 队列系：Queue，LifoQueue，PriorityQueue，deque 初步了解

原创 数据库开发技术 作者：程序员小钟同学 时间：2021-05-25 17:30:09 0 删除 编辑

python 自带的安全队列Queue

特点：

  • 同步的
  • 线程安全
  • 都实现了锁原语
  • 能够在多线程中直接使用

分类：

  • FIFO队列Queue（元素出入规则：先入先出)默认的

  • LIFO队列LifoQueue（元素出入规则：后入先出-类似栈（子弹弹夹模式）)

  • 优先级队列PriorityQueue

  • SimpleQueue 无界的 FIFO 队列构造函数。简单的队列，缺少任务跟踪等高级功能（3.7 新版功能.）

  • collections.deque 是无界队列的替代实现，具有快速原子的 append() 和 popleft() 操作，不需要锁定。（需要collections的包）

和之间自定义简单的队列对比 

import time
import threading


class ThreadSafeQueue:

    def __init__(self, max_size=0):
        # 定义存储的元素列表
        self.queue = []
        # 最大的数据
        self.max_size = max_size
        # 线程互斥锁，保证多线程对列表操作的串行化
        self.lock = threading.Lock()
        # 定义的条件变量对象
        self.condition = threading.Condition()

    # 获取当前队列元素的数量--获取的过程也是进行互斥锁的
    def size(self):
        with self.lock:
            size = len(self.queue)
        return size

    # 从队列里面获取元素
    def get(self, index):
        with self.lock:
            item = self.queue[index]
        return item

    # 往队列里面放入元素
    def put(self, item):
        if self.max_size != 0 and self.size() > self.max_size:
            return "队列已经满了！"

        # 进行串行的添加对象
        with self.lock:
            print('写入元素: %s' % item)
            self.queue.append(item)
            print("当前队列的元素:", self.queue)
        # 唤醒其他线程继续执行

        with self.condition:
            self.condition.notify()

    # 批量插入元素
    def batch_put(self, item_list):
        for item in item_list:
            self.put(item)

    # 从队列取出元素block-是否需要阻塞的等待
    def pop(self, block=True, timeout=None):
        if self.size() == 0:
            # 需要阻塞等待
            if block:
                with self.condition:
                    self.condition.wait(timeout=timeout)
            else:
                return None
        # 进行安全的获取原始--加锁
        with self.lock:
            item = None
            if len(self.queue) > 0:
                item = self.queue.pop()
        print("=====>" * 10)
        print('从队列中获取元素: %s' % item)
        print("=====>" * 10)
        return item


class Producer(threading.Thread):

    def run(self):
        ssds = "你是一个大爷"
        import random
        while True:
            item = random.choice(ssds)

            queue.put(item)
            time.sleep(1)


class Consumer(threading.Thread):

    def run(self):
          while True:
              item = queue.pop(block=False, timeout=-1)

              time.sleep(1)


if __name__ == '__main__':
    queue = ThreadSafeQueue(max_size=100)

    producer = Producer()
    producer2 = Producer()
    consumer = Consumer()

    producer.start()
    producer2.start()

    consumer.start()
    producer.join()
    producer2.join()
    consumer.join()

自带的队列几个方法介绍： 


来自官方文档：
https://docs.python.org/zh-cn/3.7/library/queue.html

  • empty() 如果队列为空，返回True,反之False

  • qsize() 返回队列里元素数量，注意，qsize() > 0 不保证后续的 get() 不被阻塞，qsize() < maxsize 也不保证 put() 不被阻塞。

  • full() 判断对队列是否已满，返回True,反之False

  • get([block[, timeout]])获取队列，timeout等待时间

  • put(item, block=True, timeout=None)（放一个元素进队列） get（从队列中取出一个元素） 先进先出原则

PS：将 item 放入队列。如果可选参数 block 是 true 并且 timeout 是 None (默认)，则在必要时阻塞至有空闲插槽可用。如果 timeout 是个正数，将最多阻塞 timeout 秒，如果在这段时间没有可用的空闲插槽，将引发 Full 异常。反之 (block 是 false)，如果空闲插槽立即可用，则把 item 放入队列，否则引发 Full 异常 ( 在这种情况下，timeout 将被忽略)。

  • get_nowait（立即取出一个元素，不阻塞）

  • put_nowait（立即放入一个元素，不阻塞）

  • task_done（在完成一项任务之后，向任务已经完成的队列发送一个信号）

  • join(阻塞调用线程，直到队列中的所有任务被处理掉)

队列几个示例的说明：

1）get阻塞 

import queue
q = queue.Queue()
q.put('a')
print(q.get())
q.put('b')
print(q.get())
print(q.get())

上面：如果里面没有元素的时候，进行get()的获取的时候，会阻塞，它需要等待队列中存在数据了才会去获取。

2）队列其他方法 

import queue
q = queue.Queue()
q.put('a')
print(q.get())
# print(q.get())
print('队列是否为空：',q.empty())
print('队列元素数量',q.qsize())
q.put('b')
q.put('c')
q.put('d')
print('队列是否为空：',q.empty())
print('队列元素数量',q.qsize())
print(q.get())
print(q.get())
print(q.get())

输出

a
队列是否为空： True
队列元素数量 0
队列是否为空： False
队列元素数量 3
b
c
d

3）Queue生产者消费者模式（先进先出模式） 

from queue import Queue
import time, threading


class Producer(threading.Thread):

    def run(self):
        ssds = "你是一个大爷"
        import random
        while True:
            item = random.choice(ssds)
            queue.put(item)
            print('写入元素: %s' % item)
            print('当前队列内容: %s' % queue.queue)
            time.sleep(1)


class Consumer(threading.Thread):

    def run(self):
        while True:
            try:
                item = queue.get(block=False, timeout=1)
                print("=====>" * 10)
                print('从队列中获取元素: %s' % item)
                print("=====>" * 10)
                time.sleep(1)
                queue.task_done()
            except:
                pass



if __name__ == '__main__':
    # queue = ThreadSafeQueue(max_size=100)
    queue = Queue(maxsize=100)

    producer = Producer()
    producer2 = Producer()
    consumer = Consumer()

    producer.start()
    producer2.start()

    consumer.start()
    producer.join()
    producer2.join()
    consumer.join()



输出：

写入元素: 爷=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>写入元素: 爷
当前队列内容: deque(['爷'])
当前队列内容: deque(['爷'])

从队列中获取元素: 爷
=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>

写入元素: 大
当前队列内容: deque(['爷', '大'])写入元素: 一
=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>

从队列中获取元素: 爷
=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>当前队列内容: deque(['大', '一'])

写入元素: 你写入元素: 一=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>
从队列中获取元素: 大

当前队列内容: deque(['一', '你', '一'])

当前队列内容: deque(['一', '你', '一'])
=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>
写入元素: 爷写入元素: 爷

当前队列内容: deque(['一', '你', '一', '爷', '爷'])
当前队列内容: deque(['一', '你', '一', '爷', '爷'])
=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>
从队列中获取元素: 一

4）LifoQueue 先进后出的模式，类似栈 

from queue import LifoQueue

lifoQueue = LifoQueue()
lifoQueue.put("第1个进来")
lifoQueue.put("第2个进来")
lifoQueue.put("第3个进来")
print('当前队列内容：', lifoQueue.queue)

lifoQueue.get()
lifoQueue.get()
print('当前队列内容：', lifoQueue.queue)

输出：

当前队列内容： ['第1个进来', '第2个进来', '第3个进来']
当前队列内容： ['第1个进来']

5）优先队列

  • 优先队列让你能够以任意顺序添加对象，并随时（可能是在两次添加对象之间）找出（并删除）最小的元素。

  • 可以使用对应的数据编号对应元素的排序权重

  • 设置的值越小，优先级越高

  • 可以用于快速访问容器中键值最小或最大的元素

  • PriorityQueue 线程安全的队列，提供锁语义来支持多个并发的生产者和消费者

  • 有一定的开销，内部主要是基于heapq

示例1）排序后值小的，具有出栈的优先权

from queue import PriorityQueue  # 优先队列

pq = PriorityQueue()
pq.put("第2-0个进来")
pq.put("第1-0个进来")
pq.put("第2-1个进来")
pq.put("第3个进来")
pq.put("第1-1个进来")
pq.put("第2-2个进来")
pq.put("第2-3个进来")

while not pq.empty():
    print(pq.get())

输出：

第1-0个进来
第1-1个进来
第2-0个进来
第2-1个进来
第2-2个进来
第2-3个进来
第3个进来

示例2）使用元祖的形式传入，并编号

from queue import PriorityQueue  # 优先队列

pq = PriorityQueue()
pq.put((2,"第2个进来"))
pq.put((0,"第0个进来"))
pq.put((0,"第0-0个进来"))
pq.put((1,"第1个进来"))
pq.put((2,"第2.2个进来"))

while not pq.empty():
    print(pq.get())

输出：

(0, '第0-0个进来')
(0, '第0个进来')
(1, '第1个进来')
(2, '第2.2个进来')
(2, '第2个进来')

异常的情况：

from queue import PriorityQueue  # 优先队列
pq = PriorityQueue()
pq.put((0,3))
pq.put((0,"第0-0个进来"))

while not pq.empty():
    print(pq.get())

输出错误：

    heappush(self.queue, item)
TypeError: '<' not supported between instances of 'str' and 'int'

因为优先队列内部内部使用的是堆排序。所以必须保证入栈的元素能够进行大小的比较（同类型）。

关于python中的heapq：

  • 堆（heap）是一种数据结构（本质是一个二叉树）
  • 堆（heap）+q（其中的q表示队列）= heapq模块

关于堆的了解： https://blog.csdn.net/weixin_45459565/article/details/103911510

6）collections.deque——快速和稳健的双向链表队列

  • 双端队列,分左右可以进行元素的插入或移除
  • 可以当做队列也可用作栈方式进行使用
  • 随机访问位于栈中间元素性能相对较低
from collections import deque
q = deque()
print('右侧添加一个元素：',q.append('你好'),'元素列表内容：',q)
print('右侧添加一个元素：',q.append('我是你大爷'),'元素列表内容：',q)
print('右侧添加一个元素：',q.append('你妹'),'元素列表内容：',q)
print('右侧添加一组元素：',q.extend(['吃饭','睡觉','打豆豆']),'元素列表内容：',q)
print("队列内容",q)
print('左弹出元素：',q.popleft(),'元素列表内容：',q)
print('右弹出元素：',q.pop(),'元素列表内容：',q)
print('左边添加一个元素：',q.appendleft("我是谁"),'元素列表内容：',q)
print('左边添加一组元素：',q.extendleft(["大爷","小爷",'大少爷']),'元素列表内容：',q)
print('顺时针-翻转队列：',q.rotate(1),'元素列表内容：',q)
print('逆时针-翻转队列：',q.rotate(-1),'元素列表内容：',q)
print('队列翻转1：',q.reverse(),'元素列表内容：',q)
print('队列翻转2：',q.reverse(),'元素列表内容：',q)
print('判断元素是否在队列中：',('大少爷' in q))
print('判断元素是否在队列中：',('我是你大爷' in q))
print('判断元素是否在队列中：',('我是你大爷-哈哈' in q))

print("当前区间列表：","['大少爷', '小爷', '大爷', '我是谁', '我是你大爷', '你妹', '吃饭', '睡觉']")
print('判断元素索引的索引-指定查找的区间(注意这个区间范围[0,5]如果再这个索引区间内找不到会找不到这个值的错误，索引是4)',(q.index("我是你大爷",0,5) ))

输出：

右侧添加一个元素： None 元素列表内容： deque(['你好'])
右侧添加一个元素： None 元素列表内容： deque(['你好', '我是你大爷'])
右侧添加一个元素： None 元素列表内容： deque(['你好', '我是你大爷', '你妹'])
右侧添加一组元素： None 元素列表内容： deque(['你好', '我是你大爷', '你妹', '吃饭', '睡觉', '打豆豆'])
队列内容 deque(['你好', '我是你大爷', '你妹', '吃饭', '睡觉', '打豆豆'])
左弹出元素： 你好 元素列表内容： deque(['我是你大爷', '你妹', '吃饭', '睡觉', '打豆豆'])
右弹出元素： 打豆豆 元素列表内容： deque(['我是你大爷', '你妹', '吃饭', '睡觉'])
左边添加一个元素： None 元素列表内容： deque(['我是谁', '我是你大爷', '你妹', '吃饭', '睡觉'])
左边添加一组元素： None 元素列表内容： deque(['大少爷', '小爷', '大爷', '我是谁', '我是你大爷', '你妹', '吃饭', '睡觉'])
顺时针-翻转队列： None 元素列表内容： deque(['睡觉', '大少爷', '小爷', '大爷', '我是谁', '我是你大爷', '你妹', '吃饭'])
逆时针-翻转队列： None 元素列表内容： deque(['大少爷', '小爷', '大爷', '我是谁', '我是你大爷', '你妹', '吃饭', '睡觉'])
队列翻转1： None 元素列表内容： deque(['睡觉', '吃饭', '你妹', '我是你大爷', '我是谁', '大爷', '小爷', '大少爷'])
队列翻转2： None 元素列表内容： deque(['大少爷', '小爷', '大爷', '我是谁', '我是你大爷', '你妹', '吃饭', '睡觉'])
判断元素是否在队列中： True
判断元素是否在队列中： True
判断元素是否在队列中： False
当前区间列表： ['大少爷', '小爷', '大爷', '我是谁', '我是你大爷', '你妹', '吃饭', '睡觉']
判断元素索引的索引-指定查找的区间(注意这个区间范围0,5，-我是你大爷索引是4，如果再这个索引区间内找不到会找不到这个值的错误)： 4

个人其他博客地址

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

小钟同学 | 文 【原创】| QQ：308711822

  • 1：本文相关描述主要是个人的认知和见解，如有不当之处，还望各位大佬指正。
  • 2：关于文章内容，有部分内容参考自互联网整理，如有链接会声明标注；如没有及时标注备注的链接的，如有侵权请联系，我会立即删除处理哟。


原文作者：程序员小钟同学；如有侵权，请联系管理员删除。

0
0
上一篇： python 多线程同步问题：障碍对象Barrier初步了解
下一篇： python 多线程之ThreadPoolExecutor线程池简单使用
请登录后发表评论 登录
全部评论
程序员小钟同学

注册时间：2021-02-20

  • 博文量
    224
  • 访问量
    111711

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593