python基础-进线程下的queue、及其生产者消费者模型(2种方式)

2024-08-31 22:38

本文主要是介绍python基础-进线程下的queue、及其生产者消费者模型(2种方式),希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

      • 进程下的Queue
        • Queue
        • LifoQueue
        • PriorityQueue
      • 进程下的生成者消费者模型
        • 生产者发送信号子进程
        • 生产者发送信号主进程
        • 生产者发送多个信号主进程
      • 消费者发送信号JoinableQueue

进程下的Queue

进程彼此之间互相隔离,要实现进程间通信(IPC),multiprocessing模块支持两种形式:队列和管道,这两种方式都是使用消息传递的

创建队列的类(底层就是以管道和锁定的方式实现):
1
maxsize是队列中允许最大项数,省略则无大小限制。

2
q.put方法用以插入数据到队列中,put方法还有两个可选参数:blocked和timeout。如果blocked为True(默认值),并且timeout为正值,该方法会阻塞timeout指定的时间,直到该队列有剩余的空间。如果超时,会抛出Queue.Full异常。如果blocked为False,但该Queue已满,会立即抛出Queue.Full异常。

3
q.get方法可以从队列读取并且删除一个元素。同样,get方法有两个可选参数:blocked和timeout。如果blocked为True(默认值),并且timeout为正值,那么在等待时间内没有取到任何元素,会抛出Queue.Empty异常。如果blocked为False,有两种情况存在,如果Queue有一个值可用,则立即返回该值,否则,如果队列为空,则立即抛出Queue.Empty异常.

4
q.get_nowait():同q.get(False) q.put_nowait():同q.put(False)

5
q.empty():调用此方法时q为空则返回True,该结果不可靠,比如在返回True的过程中,如果队列中又加入了项目。

6
q.full():调用此方法时q已满则返回True,该结果不可靠,比如在返回True的过程中,如果队列中的项目被取走。

7
q.qsize():返回队列中目前项目的正确数量,结果也不可靠,理由同q.empty()和q.full()一样
 

Queue

先进先出


from multiprocessing import Process,Queue
import time
q=Queue(3)#put ,get ,put_nowait,get_nowait,full,empty
q.put(1)
q.put(3)
q.put(3)
print(q.full()) #满了print(q.get())
print(q.get())
print(q.get())
print(q.empty()) #空了

输出如下:

True
1
3
3
True
LifoQueue

后进先出

import queueq=queue.LifoQueue()
q.put('first')
q.put('second')
q.put('third')print(q.get())
print(q.get())
print(q.get())

输出如下:

third
second
first
PriorityQueue

存储数据时可设置优先级的队列、
put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高

import queueq=queue.PriorityQueue()
q.put((20,'a'))
q.put((10,'b'))
q.put((30,'c'))print(q.get())
print(q.get())
print(q.get())

输出如下:

(10, 'b')
(20, 'a')
(30, 'c')

进程下的生成者消费者模型

生产者消费者模型

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

为什么要使用生产者和消费者模式

在线程世界里,生产者就是生产数据的线程,消费者就是消费数据的线程。在多线程开发当中,如果生产者处理速度很快,而消费者处理速度很慢,那么生产者就必须等待消费者处理完,才能继续生产数据。同样的道理,如果消费者的处理能力大于生产者,那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯,而通过阻塞队列来进行通讯,所以生产者生产完数据之后不用等待消费者处理,直接扔给阻塞队列,消费者不找生产者要数据,而是直接从阻塞队列里取,阻塞队列就相当于一个缓冲区,平衡了生产者和消费者的处理能力。

基于队列实现生产者消费者模型

生产者发送信号(子进程)
from multiprocessing import Process,Queue
import time,random,osdef procducer(q):for i in range(5):res='包子%s' %itime.sleep(0.5)q.put(res)print('%s 生产了 %s' %(os.getpid(),res))q.put(None)def consumer(q):while True:res=q.get()if res is None:breakprint('%s 吃 %s' %(os.getpid(),res))time.sleep(random.randint(2,3))if __name__ == '__main__':q=Queue()p=Process(target=procducer,args=(q,))c=Process(target=consumer,args=(q,))p.start()c.start()p.join()print('主')

输出如下:

E:\python\python_sdk\python.exe "E:/python/py_pro/上课内容/7 生产者消费者模型.py"
2360 生产了 包子0
14576 吃 包子0
2360 生产了 包子1
2360 生产了 包子2
2360 生产了 包子3
14576 吃 包子1
2360 生产了 包子414576 吃 包子2
14576 吃 包子3
14576 吃 包子4Process finished with exit code 0

注意:结束信号None,不一定要由生产者发,主进程里同样可以发,但主进程需要等生产者结束后才应该发送该信号

生产者发送信号(主进程)
from multiprocessing import Process,Queue
import time,random,osdef procducer(q):for i in range(5):res='包子%s' %itime.sleep(0.5)q.put(res)print('%s 生产了 %s' %(os.getpid(),res))def consumer(q):while True:res=q.get()if res is None:breakprint('%s 吃 %s' %(os.getpid(),res))time.sleep(random.randint(2,3))if __name__ == '__main__':q=Queue()p=Process(target=procducer,args=(q,))c=Process(target=consumer,args=(q,))p.start()c.start()p.join()q.put(None)print('主')

但上述解决方式,在有多个生产者和多个消费者时,我们则需要用一个很low的方式去解决

生产者发送多个信号(主进程)
from multiprocessing import Process,Queue
import time,random,os
def consumer(q):while True:res=q.get()if res is None:break #收到结束信号则结束time.sleep(random.randint(1,3))print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))def producer(name,q):for i in range(2):time.sleep(random.randint(1,3))res='%s%s' %(name,i)q.put(res)print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))if __name__ == '__main__':q=Queue()#生产者们:即厨师们p1=Process(target=producer,args=('包子',q))p2=Process(target=producer,args=('骨头',q))p3=Process(target=producer,args=('泔水',q))#消费者们:即吃货们c1=Process(target=consumer,args=(q,))c2=Process(target=consumer,args=(q,))#开始p1.start()p2.start()p3.start()c1.start()c2.start()p1.join() #必须保证生产者全部生产完毕,才应该发送结束信号p2.join()p3.join()q.put(None) #有几个消费者就应该发送几次结束信号Noneq.put(None) #发送结束信号print('主')

其实我们的思路无非是发送结束信号而已,有另外一种队列提供了这种机制

消费者发送信号JoinableQueue

JoinableQueue([maxsize]):这就像是一个Queue对象,但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。

参数介绍:
maxsize是队列中允许最大项数,省略则无大小限制。
  #方法介绍:
JoinableQueue的实例p除了与Queue对象相同的方法之外还具有:
q.task_done():使用者使用此方法发出信号,表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量,将引发ValueError异常
q.join():生产者调用此方法进行阻塞,直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done()方法为止

from multiprocessing import Process, JoinableQueue
import time, random, osdef consumer(q):while True:res = q.get()time.sleep(random.randint(1, 3))print('\033[45m%s 吃 %s\033[0m' % (os.getpid(), res))q.task_done()  # 向q.join()发送一次信号,证明一个数据已经被取走了def producer(name, q):for i in range(5):time.sleep(random.randint(1, 3))res = '%s%s' % (name, i)q.put(res)print('\033[44m%s 生产了 %s\033[0m' % (os.getpid(), res))q.join()if __name__ == '__main__':q = JoinableQueue()# 生产者们:即厨师们p1 = Process(target=producer, args=('包子', q))p2 = Process(target=producer, args=('骨头', q))p3 = Process(target=producer, args=('泔水', q))# 消费者们:即吃货们c1 = Process(target=consumer, args=(q,))c2 = Process(target=consumer, args=(q,))c1.daemon = Truec2.daemon = True# 开始p_l = [p1, p2, p3, c1, c2]for p in p_l:p.start()p1.join()p2.join()p3.join()print('主')# 主进程等--->p1,p2,p3等---->c1,c2# p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据# 因而c1,c2也没有存在的价值了,应该随着主进程的结束而结束,所以设置成守护进程

输出如下:

E:\python\python_sdk\python.exe E:/python/py_pro/demo.py
12680 生产了 泔水0
10328 生产了 骨头0
4000 生产了 包子0
5288 吃 泔水0
12680 生产了 泔水1
9172 吃 骨头0
10328 生产了 骨头1
4000 生产了 包子1
5288 吃 包子0
12680 生产了 泔水2
10328 生产了 骨头2
9172 吃 泔水1
5288 吃 骨头1
4000 生产了 包子2
5288 吃 泔水2
9172 吃 包子1
5288 吃 骨头2
9172 吃 包子2

上述代码大概意思就是:
开3个子进程p1,p2,p3,每个子进程生产3个对应的食物,每个子进程做完任务后,进行p.join进行阻塞,直到消费者取完队列为止,在子进程运行过程中,开启另外2个子进程进行消费,每消费一个,向q.join()发送一次信号,证明一个数据已经被取走了,等队列数据被取完,生产子进程p1,p2,p3,任务就执行完毕,就执行 p1.join()、p2.join()、p3.join()之后代码,然后执行主进程代码“main”,由于生产子进程生产的数据被消费者取出完毕,此刻消费者依然没有退出,所以我们需要设置消费者进程为守护进程,在主进程执行完毕,主动释放消费子进程

这篇关于python基础-进线程下的queue、及其生产者消费者模型(2种方式)的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1125197

相关文章

大模型研发全揭秘:客服工单数据标注的完整攻略

在人工智能(AI)领域,数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者,掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中,工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注,不仅能够帮助提升客服自动化系统的智能化水平,还能优化客户服务流程,提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行

python: 多模块(.py)中全局变量的导入

文章目录 global关键字可变类型和不可变类型数据的内存地址单模块(单个py文件)的全局变量示例总结 多模块(多个py文件)的全局变量from x import x导入全局变量示例 import x导入全局变量示例 总结 global关键字 global 的作用范围是模块(.py)级别: 当你在一个模块(文件)中使用 global 声明变量时,这个变量只在该模块的全局命名空

Andrej Karpathy最新采访:认知核心模型10亿参数就够了,AI会打破教育不公的僵局

夕小瑶科技说 原创  作者 | 海野 AI圈子的红人,AI大神Andrej Karpathy,曾是OpenAI联合创始人之一,特斯拉AI总监。上一次的动态是官宣创办一家名为 Eureka Labs 的人工智能+教育公司 ,宣布将长期致力于AI原生教育。 近日,Andrej Karpathy接受了No Priors(投资博客)的采访,与硅谷知名投资人 Sara Guo 和 Elad G

【Python编程】Linux创建虚拟环境并配置与notebook相连接

1.创建 使用 venv 创建虚拟环境。例如,在当前目录下创建一个名为 myenv 的虚拟环境: python3 -m venv myenv 2.激活 激活虚拟环境使其成为当前终端会话的活动环境。运行: source myenv/bin/activate 3.与notebook连接 在虚拟环境中,使用 pip 安装 Jupyter 和 ipykernel: pip instal

内核启动时减少log的方式

内核引导选项 内核引导选项大体上可以分为两类:一类与设备无关、另一类与设备有关。与设备有关的引导选项多如牛毛,需要你自己阅读内核中的相应驱动程序源码以获取其能够接受的引导选项。比如,如果你想知道可以向 AHA1542 SCSI 驱动程序传递哪些引导选项,那么就查看 drivers/scsi/aha1542.c 文件,一般在前面 100 行注释里就可以找到所接受的引导选项说明。大多数选项是通过"_

零基础学习Redis(10) -- zset类型命令使用

zset是有序集合,内部除了存储元素外,还会存储一个score,存储在zset中的元素会按照score的大小升序排列,不同元素的score可以重复,score相同的元素会按照元素的字典序排列。 1. zset常用命令 1.1 zadd  zadd key [NX | XX] [GT | LT]   [CH] [INCR] score member [score member ...]

Retrieval-based-Voice-Conversion-WebUI模型构建指南

一、模型介绍 Retrieval-based-Voice-Conversion-WebUI(简称 RVC)模型是一个基于 VITS(Variational Inference with adversarial learning for end-to-end Text-to-Speech)的简单易用的语音转换框架。 具有以下特点 简单易用:RVC 模型通过简单易用的网页界面,使得用户无需深入了

【机器学习】高斯过程的基本概念和应用领域以及在python中的实例

引言 高斯过程(Gaussian Process,简称GP)是一种概率模型,用于描述一组随机变量的联合概率分布,其中任何一个有限维度的子集都具有高斯分布 文章目录 引言一、高斯过程1.1 基本定义1.1.1 随机过程1.1.2 高斯分布 1.2 高斯过程的特性1.2.1 联合高斯性1.2.2 均值函数1.2.3 协方差函数(或核函数) 1.3 核函数1.4 高斯过程回归(Gauss

透彻!驯服大型语言模型(LLMs)的五种方法,及具体方法选择思路

引言 随着时间的发展,大型语言模型不再停留在演示阶段而是逐步面向生产系统的应用,随着人们期望的不断增加,目标也发生了巨大的变化。在短短的几个月的时间里,人们对大模型的认识已经从对其zero-shot能力感到惊讶,转变为考虑改进模型质量、提高模型可用性。 「大语言模型(LLMs)其实就是利用高容量的模型架构(例如Transformer)对海量的、多种多样的数据分布进行建模得到,它包含了大量的先验

图神经网络模型介绍(1)

我们将图神经网络分为基于谱域的模型和基于空域的模型,并按照发展顺序详解每个类别中的重要模型。 1.1基于谱域的图神经网络         谱域上的图卷积在图学习迈向深度学习的发展历程中起到了关键的作用。本节主要介绍三个具有代表性的谱域图神经网络:谱图卷积网络、切比雪夫网络和图卷积网络。 (1)谱图卷积网络 卷积定理:函数卷积的傅里叶变换是函数傅里叶变换的乘积,即F{f*g}